2.5. Kafka MirrorMaker 2.0 클러스터 구성

절차

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

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

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

   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

   
         config:
           replication.factor: 1 

   21

   
           offset-syncs.topic.replication.factor: 1 

   22

   
           sync.topic.acls.enabled: "false" 

   23

   
           refresh.topics.interval.seconds: 60 

   24

   
           replication.policy.separator: "" 

   25

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

   26

   
       heartbeatConnector: 

   27

   
         config:
           heartbeats.topic.replication.factor: 1 

   28

   
       checkpointConnector: 

   29

   
         config:
           checkpoints.topic.replication.factor: 1 

   30

   
           refresh.groups.interval.seconds: 600 

   31

   
           sync.group.offsets.enabled: true 

   32

   
           sync.group.offsets.interval.seconds: 60 

   33

   
           emit.checkpoints.interval.seconds: 60 

   34

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

   35

   
       groupsPattern: "group1|group2|group3" 

   36

   
     resources: 

   37

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

   38

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

   39

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

   40

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

   41

   
     rack:
       topologyKey: topology.kubernetes.io/zone 

   42

   
     template: 

   43

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

   44

   
         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 

   45

   
     externalConfiguration: 

   46

   
       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 Maker 2.0 버전은 항상 동일합니다.

   2

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

   3

   Kafka Connect의 Kafka 클러스터 별칭 은 대상 Kafka 클러스터를 지정해야 합니다. Kafka 클러스터는 내부 주제로 Kafka Connect에서 사용합니다.

   4

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

   5

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

   6

   여기에 표시된 대로 TLS 메커니즘 을 사용하는 소스 클러스터에 대한 인증 또는 SASL 기반 SCRAM-SHA-256/SCRAM-SHA-512 또는 PLAIN 메커니즘을 사용합니다. ???

   7

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

   8

   TLS 인증서가 소스 Kafka 클러스터의 X.509 형식으로 저장되는 키 이름이 있는 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.0 커넥터.

   17

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

   18

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

   19

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

   20

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

   21

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

   22

   소스 및 대상 클러스터의 오프셋을 매핑하는 MirrorSourceConnector offset-syncs 내부 주제의 복제 인수입니다.

   23

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

   24

   새 주제의 검사 빈도를 변경하는 선택적 설정입니다. 기본값은 10분마다 확인용입니다.

   25

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

   26

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

   27

   연결 검사를 수행하는 MirrorHeartbeatConnector 에 대한 구성입니다. 구성이 기본 구성 옵션을 덮어씁니다.

   28

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

   29

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

   30

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

   31

   새 소비자 그룹의 점검 빈도를 변경하는 선택적 설정입니다. 기본값은 10분마다 확인용입니다.

   32

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

   33

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

   34

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

   35

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

   36

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

   37

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

   38

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

   39

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

   40

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

   41

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

   42

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

   43

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

   44

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

   45

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

   46

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

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

   oc apply -f MIRRORMAKER-CONFIGURATION-FILE

   Copy to Clipboard Toggle word wrap

절차

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

   인증에 대한 리스너 구성을 추가하고 권한을 활성화할 수 있습니다.

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

   TLS 인증을 사용한 소스 Kafka 클러스터 구성의 예

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-source-cluster
   spec:
     kafka:
       version: 3.2.3
       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.2"
       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 인증을 사용한 대상 Kafka 클러스터 구성 예

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-target-cluster
   spec:
     kafka:
       version: 3.2.3
       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.2"
       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는 리스너를 생성하고 Kafka 클러스터 내에서 인증을 활성화하기 위해 클러스터 및 클라이언트 CA(인증 기관) 인증서를 설정합니다.

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

