OpenShift에서 AMQ Streams 구성

절차

1. Kafka 리소스의 spec 속성을 편집합니다.

   구성할 수 있는 속성은 다음 예제 구성에 표시되어 있습니다.

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       replicas: 3 

   1

   
       version: 3.4.0 

   2

   
       logging: 

   3

   
         type: inline
         loggers:
           kafka.root.logger.level: "INFO"
       resources: 

   4

   
         requests:
           memory: 64Gi
           cpu: "8"
         limits:
           memory: 64Gi
           cpu: "12"
       readinessProbe: 

   5

   
         initialDelaySeconds: 15
         timeoutSeconds: 5
       livenessProbe:
         initialDelaySeconds: 15
         timeoutSeconds: 5
       jvmOptions: 

   6

   
         -Xms: 8192m
         -Xmx: 8192m
       image: my-org/my-image:latest 

   7

   
       listeners: 

   8

   
         - name: plain 

   9

   
           port: 9092 

   10

   
           type: internal 

   11

   
           tls: false 

   12

   
           configuration:
             useServiceDnsDomain: true 

   13

   
         - name: tls
           port: 9093
           type: internal
           tls: true
           authentication: 

   14

   
             type: tls
         - name: external 

   15

   
           port: 9094
           type: route
           tls: true
           configuration:
             brokerCertChainAndKey: 

   16

   
               secretName: my-secret
               certificate: my-certificate.crt
               key: my-key.key
       authorization: 

   17

   
         type: simple
       config: 

   18

   
         auto.create.topics.enable: "false"
         offsets.topic.replication.factor: 3
         transaction.state.log.replication.factor: 3
         transaction.state.log.min.isr: 2
         default.replication.factor: 3
         min.insync.replicas: 2
         inter.broker.protocol.version: "3.4"
         ssl.cipher.suites: TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384 

   19

   
         ssl.enabled.protocols: TLSv1.2
         ssl.protocol: TLSv1.2
       storage: 

   20

   
         type: persistent-claim 

   21

   
         size: 10000Gi 

   22

   
       rack: 

   23

   
         topologyKey: topology.kubernetes.io/zone
       metricsConfig: 

   24

   
         type: jmxPrometheusExporter
         valueFrom:
           configMapKeyRef: 

   25

   
             name: my-config-map
             key: my-key
       # ...
     zookeeper: 

   26

   
       replicas: 3 

   27

   
       logging: 

   28

   
         type: inline
         loggers:
           zookeeper.root.logger: "INFO"
       resources:
         requests:
           memory: 8Gi
           cpu: "2"
         limits:
           memory: 8Gi
           cpu: "2"
       jvmOptions:
         -Xms: 4096m
         -Xmx: 4096m
       storage:
         type: persistent-claim
         size: 1000Gi
       metricsConfig:
         # ...
     entityOperator: 

   29

   
       tlsSidecar: 

   30

   
         resources:
           requests:
             cpu: 200m
             memory: 64Mi
           limits:
             cpu: 500m
             memory: 128Mi
       topicOperator:
         watchedNamespace: my-topic-namespace
         reconciliationIntervalSeconds: 60
         logging: 

   31

   
           type: inline
           loggers:
             rootLogger.level: "INFO"
         resources:
           requests:
             memory: 512Mi
             cpu: "1"
           limits:
             memory: 512Mi
             cpu: "1"
       userOperator:
         watchedNamespace: my-topic-namespace
         reconciliationIntervalSeconds: 60
         logging: 

   32

   
           type: inline
           loggers:
             rootLogger.level: INFO
         resources:
           requests:
             memory: 512Mi
             cpu: "1"
           limits:
             memory: 512Mi
             cpu: "1"
     kafkaExporter: 

   33

   
       # ...
     cruiseControl: 

   34

   
       # ...

   Copy to Clipboard Toggle word wrap

   1

   복제본 노드 수입니다.

   2

   업그레이드 절차에 따라 지원되는 버전으로 변경할 수 있는 Kafka 버전.

   3

   Kafka 로거 및 로그 수준은 ConfigMap을 통해 직접 (인라인) 추가되거나 간접적으로 (외부)합니다. 사용자 지정 ConfigMap을 log4j.properties 키에 배치해야 합니다. Kafka kafka.root.logger.level 로거의 경우 로그 수준을 INFO, ERROR, WARN, TRACE, DEBUG, FATAL 또는 OFF로 설정할 수 있습니다.

   4

   지원되는 리소스 예약, 현재 cpu 및 memory 및 사용할 수 있는 최대 리소스를 지정하는 제한입니다.

   5

   컨테이너(liveness)를 다시 시작할 시기와 컨테이너가 트래픽을 수락할 수 있는 시기(가용성)를 확인할 수 있는 상태 점검 입니다.

   6

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

   7

   ADVANCED OPTION : 컨테이너 이미지 구성.이는 특별한 경우에만 권장됩니다.

   8

   리스너는 클라이언트가 부트스트랩 주소를 통해 Kafka 클러스터에 연결하는 방법을 구성합니다. 리스너는 OpenShift 클러스터 내부 또는 외부에서 연결하기 위해 내부 또는 외부 리스너로 구성됩니다.

   9

   리스너를 식별할 이름입니다. Kafka 클러스터 내에서 고유해야 합니다.

   10

   Kafka 내부의 리스너에서 사용하는 포트 번호입니다. 포트 번호는 지정된 Kafka 클러스터 내에서 고유해야 합니다. 허용되는 포트 번호는 9092이고 포트 9404 및 9999를 제외하고 이미 Prometheus 및 10.0.0.1에 사용됩니다. 리스너 유형에 따라 포트 번호가 Kafka 클라이언트를 연결하는 포트 번호와 동일하지 않을 수 있습니다.

   11

   내부 또는 cluster-ip 로 지정된 리스너 유형(broker별 ClusterIP 서비스를 사용하여 Kafka를 노출) 또는 외부 리스너의 경우 경로 (OpenShift만), 로드 밸런서,노드 포트 또는 수신 (Kubernetes만 해당)으로 지정됩니다.

   12

   각 리스너에 대해 TLS 암호화를 활성화합니다. 기본값은 false 입니다. 경로 리스너에는 TLS 암호화가 필요하지 않습니다.

   13

   클러스터 서비스 접미사(일반적으로 .cluster.local)를 포함한 정규화된 DNS 이름이 할당되는지 여부를 정의합니다.

   14

   mTLS, SCRAM-SHA-512 또는 토큰 기반 OAuth 2.0으로 지정된 리스너 인증 메커니즘.

   15

   외부 리스너 구성은 경로,loadbalancer 또는 nodeport를 통해 Kafka 클러스터가 OpenShift 외부에 노출되는 방법을 지정합니다.

   16

   외부 CA(인증 기관)에서 관리하는 Kafka 리스너 인증서에 대한 선택적 구성입니다. brokerCertECDHEAndKey 는 서버 인증서와 개인 키가 포함된 보안을 지정합니다. 활성화된 TLS 암호화를 사용하여 모든 리스너에서 Kafka 리스너 인증서를 구성할 수 있습니다.

   17

   권한 부여 를 사용하면 Kafka 브로커에서 간단한 OAUTH 2.0 또는 OPA 인증을 사용할 수 있습니다. 간단한 인증에서는 AclAuthorizer Kafka 플러그인을 사용합니다.

   18

   브로커 구성. 표준 Apache Kafka 구성을 제공하여 AMQ Streams에서 직접 관리하지 않는 속성으로 제한될 수 있습니다.

   19

   TLS 암호화가 활성화된 리스너의 SSL 속성은 특정 암호화 제품군 또는 TLS 버전을 활성화합니다.

   20

   스토리지 는 임시,persistent-claim 또는 jbod 로 구성됩니다.

   21

   영구 볼륨의 스토리지 크기가 증가할 수 있으며 추가 볼륨을 JBOD 스토리지에 추가할 수 있습니다.

   22

   영구 스토리지에는 동적 볼륨 프로비저닝을 위한 스토리지 ID 및 클래스 와 같은 추가 구성 옵션이 있습니다.

   23

   다양한 랙, 데이터 센터 또는 가용 영역에 복제본을 분산하기 위한 Rack 인식 구성입니다. topologyKey 는 rack ID가 포함된 노드 레이블과 일치해야 합니다. 이 구성에 사용된 예제에서는 표준 topology.kubernetes.io/zone 레이블을 사용하여 영역을 지정합니다.

   24

   Prometheus 지표 가 활성화되었습니다. 이 예에서는 Prometheus ScanSetting Exporter(기본 지표 내보내기)에 대한 지표가 구성됩니다.

   25

   Prometheus exporter에 대한 구성이 포함된 ConfigMap을 참조하여 활성화되는 Prometheus dashboard를 통해 Grafana 대시보드로 메트릭을 내보내는 규칙입니다. metricsConfig.valueFrom.configMapKeyRef.key 아래에 빈 파일이 포함된 ConfigMap에 대한 참조를 사용하여 추가 구성 없이 메트릭을 활성화할 수 있습니다.

   26

   Kafka 구성과 유사한 속성을 포함하는 zookeeper-specific 구성입니다.

   27

   ZooKeeper 노드 수입니다. zookeeper 클러스터 또는 ensembles는 일반적으로 홀수의 노드 (일반적으로 3, 5 또는 7)와 함께 실행됩니다. 효과적인 쿼럼을 유지하려면 대부분의 노드를 사용할 수 있어야 합니다. ZooKeeper 클러스터에서 쿼럼이 손실되면 클라이언트에 대한 응답이 중지되고 Kafka 브로커가 작동을 중지합니다. AMQ Streams에는 안정적이고 가용성이 높은 ZooKeeper 클러스터를 갖는 것이 중요합니다.

   28

   지정된 ZooKeeper 로거 및 로그 수준.

   29

   Topic Operator 및 User Operator에 대한 구성을 지정하는 entity Operator 구성

   30

   엔티티 Operator TLS 사이드카 구성. entity Operator는 TLS 사이드카를 사용하여 ZooKeeper와의 보안 통신을 지원합니다.

   31

   지정된 Topic Operator 로거 및 로그 수준. 이 예에서는 인라인 로깅을 사용합니다.

   32

   지정된 사용자 Operator 로거 및 로그 수준.

   33

   Kafka 내보내기 구성. Kafka 내보내기 는 Kafka 브로커, 특히 소비자 지연 데이터에서 메트릭 데이터를 추출하는 선택적 구성 요소입니다. Kafka 내보내기가 제대로 작동하려면 소비자 그룹을 사용 중이어야 합니다.

   34

   Kafka 클러스터를 재조정하는 데 사용되는 Cruise Control에 대한 선택적 구성입니다.

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

   oc apply -f <kafka_configuration_file>

   Copy to Clipboard Toggle word wrap

