2.4. Kafka MirrorMaker 2 클러스터 구성

절차

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
       tls: 

   13

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

   14

   
     - sourceCluster: "my-cluster-source" 

   15

   
       targetCluster: "my-cluster-target" 

   16

   
       sourceConnector: 

   17

   
         tasksMax: 10 

   18

   
         autoRestart: 

   19

   
           enabled: true
         config:
           replication.factor: 1 

   20

   
           offset-syncs.topic.replication.factor: 1 

   21

   
           sync.topic.acls.enabled: "false" 

   22

   
           refresh.topics.interval.seconds: 60 

   23

   
           replication.policy.separator: "." 

   24

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

   25

   
       heartbeatConnector: 

   26

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

   27

   
       checkpointConnector: 

   28

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

   29

   
           refresh.groups.interval.seconds: 600 

   30

   
           sync.group.offsets.enabled: true 

   31

   
           sync.group.offsets.interval.seconds: 60 

   32

   
           emit.checkpoints.interval.seconds: 60 

   33

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

   34

   
       groupsPattern: "group1|group2|group3" 

   35

   
     resources: 

   36

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

   37

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

   38

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

   39

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

   40

   
     rack:
       topologyKey: topology.kubernetes.io/zone 

   41

   
     template: 

   42

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

   43

   
         env:
           - name: OTEL_SERVICE_NAME
             value: my-otel-service
           - name: OTEL_EXPORTER_OTLP_ENDPOINT
             value: "http://otlp-host:4317"
     tracing:
       type: opentelemetry 

   44

   
     externalConfiguration: 

   45

   
       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

   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

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

   14

   MirrorMaker 2 커넥터.

   15

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

   16

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

   17

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

   18

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

   19

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

   20

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

   21

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

   22

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

   23

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

   24

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

   25

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

   26

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

   27

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

   28

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

   29

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

   30

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

   31

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

   32

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

   33

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

   34

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

   35

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

   36

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

   37

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

   38

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

   39

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

   40

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

   41

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

   42

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

   43

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

   44

   OpenTelemetry를 사용하여 분산 추적을 활성화합니다.

   45

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

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

   oc apply -f MIRRORMAKER-CONFIGURATION-FILE

절차

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: {}

   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: {}

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

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

   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

   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

   참고

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

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

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

   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"

   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>

절차

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

   oc get KafkaMirrorMaker2

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

   oc describe KafkaMirrorMaker2 KAFKAMIRRORMAKER-2-NAME

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"

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

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

절차

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

   oc get KafkaMirrorMaker2

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

   oc describe KafkaMirrorMaker2 KAFKAMIRRORMAKER-2-NAME

3. 커넥터 작업을 다시 시작하려면 OpenShift에서 KafkaMirrorMaker2 리소스에 주석을 답니다. 이 예에서 oc annotate restarts 작업 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"

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

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