6.4. Kafka Connect 배포

절차

1. .spec.build.output 에서 컨테이너 레지스트리를 지정하고 .spec.build.plugins 에서 추가 커넥터를 지정하여 KafkaConnect 사용자 정의 리소스를 구성합니다.

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnect
   metadata:
     name: my-connect-cluster
   spec: 

   1

   
     #...
     build:
       output: 

   2

   
         type: docker
         image: my-registry.io/my-org/my-connect-cluster:latest
         pushSecret: my-registry-credentials
       plugins: 

   3

   
         - name: debezium-postgres-connector
           artifacts:
             - type: tgz
               url: https://repo1.maven.org/maven2/io/debezium/debezium-connector-postgres/2.1.3.Final/debezium-connector-postgres-2.1.3.Final-plugin.tar.gz
               sha512sum: c4ddc97846de561755dc0b021a62aba656098829c70eb3ade3b817ce06d852ca12ae50c0281cc791a5a131cb7fc21fb15f4b8ee76c6cae5dd07f9c11cb7c6e79
         - name: camel-telegram
           artifacts:
             - type: tgz
               url: https://repo.maven.apache.org/maven2/org/apache/camel/kafkaconnector/camel-telegram-kafka-connector/0.11.5/camel-telegram-kafka-connector-0.11.5-package.tar.gz
               sha512sum: d6d9f45e0d1dbfcc9f6d1c7ca2046168c764389c78bc4b867dab32d24f710bb74ccf2a007d7d7a8af2dfca09d9a52ccbc2831fc715c195a3634cca055185bd91
     #...

   1

   Kafka Connect 클러스터의 사양 입니다.

   2

   (필수) 새 이미지가 푸시되는 컨테이너 레지스트리의 설정(필수)입니다.

   3

   (필수) 새 컨테이너 이미지에 추가할 커넥터 플러그인 및 아티팩트 목록. 각 플러그인은 하나 이상의 아티팩트 로 구성해야 합니다.

2. 리소스를 생성하거나 업데이트합니다.

   $ oc apply -f <kafka_connect_configuration_file>

3. 새 컨테이너 이미지가 빌드될 때까지 기다린 후 Kafka Connect 클러스터가 배포될 때까지 기다립니다.
4. Kafka Connect REST API 또는 KafkaConnector 사용자 정의 리소스를 사용하여 사용자가 추가한 커넥터 플러그인을 사용합니다.

절차

1. registry.redhat.io/amq-streams/kafka-35-rhel8:2.5.1 을 기본 이미지로 사용하여 새 Dockerfile 을 생성합니다.

   FROM registry.redhat.io/amq-streams/kafka-35-rhel8:2.5.1
   USER root:root
   COPY ./my-plugins/ /opt/kafka/plugins/
   USER 1001

   plugins 파일 예

   $ tree ./my-plugins/
   ./my-plugins/
   ├── debezium-connector-mongodb
   │   ├── bson-<version>.jar
   │   ├── CHANGELOG.md
   │   ├── CONTRIBUTE.md
   │   ├── COPYRIGHT.txt
   │   ├── debezium-connector-mongodb-<version>.jar
   │   ├── debezium-core-<version>.jar
   │   ├── LICENSE.txt
   │   ├── mongodb-driver-core-<version>.jar
   │   ├── README.md
   │   └── # ...
   ├── debezium-connector-mysql
   │   ├── CHANGELOG.md
   │   ├── CONTRIBUTE.md
   │   ├── COPYRIGHT.txt
   │   ├── debezium-connector-mysql-<version>.jar
   │   ├── debezium-core-<version>.jar
   │   ├── LICENSE.txt
   │   ├── mysql-binlog-connector-java-<version>.jar
   │   ├── mysql-connector-java-<version>.jar
   │   ├── README.md
   │   └── # ...
   └── debezium-connector-postgres
       ├── CHANGELOG.md
       ├── CONTRIBUTE.md
       ├── COPYRIGHT.txt
       ├── debezium-connector-postgres-<version>.jar
       ├── debezium-core-<version>.jar
       ├── LICENSE.txt
       ├── postgresql-<version>.jar
       ├── protobuf-java-<version>.jar
       ├── README.md
       └── # ...

   COPY 명령은 컨테이너 이미지에 복사할 플러그인 파일을 가리킵니다.

   이 예제에서는 모든 파일이 간결하게 나열되지는 않지만 Debezium 커넥터(MongoDB, MySQL 및 PostgreSQL)에 대한 플러그인을 추가합니다. Kafka Connect에서 실행 중인 Debezium은 다른 Kafka Connect 작업과 동일합니다.