절차

1. OpenShift 콘솔을 사용하여 터미널을 열거나 CLI에서 exec 명령을 실행합니다.

   예를 들면 다음과 같습니다.

   oc exec -ti my-cluster-zookeeper-0 -- bin/kafka-topics.sh --list --zookeeper localhost:12181

   Copy to Clipboard Toggle word wrap

   localhost:12181 을 사용해야 합니다.

   이제 ZooKeeper에 Kafka 명령을 실행할 수 있습니다.

절차

1. 삭제할 Pod 의 이름을 찾습니다.

   Kafka 브로커 Pod의 이름은 < cluster-name> -kafka- <index >입니다. 여기서 < index >는 0에서 시작하여 총 복제본 수에서 1을 뺀 값입니다. 예: my-cluster-kafka-0.

2. OpenShift에서 포드 리소스에 주석을 답니다.

   oc annotate:

   oc annotate pod cluster-name-kafka-index strimzi.io/delete-pod-and-pvc=true

   Copy to Clipboard Toggle word wrap
3. 다음 조정이 있을 때까지 기다린 후 기본 영구 볼륨 클레임을 사용하여 주석이 달린 Pod가 삭제되고 다시 생성됩니다.

절차

1. 삭제할 Pod 의 이름을 찾습니다.

   zookeeper Pod의 이름은 < cluster-name> -zookeeper- <index >입니다. 여기서 < index >는 0에서 시작하여 총 복제본 수에서 1을 뺀 값입니다. 예: my-cluster-zookeeper-0.