3. 두 개의 Kafka 리소스를 구성합니다. 하나는 소스 Kafka 클러스터의 사용자용이고 하나는 대상 Kafka 클러스터 사용자용입니다.

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

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

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

   TLS 클라이언트 인증을 위한 소스 사용자 구성 예

   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
           operation: Create
         - resource: # Not needed if offset-syncs.topic.location=target
             type: topic
             name: mm2-offset-syncs.my-target-cluster.internal
           operation: DescribeConfigs
         - resource: # Not needed if offset-syncs.topic.location=target
             type: topic
             name: mm2-offset-syncs.my-target-cluster.internal
           operation: Write
         - resource: # Needed for every topic which is mirrored
             type: topic
             name: "*"
           operation: Read
         - resource: # Needed for every topic which is mirrored
             type: topic
             name: "*"
           operation: DescribeConfigs
         # MirrorCheckpointConnector
         - resource:
             type: cluster
           operation: Describe
         - resource: # Needed for every group for which offsets are synced
             type: group
             name: "*"
           operation: Describe
         - resource: # Not needed if offset-syncs.topic.location=target
             type: topic
             name: mm2-offset-syncs.my-target-cluster.internal
           operation: Read

   Copy to Clipboard Toggle word wrap

   TLS 클라이언트 인증을 위한 대상 사용자 구성 예

   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
           operation: Read
         - resource:
             type: topic
             name: mirrormaker2-cluster-configs
           operation: Read
         - resource:
             type: topic
             name: mirrormaker2-cluster-configs
           operation: Describe
         - resource:
             type: topic
             name: mirrormaker2-cluster-configs
           operation: DescribeConfigs
         - resource:
             type: topic
             name: mirrormaker2-cluster-configs
           operation: Write
         - resource:
             type: topic
             name: mirrormaker2-cluster-configs
           operation: Create
         - resource:
             type: topic
             name: mirrormaker2-cluster-status
           operation: Read
         - resource:
             type: topic
             name: mirrormaker2-cluster-status
           operation: Describe
         - resource:
             type: topic
             name: mirrormaker2-cluster-status
           operation: DescribeConfigs
         - resource:
             type: topic
             name: mirrormaker2-cluster-status
           operation: Write
         - resource:
             type: topic
             name: mirrormaker2-cluster-status
           operation: Create
         - resource:
             type: topic
             name: mirrormaker2-cluster-offsets
           operation: Read
         - resource:
             type: topic
             name: mirrormaker2-cluster-offsets
           operation: Write
         - resource:
             type: topic
             name: mirrormaker2-cluster-offsets
           operation: Describe
         - resource:
             type: topic
             name: mirrormaker2-cluster-offsets
           operation: DescribeConfigs
         - resource:
             type: topic
             name: mirrormaker2-cluster-offsets
           operation: Create
         # MirrorSourceConnector
         - resource: # Needed for every topic which is mirrored
             type: topic
             name: "*"
           operation: Create
         - resource: # Needed for every topic which is mirrored
             type: topic
             name: "*"
           operation: Alter
         - resource: # Needed for every topic which is mirrored
             type: topic
             name: "*"
           operation: AlterConfigs
         - resource: # Needed for every topic which is mirrored
             type: topic
             name: "*"
           operation: Write
         # MirrorCheckpointConnector
         - resource:
             type: cluster
           operation: Describe
         - resource:
             type: topic
             name: my-source-cluster.checkpoints.internal
           operation: Create
         - resource:
             type: topic
             name: my-source-cluster.checkpoints.internal
           operation: Describe
         - resource:
             type: topic
             name: my-source-cluster.checkpoints.internal
           operation: Write
         - resource: # Needed for every group for which the offset is synced
             type: group
             name: "*"
           operation: Read
         - resource: # Needed for every group for which the offset is synced
             type: group
             name: "*"
           operation: Describe
         - resource: # Needed for every topic which is mirrored
             type: topic
             name: "*"
           operation: Read
         # MirrorHeartbeatConnector
         - resource:
             type: topic
             name: heartbeats
           operation: Create
         - resource:
             type: topic
             name: heartbeats
           operation: Describe
         - resource:
             type: topic
             name: heartbeats
           operation: Write

   Copy to Clipboard Toggle word wrap

   참고

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

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

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

   Copy to Clipboard Toggle word wrap

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

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

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

   TLS 인증을 사용한 MirrorMaker 2.0 구성의 예

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaMirrorMaker2
   metadata:
     name: my-mirror-maker-2
   spec:
     version: 3.2.3
     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.0 커넥터를 제어하는 KafkaMirrorMaker2 사용자 정의 리소스의 이름을 찾습니다.

   oc get KafkaMirrorMaker2

   Copy to Clipboard Toggle word wrap

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

   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.0 커넥터가 다시 시작됩니다. 재시작 요청이 수락되면 KafkaMirrorMaker2 사용자 정의 리소스에서 주석이 제거됩니다.

절차

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

   oc get KafkaMirrorMaker2

   Copy to Clipboard Toggle word wrap

2. Kafka MirrorMaker 2.0 커넥터의 이름과 KafkaMirrorMaker2 사용자 정의 리소스에서 다시 시작할 작업의 ID를 찾습니다. 작업 ID는 0부터 시작하여 음수가 아닌 정수입니다.

   oc describe KafkaMirrorMaker2 KAFKAMIRRORMAKER-2-NAME

   Copy to Clipboard Toggle word wrap

3. 커넥터 작업을 다시 시작하려면 OpenShift의 KafkaMirrorMaker2 리소스에 주석을 답니다. 이 예제에서 oc annotate start task 0 of a connector named my-source->my-target.MirrorSourceConnector:

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

   Copy to Clipboard Toggle word wrap

4. 다음 조정이 발생할 때까지 기다립니다(기본적으로 2분마다).

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