2. 컨테이너 이미지를 빌드합니다.
3. 사용자 정의 이미지를 컨테이너 레지스트리로 내보냅니다.

4. 새 컨테이너 이미지를 가리킵니다.

   다음 방법 중 하나로 이미지를 가리킬 수 있습니다.

   • KafkaConnect 사용자 정의 리소스의 KafkaConnect.spec.image 속성을 편집합니다.

     설정된 경우 이 속성은 Cluster Operator의 STRIMZI_KAFKA_CONNECT_IMAGES 환경 변수를 덮어씁니다.

     apiVersion: kafka.strimzi.io/v1beta2
     kind: KafkaConnect
     metadata:
       name: my-connect-cluster
     spec: 

     1

     
       #...
       image: my-new-container-image 

     2

     
       config: 

     3

     
         #...

     1

     Kafka Connect 클러스터의 사양 입니다.

     2

     포드의 Docker 이미지입니다.

     3

     Kafka Connect 작업자 구성(연결이 아님).

   • install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml 파일에서 STRIMZI_KAFKA_CONNECT_IMAGES 환경 변수를 편집하여 새 컨테이너 이미지를 가리키는 다음 Cluster Operator를 다시 설치합니다.

절차

1. 다음 방법 중 하나로 Kafka Connect에 ScanSetting SourceConnector 및 ScanSettingSink Connector 플러그인을 추가합니다.

   • 플러그인이 자동으로 새 컨테이너 이미지를 빌드하도록 Kafka Connect 구성
   • 기본 Kafka Connect 이미지에서 Docker 이미지를 생성 (수동적으로 또는 연속 통합 사용)

2. Kafka Connect 구성에서 strimzi.io/use-connector-resources 주석을 true 로 설정합니다.

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnect
   metadata:
     name: my-connect-cluster
     annotations:
       strimzi.io/use-connector-resources: "true"
   spec:
       # ...

   KafkaConnector 리소스를 활성화하면 Cluster Operator에서 해당 리소스를 감시합니다.

3. examples/connect/source-connector.yaml 파일을 편집합니다.

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: my-source-connector 

   1

   
     labels:
       strimzi.io/cluster: my-connect-cluster 

   2

   
   spec:
     class: org.apache.kafka.connect.file.FileStreamSourceConnector 

   3

   
     tasksMax: 2 

   4

   
     autoRestart: 

   5

   
       enabled: true
     config: 

   6

   
       file: "/opt/kafka/LICENSE" 

   7

   
       topic: my-topic 

   8

   
       # ...

   1

   커넥터의 이름으로 사용되는 KafkaConnector 리소스의 이름입니다. OpenShift 리소스에 유효한 이름을 사용합니다.

   2

   Kafka Connect 클러스터의 이름이 에서 커넥터 인스턴스를 생성합니다. 커넥터를 연결하는 Kafka Connect 클러스터와 동일한 네임스페이스에 배포해야 합니다.

   3

   커넥터 클래스의 전체 이름 또는 별칭입니다. Kafka Connect 클러스터에서 사용하는 이미지에 있어야 합니다.

   4

   커넥터가 생성할 수 있는 최대 Kafka Connect 작업 수입니다.

   5

   실패한 커넥터 및 작업의 자동 재시작을 활성화합니다.

   6

   커넥터 구성은 키-값 쌍으로 설정됩니다.

   7

   이 예제 소스 커넥터 구성은 /opt/kafka/LICENSE 파일에서 데이터를 읽습니다.

   8

   소스 데이터를 게시하는 Kafka 주제입니다.