2. OpenShift에서 포드 리소스에 주석을 답니다.

   oc annotate:

   oc annotate pod cluster-name-zookeeper-index strimzi.io/delete-pod-and-pvc=true

   Copy to Clipboard Toggle word wrap
3. 다음 조정이 있을 때까지 기다린 후 기본 영구 볼륨 클레임을 사용하여 주석이 달린 Pod가 삭제되고 다시 생성됩니다.

절차

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

   구성할 수 있는 속성은 다음 예제 구성에 표시되어 있습니다.

   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/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
     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

   Kafka Connect 클러스터에 대한 인증: mTLS,토큰 기반 OAuth, SASL 기반 SCRAM 기반 SCRAM-256/SCRAM-SHA -512, 또는 PLAIN 으로 지정됩니다. 기본적으로 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

   컨테이너(liveness)를 다시 시작할 시기와 컨테이너가 트래픽을 수락할 수 있는 시기(가용성)를 확인할 수 있는 상태 점검 입니다.

   15

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

   16

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

   17

   ADVANCED OPTION : 컨테이너 이미지 구성.이는 특별한 경우에만 권장됩니다.

   18

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

   19

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

   20

   환경 변수는 분산 추적을 위해 설정됩니다.

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

   oc apply -f KAFKA-CONNECT-CONFIG-FILE

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

