2.3. Kafka Connect 클러스터 구성

절차

1. KafkaConnect 리소스의 사양 속성을 편집합니다.

   구성할 수 있는 속성은 이 예제 구성에 표시됩니다.

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnect 

   1

   
   metadata:
     name: my-connect-cluster
     annotations:
       strimzi.io/use-connector-resources: "true" 

   2

   
   spec:
     replicas: 3 

   3

   
     authentication: 

   4

   
       type: tls
       certificateAndKey:
         certificate: source.crt
         key: source.key
         secretName: my-user-source
     bootstrapServers: my-cluster-kafka-bootstrap:9092 

   5

   
     tls: 

   6

   
       trustedCertificates:
         - secretName: my-cluster-cluster-cert
           certificate: ca.crt
         - secretName: my-cluster-cluster-cert
           certificate: ca2.crt
     config: 

   7

   
       group.id: my-connect-cluster
       offset.storage.topic: my-connect-cluster-offsets
       config.storage.topic: my-connect-cluster-configs
       status.storage.topic: my-connect-cluster-status
       key.converter: org.apache.kafka.connect.json.JsonConverter
       value.converter: org.apache.kafka.connect.json.JsonConverter
       key.converter.schemas.enable: true
       value.converter.schemas.enable: true
       config.storage.replication.factor: 3
       offset.storage.replication.factor: 3
       status.storage.replication.factor: 3
     build: 

   8

   
       output: 

   9

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

   10

   
         - name: debezium-postgres-connector
           artifacts:
             - type: tgz
               url: https://repo1.maven.org/maven2/io/debezium/debezium-connector-postgres/1.3.1.Final/debezium-connector-postgres-1.3.1.Final-plugin.tar.gz
               sha512sum: 962a12151bdf9a5a30627eebac739955a4fd95a08d373b86bdcea2b4d0c27dd6e1edd5cb548045e115e33a9e69b1b2a352bee24df035a0447cb820077af00c03
         - name: camel-telegram
           artifacts:
             - type: tgz
               url: https://repo.maven.apache.org/maven2/org/apache/camel/kafkaconnector/camel-telegram-kafka-connector/0.7.0/camel-telegram-kafka-connector-0.7.0-package.tar.gz
               sha512sum: a9b1ac63e3284bea7836d7d24d84208c49cdf5600070e6bd1535de654f6920b74ad950d51733e8020bf4187870699819f54ef5859c7846ee4081507f48873479
     externalConfiguration: 

   11

   
       env:
         - name: AWS_ACCESS_KEY_ID
           valueFrom:
             secretKeyRef:
               name: aws-creds
               key: awsAccessKey
         - name: AWS_SECRET_ACCESS_KEY
           valueFrom:
             secretKeyRef:
               name: aws-creds
               key: awsSecretAccessKey
     resources: 

   12

   
       requests:
         cpu: "1"
         memory: 2Gi
       limits:
         cpu: "2"
         memory: 2Gi
     logging: 

   13

   
       type: inline
       loggers:
         log4j.rootLogger: "INFO"
     readinessProbe: 

   14

   
       initialDelaySeconds: 15
       timeoutSeconds: 5
     livenessProbe:
       initialDelaySeconds: 15
       timeoutSeconds: 5
     metricsConfig: 

   15

   
       type: jmxPrometheusExporter
       valueFrom:
         configMapKeyRef:
           name: my-config-map
           key: my-key
     jvmOptions: 

   16

   
       "-Xmx": "1g"
       "-Xms": "1g"
     image: my-org/my-image:latest 

   17

   
     rack:
       topologyKey: topology.kubernetes.io/zone 

   18

   
     template: 

   19

   
       pod:
         affinity:
           podAntiAffinity:
             requiredDuringSchedulingIgnoredDuringExecution:
               - labelSelector:
                   matchExpressions:
                     - key: application
                       operator: In
                       values:
                         - postgresql
                         - mongodb
                 topologyKey: "kubernetes.io/hostname"
       connectContainer: 

   20

   
         env:
           - name: JAEGER_SERVICE_NAME
             value: my-jaeger-service
           - name: JAEGER_AGENT_HOST
             value: jaeger-agent-name
           - name: JAEGER_AGENT_PORT
             value: "6831"

   Copy to Clipboard Toggle word wrap

   1

   KafkaConnect 를 사용합니다.

   2

   Kafka Connect 클러스터의 KafkaConnectors를 활성화합니다.

   3

   작업을 실행하는 작업자 의 복제본 노드 수입니다.

   4

   여기에 표시된 대로, OAuth 전달 토큰 을 사용하거나 SASL 기반 SCRAM-SHA-256/SCRAM-SHA-512 또는 PLAIN 메커니즘을 사용하여 Kafka Connect 클러스터에 대한 인증입니다. ??? 기본적으로 Kafka Connect는 일반 텍스트 연결을 사용하여 Kafka 브로커에 연결합니다.

   5

   Kafka Connect 클러스터 연결을 위한 부트스트랩 서버 입니다.

   6

   TLS 인증서가 클러스터의 X.509 형식으로 저장되는 키 이름으로 TLS 암호화 인증서가 동일한 시크릿에 저장된 경우 여러 번 나열할 수 있습니다.

   7

   작업자의 Kafka Connect 구성 (연결이 아님). 표준 Apache Kafka 구성은 AMQ Streams에서 직접 관리하지 않는 속성으로 제한될 수 있습니다.

   8

   커넥터 플러그인을 사용하여 컨테이너 이미지를 빌드하기 위한 구성 속성을 자동으로 빌드 합니다.

   9

   (필수) 새 이미지를 내보내는 컨테이너 레지스트리의 구성입니다.

   10

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

   11

   여기 또는 볼륨과 같이 환경 변수를 사용하여 Kafka 커넥터에 대한 외부 구성입니다. 구성 공급자 플러그인 을 사용하여 외부 소스에서 구성 값을 로드할 수도 있습니다.

   12

   지원되는 리소스 예약 요청, 현재 cpu 및 memory, 사용할 수 있는 최대 리소스를 지정합니다.

   13

   지정된 Kafka Connect 로거 및 로그 수준이 직접(인라인) 또는 ConfigMap을 통해 간접적으로(외부)됩니다. 사용자 정의 ConfigMap은 log4j.properties 또는 log4j2.properties 키 아래에 배치해야 합니다. Kafka Connect log4j.rootLogger 로거의 경우 로그 수준을 INFO, ERROR, WARN, TRACE, DEBUG, FATAL 또는 OFF로 설정할 수 있습니다.

   14

   컨테이너를 다시 시작할 시기(라이브)와 컨테이너가 트래픽을 허용할 시기(준비)를 확인할 상태 점검 입니다.

   15

   Prometheus 지표: 이 예제에서 Prometheus Cryostat 내보내기에 대한 구성이 포함된 ConfigMap을 참조하여 활성화됩니다. metricsConfig.valueFrom.configMapKeyRef.key 아래에 빈 파일이 포함된 ConfigMap에 대한 참조를 사용하여 추가 구성 없이 메트릭을 활성화할 수 있습니다.

   16

   Kafka Connect 를 실행하는 VM(가상 머신)의 성능을 최적화하는 JVM 구성 옵션입니다.

   17

   ADVANCED OPTION : 특수한 상황에서만 권장되는 컨테이너 이미지 구성입니다.

   18

   SPECIALIZED OPTION: 배포에 대한 Rack 인식 구성입니다. 이는 지역이 아닌 동일한 위치 내의 배포를 위한 특수 옵션입니다. 리더 복제본 대신 커넥터가 가장 가까운 복제본에서 사용할 수 있도록 하려면 이 옵션을 사용합니다. 경우에 따라 가장 가까운 복제본에서 소비하면 네트워크 사용률을 개선하거나 비용을 절감할 수 있습니다. topologyKey 는 랙 ID가 포함된 노드 레이블과 일치해야 합니다. 이 구성에 사용된 예제에서는 표준 topology.kubernetes.io/zone 레이블을 사용하는 영역을 지정합니다. 가장 가까운 복제본에서 사용하려면 Kafka 브로커 구성에서 RackAwareReplicaSelector 를 활성화합니다.

   19

   템플릿 사용자 지정. 여기에서 Pod는 유사성 방지를 사용하여 예약되므로 이름이 동일한 노드에 Pod가 예약되지 않습니다.

   20

   환경 변수는 Jaeger를 사용하여 분산 추적에도 설정됩니다.

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

   oc apply -f KAFKA-CONNECT-CONFIG-FILE

   Copy to Clipboard Toggle word wrap