4. OpenShift 클러스터에서 소스 KafkaConnector 를 생성합니다.

   oc apply -f examples/connect/source-connector.yaml

5. examples/connect/sink-connector.yaml 파일을 생성합니다.

   touch examples/connect/sink-connector.yaml

6. 다음 YAML을 sink-connector.yaml 파일에 붙여넣습니다.

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: my-sink-connector
     labels:
       strimzi.io/cluster: my-connect
   spec:
     class: org.apache.kafka.connect.file.FileStreamSinkConnector 

   1

   
     tasksMax: 2
     config: 

   2

   
       file: "/tmp/my-file" 

   3

   
       topics: my-topic 

   4

   1

   커넥터 클래스의 전체 이름 또는 별칭입니다. Kafka Connect 클러스터에서 사용하는 이미지에 있어야 합니다.

   2

   커넥터 구성은 키-값 쌍으로 설정됩니다.

   3

   소스 데이터를 게시하는 임시 파일입니다.

   4

   Kafka 주제에서 소스 데이터를 읽습니다.

7. OpenShift 클러스터에 싱크 KafkaConnector 를 생성합니다.

   oc apply -f examples/connect/sink-connector.yaml

8. 커넥터 리소스가 생성되었는지 확인합니다.

   oc get kctr --selector strimzi.io/cluster=<my_connect_cluster> -o name
   
   my-source-connector
   my-sink-connector

   <my_connect_cluster>를 Kafka Connect 클러스터 이름으로 바꿉니다.

9. 컨테이너에서 kafka-console-consumer.sh 를 실행하여 소스 커넥터가 항목에 기록된 메시지를 읽습니다.

   oc exec <my_kafka_cluster>-kafka-0 -i -t -- bin/kafka-console-consumer.sh --bootstrap-server <my_kafka_cluster>-kafka-bootstrap.NAMESPACE.svc:9092 --topic my-topic --from-beginning

   <my_kafka_cluster>를 Kafka 클러스터 이름으로 바꿉니다.

절차

1. 재시작하려는 Kafka 커넥터를 제어하는 KafkaConnector 사용자 정의 리소스의 이름을 찾습니다.

   oc get KafkaConnector

2. OpenShift에서 KafkaConnector 리소스에 주석을 달아 커넥터를 다시 시작합니다.

   oc annotate KafkaConnector <kafka_connector_name> strimzi.io/restart=true

   restart 주석은 true 로 설정됩니다.

3. 다음 조정이 수행될 때까지 기다립니다(기본적으로 2분 마다).

   조정 프로세스에서 주석이 감지된 한 Kafka 커넥터가 재시작됩니다. Kafka Connect에서 재시작 요청을 수락하면 주석이 KafkaConnector 사용자 정의 리소스에서 제거됩니다.

절차

1. 재시작하려는 Kafka 커넥터 작업을 제어하는 KafkaConnector 사용자 정의 리소스의 이름을 찾습니다.

   oc get KafkaConnector

2. KafkaConnector 사용자 정의 리소스에서 재시작할 작업의 ID를 찾습니다. 작업 ID는 0부터 시작하여 음수가 아닌 정수입니다.

   oc describe KafkaConnector <kafka_connector_name>

3. OpenShift에서 KafkaConnector 리소스에 주석을 달아 ID를 사용하여 커넥터 작업을 다시 시작합니다.

   oc annotate KafkaConnector <kafka_connector_name> strimzi.io/restart-task=0

   이 예에서는 작업 0 이 다시 시작됩니다.

4. 다음 조정이 수행될 때까지 기다립니다(기본적으로 2분 마다).

   조정 프로세스에서 주석을 탐지한 경우 Kafka 커넥터 작업이 재시작됩니다. Kafka Connect에서 재시작 요청을 수락하면 주석이 KafkaConnector 사용자 정의 리소스에서 제거됩니다.

1. 구성과 함께 KafkaConnector 리소스를 배포하여 커넥터 인스턴스를 생성합니다.
2. strimzi.io/use-connector-resources 주석을 true 로 설정하여 Kafka Connect 구성에서 KafkaConnector 리소스를 활성화합니다.