절차

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

   다음 예제에서는 리터럴 이름 값을 사용하여 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
           operations:
             - Create
             - Describe
             - Read
             - Write
           host: "*"
         # access to status.storage.topic
         - resource:
             type: topic
             name: connect-cluster-status
             patternType: literal
           operations:
             - Create
             - Describe
             - Read
             - Write
           host: "*"
         # access to config.storage.topic
         - resource:
             type: topic
             name: connect-cluster-configs
             patternType: literal
           operations:
             - Create
             - Describe
             - Read
             - Write
           host: "*"
         # consumer group
         - resource:
             type: group
             name: connect-cluster
             patternType: literal
           operations:
             - Read
           host: "*"

   Copy to Clipboard Toggle word wrap

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

   oc apply -f KAFKA-USER-CONFIG-FILE

   Copy to Clipboard Toggle word wrap

절차

1. KafkaMirrorMaker2 리소스의 spec 속성을 편집합니다.

   구성할 수 있는 속성은 다음 예제 구성에 표시되어 있습니다.

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaMirrorMaker2
   metadata:
     name: my-mirror-maker2
   spec:
     version: 3.4.0 

   1

   
     replicas: 3 

   2

   
     connectCluster: "my-cluster-target" 

   3

   
     clusters: 

   4

   
     - alias: "my-cluster-source" 

   5

   
       authentication: 

   6

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

   7

   
       tls: 

   8

   
         trustedCertificates:
         - certificate: ca.crt
           secretName: my-cluster-source-cluster-ca-cert
     - alias: "my-cluster-target" 

   9

   
       authentication: 

   10

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

   11

   
       config: 

   12

   
         config.storage.replication.factor: 1
         offset.storage.replication.factor: 1
         status.storage.replication.factor: 1
         ssl.cipher.suites: TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384 

   13

   
         ssl.enabled.protocols: TLSv1.2
         ssl.protocol: TLSv1.2
         ssl.endpoint.identification.algorithm: HTTPS 

   14

   
       tls: 

   15

   
         trustedCertificates:
         - certificate: ca.crt
           secretName: my-cluster-target-cluster-ca-cert
     mirrors: 

   16

   
     - sourceCluster: "my-cluster-source" 

   17

   
       targetCluster: "my-cluster-target" 

   18

   
       sourceConnector: 

   19

   
         tasksMax: 10 

   20

   
         autoRestart: 

   21

   
           enabled: true
         config:
           replication.factor: 1 

   22

   
           offset-syncs.topic.replication.factor: 1 

   23

   
           sync.topic.acls.enabled: "false" 

   24

   
           refresh.topics.interval.seconds: 60 

   25

   
           replication.policy.separator: "." 

   26

   
           replication.policy.class: "org.apache.kafka.connect.mirror.IdentityReplicationPolicy" 

   27

   
       heartbeatConnector: 

   28

   
         autoRestart:
           enabled: true
         config:
           heartbeats.topic.replication.factor: 1 

   29

   
       checkpointConnector: 

   30

   
         autoRestart:
           enabled: true
         config:
           checkpoints.topic.replication.factor: 1 

   31

   
           refresh.groups.interval.seconds: 600 

   32

   
           sync.group.offsets.enabled: true 

   33

   
           sync.group.offsets.interval.seconds: 60 

   34

   
           emit.checkpoints.interval.seconds: 60 

   35

   
           replication.policy.class: "org.apache.kafka.connect.mirror.IdentityReplicationPolicy"
       topicsPattern: "topic1|topic2|topic3" 

   36

   
       groupsPattern: "group1|group2|group3" 

   37

   
     resources: 

   38

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

   39

   
       type: inline
       loggers:
         connect.root.logger.level: "INFO"
     readinessProbe: 

   40

   
       initialDelaySeconds: 15
       timeoutSeconds: 5
     livenessProbe:
       initialDelaySeconds: 15
       timeoutSeconds: 5
     jvmOptions: 

   41

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

   42

   
     rack:
       topologyKey: topology.kubernetes.io/zone 

   43

   
     template: 

   44

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

   45

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

   46

   
     externalConfiguration: 

   47

   
       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

   Copy to Clipboard Toggle word wrap

   1

   Kafka Connect 및 Mirror 2.0 버전 2.0은 항상 동일합니다.

   2

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

   3

   Kafka Connect의 Kafka 클러스터 별칭 - 대상 Kafka 클러스터를 지정해야 합니다. Kafka 클러스터는 내부 항목에 대해 Kafka Connect에서 사용합니다.

   4

   동기화되는 Kafka 클러스터의 사양 입니다.

   5

   소스 Kafka 클러스터의 클러스터 별칭 입니다.

   6

   mTLS,토큰 기반 OAuth, SASL 기반 SCRAM -256/SCRAM-SHA -512, 또는 PLAIN 으로 지정된 소스 클러스터에 대한 인증입니다.

   7

   소스 Kafka 클러스터에 연결하기 위한 부트스트랩 서버 입니다.

   8

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

   9

   대상 Kafka 클러스터의 클러스터 별칭 입니다.

   10

   대상 Kafka 클러스터에 대한 인증은 소스 Kafka 클러스터와 동일한 방식으로 구성됩니다.

   11

   대상 Kafka 클러스터에 연결하기 위한 부트스트랩 서버 입니다.

   12

   Kafka Connect 구성. 표준 Apache Kafka 구성을 제공하여 AMQ Streams에서 직접 관리하지 않는 속성으로 제한될 수 있습니다.

   13

   외부 리스너가 TLS 버전에 대한 특정 암호화 제품군으로 실행되도록 하는 SSL 속성입니다.

   14

   HTTPS 로 설정하여 호스트 이름 확인이 활성화됩니다. 빈 문자열은 확인을 비활성화합니다.

   15

   대상 Kafka 클러스터에 대한 TLS 암호화는 소스 Kafka 클러스터와 동일한 방식으로 구성됩니다.

   16

   MirrorMaker 2 커넥터.

   17

   MirrorMaker 2 커넥터에서 사용하는 소스 클러스터의 클러스터 별칭 입니다.

   18

   MirrorMaker 2 커넥터에서 사용하는 대상 클러스터의 클러스터 별칭 입니다.

   19

   원격 주제를 생성하는 MirrorSourceConnector 구성입니다. 구성이 기본 구성 옵션을 덮어씁니다.

   20

   커넥터가 생성할 수 있는 최대 작업 수입니다. 작업은 데이터 복제를 처리하고 병렬로 실행합니다. 인프라에서 처리 오버헤드를 지원하는 경우 이 값을 늘리면 처리량이 향상될 수 있습니다. Kafka Connect는 클러스터 멤버 간에 작업을 배포합니다. 작업자보다 많은 작업이 있는 경우 작업자에는 여러 작업이 할당됩니다. 싱크 커넥터의 경우 소비된 각 주제 파티션에 대해 하나의 작업이 있어야 합니다. 소스 커넥터의 경우 병렬로 실행할 수 있는 작업 수도 외부 시스템에 따라 달라질 수 있습니다. 커넥터는 병렬 처리를 수행할 수 없는 경우 최대 작업 수보다 적은 수를 생성합니다.

   21

   실패한 커넥터 및 작업의 자동 재시작을 활성화합니다. 재시작 횟수는 최대 7개까지 수행되며, 이 후에는 수동으로 다시 시작해야 합니다.

   22

   대상 클러스터에서 생성된 미러링된 항목의 복제 요인입니다.

   23

   소스 및 대상 클러스터 의 오프셋을 매핑하는 MirrorSourceConnector 오프셋의 복제 요인입니다.

   24

   ACL 규칙 동기화 가 활성화되면 ACL이 동기화된 항목에 적용됩니다. 기본값은 true입니다. 이 기능은 User Operator와 호환되지 않습니다. User Operator를 사용하는 경우 이 속성을 false 로 설정합니다.

   25

   새 항목에 대한 검사 빈도를 변경하는 선택적 설정입니다. 기본값은 10분마다 점검을 위한 것입니다.

   26

   원격 주제 이름 변경에 사용되는 구분자를 정의합니다.

   27

   원격 주제의 자동 이름 변경을 재정의하는 정책을 추가합니다. 소스 클러스터 이름 앞에 이름 앞에 추가하지 않고 주제에서 원래 이름을 유지합니다. 이 선택적 설정은 활성/수동 백업 및 데이터 마이그레이션에 유용합니다. 주제 오프셋 동기화를 구성하려면 이 속성도 checkpointConnector.config 에 대해 설정해야 합니다.

   28

   연결 확인을 수행하는 MirrorHeartbeatConnector 구성입니다. 구성이 기본 구성 옵션을 덮어씁니다.

   29

   대상 클러스터에서 생성된 하트비트 항목에 대한 복제 요인입니다.

   30

   오프셋을 추적하는 MirrorCheckpointConnector 구성입니다. 구성이 기본 구성 옵션을 덮어씁니다.

   31

   대상 클러스터에서 생성된 체크포인트 항목에 대한 복제 요인입니다.

   32

   새 소비자 그룹에 대한 검사 빈도를 변경하는 선택적 설정입니다. 기본값은 10분마다 점검을 위한 것입니다.

   33

   활성/수동 구성에서 복구에 유용한 소비자 그룹 오프셋을 동기화하는 선택적 설정입니다. 동기화는 기본적으로 활성화되어 있지 않습니다.

   34

   소비자 그룹 오프셋의 동기화가 활성화된 경우 동기화 빈도를 조정할 수 있습니다.

   35

   오프셋 추적에 대한 검사 빈도를 조정합니다. 오프셋 동기화의 빈도를 변경하는 경우 이러한 검사의 빈도를 조정해야 할 수도 있습니다.

   36

   쉼표로 구분된 목록 또는 정규식 패턴으로 정의된 소스 클러스터의 주제 복제 주제 복제 . 소스 커넥터는 지정된 주제를 복제합니다. 검사점 커넥터는 지정된 항목에 대한 오프셋을 추적합니다. 여기에서는 이름으로 세 가지 주제를 요청합니다.

   37

   쉼표로 구분된 목록 또는 정규식 패턴으로 정의된 소스 클러스터의 소비자 그룹 복제입니다. 검사점 커넥터는 지정된 소비자 그룹을 복제합니다. 여기에서는 이름별로 세 개의 소비자 그룹을 요청합니다.

   38

   지원되는 리소스 예약, 현재 cpu 및 memory 및 사용할 수 있는 최대 리소스를 지정하는 제한입니다.

   39

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

   40

   컨테이너(liveness)를 다시 시작할 시기와 컨테이너가 트래픽을 수락할 수 있는 시기(가용성)를 확인할 수 있는 상태 점검 입니다.

   41

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

   42

   ADVANCED OPTION : 컨테이너 이미지 구성.이는 특별한 경우에만 권장됩니다.

   43

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

   44

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

   45

   환경 변수는 분산 추적을 위해 설정됩니다.

   46

   Jaeger에 대해 분산 추적이 활성화됩니다.

   47

   Kafka MirrorMaker에 환경 변수로 마운트된 OpenShift Secret의 외부 구성 구성 공급자 플러그인 을 사용하여 외부 소스에서 구성 값을 로드할 수도 있습니다.

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

   oc apply -f MIRRORMAKER-CONFIGURATION-FILE

   Copy to Clipboard Toggle word wrap