3. Kafka Connect에 대한 권한 부여가 활성화된 경우 Kafka Connect 사용자를 구성하여 Kafka Connect 소비자 그룹 및 항목에 대한 액세스를 활성화합니다.

절차

1. KafkaUser 리소스에서 권한 부여 속성을 편집하여 사용자에게 액세스 권한을 제공합니다.

   다음 예에서 액세스 권한은 리터럴 이름 값을 사용하여 Kafka Connect 주제 및 소비자 그룹에 대해 구성됩니다.

   Expand

   속성	이름
   offset.storage.topic	connect-cluster-offsets
   status.storage.topic	connect-cluster-status
   config.storage.topic	connect-cluster-configs
   group	connect-cluster

   Show more

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaUser
   metadata:
     name: my-user
     labels:
       strimzi.io/cluster: my-cluster
   spec:
     # ...
     authorization:
       type: simple
       acls:
         # access to offset.storage.topic
         - resource:
             type: topic
             name: connect-cluster-offsets
             patternType: literal
           operation: Write
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-offsets
             patternType: literal
           operation: Create
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-offsets
             patternType: literal
           operation: Describe
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-offsets
             patternType: literal
           operation: Read
           host: "*"
         # access to status.storage.topic
         - resource:
             type: topic
             name: connect-cluster-status
             patternType: literal
           operation: Write
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-status
             patternType: literal
           operation: Create
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-status
             patternType: literal
           operation: Describe
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-status
             patternType: literal
           operation: Read
           host: "*"
         # access to config.storage.topic
         - resource:
             type: topic
             name: connect-cluster-configs
             patternType: literal
           operation: Write
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-configs
             patternType: literal
           operation: Create
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-configs
             patternType: literal
           operation: Describe
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-configs
             patternType: literal
           operation: Read
           host: "*"
         # consumer group
         - resource:
             type: group
             name: connect-cluster
             patternType: literal
           operation: Read
           host: "*"

   Copy to Clipboard Toggle word wrap

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

   oc apply -f KAFKA-USER-CONFIG-FILE

   Copy to Clipboard Toggle word wrap