절차

1. 두 개의 Kafka 리소스를 구성합니다. 하나는 소스 Kafka 클러스터를 보호하고 하나는 대상 Kafka 클러스터를 보호합니다.

   인증을 위한 리스너 구성을 추가하고 권한 부여를 활성화할 수 있습니다.

   이 예에서 내부 리스너는 TLS 암호화 및 mTLS 인증을 사용하는 Kafka 클러스터에 대해 구성됩니다. Kafka 간단한 인증이 활성화됩니다.

   TLS 암호화 및 mTLS 인증을 사용하는 소스 Kafka 클러스터 구성의 예

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-source-cluster
   spec:
     kafka:
       version: 3.4.0
       replicas: 1
       listeners:
         - name: tls
           port: 9093
           type: internal
           tls: true
           authentication:
             type: tls
       authorization:
         type: simple
       config:
         offsets.topic.replication.factor: 1
         transaction.state.log.replication.factor: 1
         transaction.state.log.min.isr: 1
         default.replication.factor: 1
         min.insync.replicas: 1
         inter.broker.protocol.version: "3.4"
       storage:
         type: jbod
         volumes:
         - id: 0
           type: persistent-claim
           size: 100Gi
           deleteClaim: false
     zookeeper:
       replicas: 1
       storage:
         type: persistent-claim
         size: 100Gi
         deleteClaim: false
     entityOperator:
       topicOperator: {}
       userOperator: {}

   Copy to Clipboard Toggle word wrap

   TLS 암호화 및 mTLS 인증을 사용하는 대상 Kafka 클러스터 구성의 예

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-target-cluster
   spec:
     kafka:
       version: 3.4.0
       replicas: 1
       listeners:
         - name: tls
           port: 9093
           type: internal
           tls: true
           authentication:
             type: tls
       authorization:
         type: simple
       config:
         offsets.topic.replication.factor: 1
         transaction.state.log.replication.factor: 1
         transaction.state.log.min.isr: 1
         default.replication.factor: 1
         min.insync.replicas: 1
         inter.broker.protocol.version: "3.4"
       storage:
         type: jbod
         volumes:
           - id: 0
             type: persistent-claim
             size: 100Gi
             deleteClaim: false
     zookeeper:
       replicas: 1
       storage:
         type: persistent-claim
         size: 100Gi
         deleteClaim: false
     entityOperator:
       topicOperator: {}
       userOperator: {}

   Copy to Clipboard Toggle word wrap

2. 별도의 네임스페이스에서 Kafka 리소스를 생성하거나 업데이트합니다.

   oc apply -f <kafka_configuration_file> -n <namespace>

   Copy to Clipboard Toggle word wrap

   Cluster Operator는 리스너를 생성하고 클러스터 및 클라이언트 CA(인증 기관) 인증서를 설정하여 Kafka 클러스터 내에서 인증을 활성화합니다.

   인증서는 < cluster_name> -cluster-ca-cert 에서 생성됩니다.

3. 소스 Kafka 클러스터의 사용자 및 대상 Kafka 클러스터의 사용자용으로 하나씩 두 개의 KafkaUser 리소스를 구성합니다.

   1. 해당 소스 및 대상 Kafka 클러스터와 동일한 인증 및 권한 부여 유형을 구성합니다. 예를 들어 소스 Kafka 클러스터의 Kafka 구성에서 tls 인증 및 간단한 권한 부여 유형을 사용한 경우 KafkaUser 구성에서 동일하게 사용합니다.

   2. 소스 및 대상 Kafka 클러스터에서 작업을 허용하도록 MirrorMaker 2에 필요한 ACL을 구성합니다.

      ACL은 내부 MirrorMaker 커넥터와 기본 Kafka Connect 프레임워크에서 사용합니다.

   mTLS 인증에 대한 소스 사용자 구성 예

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaUser
   metadata:
     name: my-source-user
     labels:
       strimzi.io/cluster: my-source-cluster
   spec:
     authentication:
       type: tls
     authorization:
       type: simple
       acls:
         # MirrorSourceConnector
         - resource: # Not needed if offset-syncs.topic.location=target
             type: topic
             name: mm2-offset-syncs.my-target-cluster.internal
           operations:
             - Create
             - DescribeConfigs
             - Read
             - Write
         - resource: # Needed for every topic which is mirrored
             type: topic
             name: "*"
           operations:
             - DescribeConfigs
             - Read
         # MirrorCheckpointConnector
         - resource:
             type: cluster
           operations:
             - Describe
         - resource: # Needed for every group for which offsets are synced
             type: group
             name: "*"
           operations:
             - Describe
         - resource: # Not needed if offset-syncs.topic.location=target
             type: topic
             name: mm2-offset-syncs.my-target-cluster.internal
           operations:
             - Read

   Copy to Clipboard Toggle word wrap

   mTLS 인증에 대한 대상 사용자 구성 예

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaUser
   metadata:
     name: my-target-user
     labels:
       strimzi.io/cluster: my-target-cluster
   spec:
     authentication:
       type: tls
     authorization:
       type: simple
       acls:
         # Underlying Kafka Connect internal topics to store configuration, offsets, or status
         - resource:
             type: group
             name: mirrormaker2-cluster
           operations:
             - Read
         - resource:
             type: topic
             name: mirrormaker2-cluster-configs
           operations:
             - Create
             - Describe
             - DescribeConfigs
             - Read
             - Write
         - resource:
             type: topic
             name: mirrormaker2-cluster-status
           operations:
             - Create
             - Describe
             - DescribeConfigs
             - Read
             - Write
         - resource:
             type: topic
             name: mirrormaker2-cluster-offsets
           operations:
             - Create
             - Describe
             - DescribeConfigs
             - Read
             - Write
         # MirrorSourceConnector
         - resource: # Needed for every topic which is mirrored
             type: topic
             name: "*"
           operations:
             - Create
             - Alter
             - AlterConfigs
             - Write
         # MirrorCheckpointConnector
         - resource:
             type: cluster
           operations:
             - Describe
         - resource:
             type: topic
             name: my-source-cluster.checkpoints.internal
           operations:
             - Create
             - Describe
             - Read
             - Write
         - resource: # Needed for every group for which the offset is synced
             type: group
             name: "*"
           operations:
             - Read
             - Describe
         # MirrorHeartbeatConnector
         - resource:
             type: topic
             name: heartbeats
           operations:
             - Create
             - Describe
             - Write

   Copy to Clipboard Toggle word wrap

   참고

   type 을 tls-external 로 설정하여 User Operator 외부에서 발급한 인증서를 사용할 수 있습니다. 자세한 내용은 6.2.93절. “KafkaUserSpec schema reference”의 내용을 참조하십시오.

4. 소스 및 대상 Kafka 클러스터에 대해 생성한 각 네임스페이스에서 KafkaUser 리소스를 생성하거나 업데이트합니다.

   oc apply -f <kafka_user_configuration_file> -n <namespace>

   Copy to Clipboard Toggle word wrap

   User Operator는 선택한 인증 유형을 기반으로 클라이언트(MirrorMaker)와 클라이언트 인증에 사용되는 보안 인증 정보를 나타내는 사용자를 생성합니다.

   User Operator는 KafkaUser 리소스와 동일한 이름으로 새 시크릿을 생성합니다. 보안에는 mTLS 인증을 위한 개인 키와 공개 키가 포함되어 있습니다. 공개 키는 클라이언트 CA에서 서명한 사용자 인증서에 포함됩니다.

5. 인증 세부 정보를 사용하여 KafkaMirrorMaker2 리소스를 구성하여 소스 및 대상 Kafka 클러스터에 연결합니다.

   TLS 암호화 및 mTLS 인증을 사용하는 MirrorMaker 2 구성의 예

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaMirrorMaker2
   metadata:
     name: my-mirror-maker-2
   spec:
     version: 3.4.0
     replicas: 1
     connectCluster: "my-target-cluster"
     clusters:
       - alias: "my-source-cluster"
         bootstrapServers: my-source-cluster-kafka-bootstrap:9093
         tls: 

   1

   
           trustedCertificates:
             - secretName: my-source-cluster-cluster-ca-cert
               certificate: ca.crt
         authentication: 

   2

   
           type: tls
           certificateAndKey:
             secretName: my-source-user
             certificate: user.crt
             key: user.key
       - alias: "my-target-cluster"
         bootstrapServers: my-target-cluster-kafka-bootstrap:9093
         tls: 

   3

   
           trustedCertificates:
             - secretName: my-target-cluster-cluster-ca-cert
               certificate: ca.crt
         authentication: 

   4

   
           type: tls
           certificateAndKey:
             secretName: my-target-user
             certificate: user.crt
             key: user.key
         config:
           # -1 means it will use the default replication factor configured in the broker
           config.storage.replication.factor: -1
           offset.storage.replication.factor: -1
           status.storage.replication.factor: -1
     mirrors:
       - sourceCluster: "my-source-cluster"
         targetCluster: "my-target-cluster"
         sourceConnector:
           config:
             replication.factor: 1
             offset-syncs.topic.replication.factor: 1
             sync.topic.acls.enabled: "false"
         heartbeatConnector:
           config:
             heartbeats.topic.replication.factor: 1
         checkpointConnector:
           config:
             checkpoints.topic.replication.factor: 1
             sync.group.offsets.enabled: "true"
         topicsPattern: "topic1|topic2|topic3"
         groupsPattern: "group1|group2|group3"

   Copy to Clipboard Toggle word wrap

   1

   소스 Kafka 클러스터의 TLS 인증서입니다. 별도의 네임스페이스에 있는 경우 Kafka 클러스터의 네임스페이스에서 클러스터 보안을 복사합니다.

   2

   TLS 메커니즘 을 사용하여 소스 Kafka 클러스터에 액세스하는 사용자 인증입니다.

   3

   대상 Kafka 클러스터의 TLS 인증서입니다.

   4

   대상 Kafka 클러스터에 액세스하기 위한 사용자 인증입니다.

6. 대상 Kafka 클러스터와 동일한 네임스페이스에서 KafkaMirrorMaker2 리소스를 생성하거나 업데이트합니다.

   oc apply -f <mirrormaker2_configuration_file> -n <namespace_of_target_cluster>

   Copy to Clipboard Toggle word wrap

절차

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

   oc get KafkaMirrorMaker2

   Copy to Clipboard Toggle word wrap

2. KafkaMirrorMaker2 사용자 정의 리소스에서 다시 시작할 Kafka MirrorMaker 2 커넥터의 이름을 찾습니다.

   oc describe KafkaMirrorMaker2 KAFKAMIRRORMAKER-2-NAME

   Copy to Clipboard Toggle word wrap

3. 커넥터를 다시 시작하려면 OpenShift에서 KafkaMirrorMaker2 리소스에 주석을 답니다. 이 예에서 oc annotate 는 my-source->my-target.MirrorSourceConnector라는 커넥터를 다시 시작합니다.

   oc annotate KafkaMirrorMaker2 KAFKAMIRRORMAKER-2-NAME "strimzi.io/restart-connector=my-source->my-target.MirrorSourceConnector"

   Copy to Clipboard Toggle word wrap

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

   조정 프로세스에서 주석을 탐지한 한 Kafka MirrorMaker 2 커넥터가 재시작됩니다. 재시작 요청이 승인되면 KafkaMirrorMaker2 사용자 정의 리소스에서 주석이 제거됩니다.