Strimzi KafkaMirrorMaker2 - src not able to connect to target cluster (src & target clusters on GKE)

[karanalang]

Hello All - I'm trying to setup KafkaMirrorMaker2 on Strimzi kafka (on GKE), and it is not working. Currently, i've setup 2 clusters without SSL and authorization enabled.

src cluster -> versa-kafka-gke-w target cluster -> versa-kafka-gke-e

The kafkamirrormaker2 is deployed in src cluster (i,e. versa-kafka-gke-w) also Kafka version is 3.0.0

Here is the kafka-deployment.yaml (same yaml is used in both the clusters, the name is changed per cluster)

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: versa-kafka-gke-w #1
spec:
  kafka:
    version: 3.0.0
    replicas: 3
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: external
        port: 9094
        type: loadbalancer
        tls: false 
    template:
      clusterCaCert:
        metadata:
          labels:
             versa_replicated : "true" 
          annotations:
            reflector.v1.k8s.emberstack.com/reflection-allowed: "true"
            reflector.v1.k8s.emberstack.com/reflection-auto-enabled: "true"
            reflector.v1.k8s.emberstack.com/reflection-allowed-namespaces: "ueba.*"
    readinessProbe:
      initialDelaySeconds: 15
      timeoutSeconds: 5
    livenessProbe:
      initialDelaySeconds: 15
      timeoutSeconds: 5     
    config:
      auto.create.topics.enable: "false"
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2
      log.message.format.version: "3.0"
      inter.broker.protocol.version: "3.0"
    storage:
      type: jbod
      volumes:
      - id: 0
        type: persistent-claim
        size: 500Gi
        deleteClaim: false
    logging: #9
      type: inline
      loggers:
        kafka.root.logger.level: "INFO"
    metricsConfig:
      type: jmxPrometheusExporter
      valueFrom:
        configMapKeyRef:
          name: kafka-metrics
          key: kafka-metrics-config.yml    
  zookeeper:
    replicas: 3
    storage:
      type: persistent-claim
      size: 20Gi
      deleteClaim: false
    resources:
      requests:
        memory: 1Gi
        cpu: "1"
      limits:
        memory: 2Gi
        cpu: "1.5"
    logging:
      type: inline
      loggers:
        zookeeper.root.logger: "INFO"
    metricsConfig:
      type: jmxPrometheusExporter
      valueFrom:
        configMapKeyRef:
          name: kafka-metrics
          key: zookeeper-metrics-config.yml    
  entityOperator: #11
    topicOperator: {}
    userOperator: {}
  kafkaExporter:
    topicRegex: ".*"
    groupRegex: ".*"

---
kind: ConfigMap
apiVersion: v1
metadata:
  name: kafka-metrics
  labels:
    app: strimzi
data:
  kafka-metrics-config.yml: |
    # See https://github.com/prometheus/jmx_exporter for more info about JMX Prometheus Exporter metrics
    lowercaseOutputName: true
    rules:
    # Special cases and very specific rules
    - pattern: kafka.server<type=(.+), name=(.+), clientId=(.+), topic=(.+), partition=(.*)><>Value
      name: kafka_server_$1_$2
      type: GAUGE
      labels:
       clientId: "$3"
       topic: "$4"
       partition: "$5"
    - pattern: kafka.server<type=(.+), name=(.+), clientId=(.+), brokerHost=(.+), brokerPort=(.+)><>Value
      name: kafka_server_$1_$2
      type: GAUGE
      labels:
       clientId: "$3"
       broker: "$4:$5"
    - pattern: kafka.server<type=(.+), cipher=(.+), protocol=(.+), listener=(.+), networkProcessor=(.+)><>connections
      name: kafka_server_$1_connections_tls_info
      type: GAUGE
      labels:
        cipher: "$2"
        protocol: "$3"
        listener: "$4"
        networkProcessor: "$5"
    - pattern: kafka.server<type=(.+), clientSoftwareName=(.+), clientSoftwareVersion=(.+), listener=(.+), networkProcessor=(.+)><>connections
      name: kafka_server_$1_connections_software
      type: GAUGE
      labels:
        clientSoftwareName: "$2"
        clientSoftwareVersion: "$3"
        listener: "$4"
        networkProcessor: "$5"
    - pattern: "kafka.server<type=(.+), listener=(.+), networkProcessor=(.+)><>(.+):"
      name: kafka_server_$1_$4
      type: GAUGE
      labels:
       listener: "$2"
       networkProcessor: "$3"
    - pattern: kafka.server<type=(.+), listener=(.+), networkProcessor=(.+)><>(.+)
      name: kafka_server_$1_$4
      type: GAUGE
      labels:
       listener: "$2"
       networkProcessor: "$3"
    # Some percent metrics use MeanRate attribute
    # Ex) kafka.server<type=(KafkaRequestHandlerPool), name=(RequestHandlerAvgIdlePercent)><>MeanRate
    - pattern: kafka.(\w+)<type=(.+), name=(.+)Percent\w*><>MeanRate
      name: kafka_$1_$2_$3_percent
      type: GAUGE
    # Generic gauges for percents
    - pattern: kafka.(\w+)<type=(.+), name=(.+)Percent\w*><>Value
      name: kafka_$1_$2_$3_percent
      type: GAUGE
    - pattern: kafka.(\w+)<type=(.+), name=(.+)Percent\w*, (.+)=(.+)><>Value
      name: kafka_$1_$2_$3_percent
      type: GAUGE
      labels:
        "$4": "$5"
    # Generic per-second counters with 0-2 key/value pairs
    - pattern: kafka.(\w+)<type=(.+), name=(.+)PerSec\w*, (.+)=(.+), (.+)=(.+)><>Count
      name: kafka_$1_$2_$3_total
      type: COUNTER
      labels:
        "$4": "$5"
        "$6": "$7"
    - pattern: kafka.(\w+)<type=(.+), name=(.+)PerSec\w*, (.+)=(.+)><>Count
      name: kafka_$1_$2_$3_total
      type: COUNTER
      labels:
        "$4": "$5"
    - pattern: kafka.(\w+)<type=(.+), name=(.+)PerSec\w*><>Count
      name: kafka_$1_$2_$3_total
      type: COUNTER
    # Generic gauges with 0-2 key/value pairs
    - pattern: kafka.(\w+)<type=(.+), name=(.+), (.+)=(.+), (.+)=(.+)><>Value
      name: kafka_$1_$2_$3
      type: GAUGE
      labels:
        "$4": "$5"
        "$6": "$7"
    - pattern: kafka.(\w+)<type=(.+), name=(.+), (.+)=(.+)><>Value
      name: kafka_$1_$2_$3
      type: GAUGE
      labels:
        "$4": "$5"
    - pattern: kafka.(\w+)<type=(.+), name=(.+)><>Value
      name: kafka_$1_$2_$3
      type: GAUGE
    # Emulate Prometheus 'Summary' metrics for the exported 'Histogram's.
    # Note that these are missing the '_sum' metric!
    - pattern: kafka.(\w+)<type=(.+), name=(.+), (.+)=(.+), (.+)=(.+)><>Count
      name: kafka_$1_$2_$3_count
      type: COUNTER
      labels:
        "$4": "$5"
        "$6": "$7"
    - pattern: kafka.(\w+)<type=(.+), name=(.+), (.+)=(.*), (.+)=(.+)><>(\d+)thPercentile
      name: kafka_$1_$2_$3
      type: GAUGE
      labels:
        "$4": "$5"
        "$6": "$7"
        quantile: "0.$8"
    - pattern: kafka.(\w+)<type=(.+), name=(.+), (.+)=(.+)><>Count
      name: kafka_$1_$2_$3_count
      type: COUNTER
      labels:
        "$4": "$5"
    - pattern: kafka.(\w+)<type=(.+), name=(.+), (.+)=(.*)><>(\d+)thPercentile
      name: kafka_$1_$2_$3
      type: GAUGE
      labels:
        "$4": "$5"
        quantile: "0.$6"
    - pattern: kafka.(\w+)<type=(.+), name=(.+)><>Count
      name: kafka_$1_$2_$3_count
      type: COUNTER
    - pattern: kafka.(\w+)<type=(.+), name=(.+)><>(\d+)thPercentile
      name: kafka_$1_$2_$3
      type: GAUGE
      labels:
        quantile: "0.$4"
  zookeeper-metrics-config.yml: |
    # See https://github.com/prometheus/jmx_exporter for more info about JMX Prometheus Exporter metrics
    lowercaseOutputName: true
    rules:
    # replicated Zookeeper
    - pattern: "org.apache.ZooKeeperService<name0=ReplicatedServer_id(\\d+)><>(\\w+)"
      name: "zookeeper_$2"
      type: GAUGE
    - pattern: "org.apache.ZooKeeperService<name0=ReplicatedServer_id(\\d+), name1=replica.(\\d+)><>(\\w+)"
      name: "zookeeper_$3"
      type: GAUGE
      labels:
        replicaId: "$2"
    - pattern: "org.apache.ZooKeeperService<name0=ReplicatedServer_id(\\d+), name1=replica.(\\d+), name2=(\\w+)><>(Packets\\w+)"
      name: "zookeeper_$4"
      type: COUNTER
      labels:
        replicaId: "$2"
        memberType: "$3"
    - pattern: "org.apache.ZooKeeperService<name0=ReplicatedServer_id(\\d+), name1=replica.(\\d+), name2=(\\w+)><>(\\w+)"
      name: "zookeeper_$4"
      type: GAUGE
      labels:
        replicaId: "$2"
        memberType: "$3"
    - pattern: "org.apache.ZooKeeperService<name0=ReplicatedServer_id(\\d+), name1=replica.(\\d+), name2=(\\w+), name3=(\\w+)><>(\\w+)"
      name: "zookeeper_$4_$5"
      type: GAUGE
      labels:
        replicaId: "$2"
        memberType: "$3"

Here is the KafkaMirrorMaker2 yaml:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: my-mirror-maker-2
spec:
  version: 3.0.0
  replicas: 3
  connectCluster: "my-target-cluster-east"
  clusters:
  - alias: "my-source-cluster-west"
    bootstrapServers: versa-kafka-gke-w-kafka-bootstrap:9092
  - alias: "my-target-cluster-east"
    bootstrapServers: versa-kafka-gke-e-kafka-bootstrap:9092
    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-west"
    targetCluster: "my-target-cluster-east"
    sourceConnector:
      config:
        replication.factor: 3
        offset-syncs.topic.replication.factor: 3
        sync.topic.acls.enabled: "true"
    heartbeatConnector:
      config:
        heartbeats.topic.replication.factor: 3
    checkpointConnector:
      config:
        checkpoints.topic.replication.factor: 3
    topicsPattern: ".*"
    groupsPattern: ".*"

Error i see when i check the logs :

command :
kc logs -f pod/my-mirror-maker-2-mirrormaker2-b6967bc5c-gmz5l -n kafka

Error:

 (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2023-02-21 07:21:58,061 WARN Couldn't resolve server versa-kafka-gke-e-kafka-bootstrap:9092 from bootstrap.servers as DNS resolution failed for versa-kafka-gke-e-kafka-bootstrap (org.apache.kafka.clients.ClientUtils) [main]
2023-02-21 07:21:58,062 ERROR Stopping due to error (org.apache.kafka.connect.cli.ConnectDistributed) [main]
org.apache.kafka.common.KafkaException: Failed to create new KafkaAdminClient
    at org.apache.kafka.clients.admin.KafkaAdminClient.createInternal(KafkaAdminClient.java:527)
    at org.apache.kafka.clients.admin.Admin.create(Admin.java:143)
    at org.apache.kafka.connect.util.ConnectUtils.lookupKafkaClusterId(ConnectUtils.java:50)
    at org.apache.kafka.connect.cli.ConnectDistributed.startConnect(ConnectDistributed.java:97)
    at org.apache.kafka.connect.cli.ConnectDistributed.main(ConnectDistributed.java:80)
Caused by: org.apache.kafka.common.config.ConfigException: No resolvable bootstrap urls given in bootstrap.servers
    at org.apache.kafka.clients.ClientUtils.parseAndValidateAddresses(ClientUtils.java:88)
    at org.apache.kafka.clients.ClientUtils.parseAndValidateAddresses(ClientUtils.java:47)
    at org.apache.kafka.clients.admin.KafkaAdminClient.createInternal(KafkaAdminClient.java:482)
    ... 4 more

I think this is expected, since the target bootstrap server - versa-kafka-gke-e-kafka-bootstrap uses a ClusterIP, so it should not be accessible from the source cluster i.e. versa-kafka-gke-w

I tried changing the target bootstrap server to -> 'loadbalancer-IP':9094, and i get error, pls see logs below :

(org.apache.kafka.clients.admin.AdminClientConfig) [connector-thread-my-source-cluster-west->my-target-cluster-east.MirrorSourceConnector]
2023-02-21 07:26:16,635 INFO Finished creating connector my-source-cluster-west->my-target-cluster-east.MirrorSourceConnector (org.apache.kafka.connect.runtime.Worker) [DistributedHerder-connect-1-1]
2023-02-21 07:26:16,647 INFO Kafka version: 3.0.0 (org.apache.kafka.common.utils.AppInfoParser) [connector-thread-my-source-cluster-west->my-target-cluster-east.MirrorSourceConnector]
2023-02-21 07:26:16,648 INFO Kafka commitId: 8cb0a5e9d3441962 (org.apache.kafka.common.utils.AppInfoParser) [connector-thread-my-source-cluster-west->my-target-cluster-east.MirrorSourceConnector]
2023-02-21 07:26:16,649 INFO Kafka startTimeMs: 1676964376647 (org.apache.kafka.common.utils.AppInfoParser) [connector-thread-my-source-cluster-west->my-target-cluster-east.MirrorSourceConnector]
2023-02-21 07:26:16,650 INFO AdminClientConfig values: 
    bootstrap.servers = [loadbalancer_ip:9094]
    client.dns.lookup = use_all_dns_ips
    client.id = 
    connections.max.idle.ms = 300000
    default.api.timeout.ms = 60000
    metadata.max.age.ms = 300000
    metric.reporters = []
    metrics.num.samples = 2
    metrics.recording.level = INFO
    metrics.sample.window.ms = 30000
    receive.buffer.bytes = 65536
    reconnect.backoff.max.ms = 1000
    reconnect.backoff.ms = 50
    request.timeout.ms = 30000
    retries = 2147483647
    retry.backoff.ms = 100
    sasl.client.callback.handler.class = null
    sasl.jaas.config = null
    sasl.kerberos.kinit.cmd = /usr/bin/kinit
    sasl.kerberos.min.time.before.relogin = 60000
    sasl.kerberos.service.name = null
    sasl.kerberos.ticket.renew.jitter = 0.05
    sasl.kerberos.ticket.renew.window.factor = 0.8
    sasl.login.callback.handler.class = null
    sasl.login.class = null
    sasl.login.refresh.buffer.seconds = 300
    sasl.login.refresh.min.period.seconds = 60
    sasl.login.refresh.window.factor = 0.8
    sasl.login.refresh.window.jitter = 0.05
    sasl.mechanism = GSSAPI
    security.protocol = PLAINTEXT
    security.providers = null
    send.buffer.bytes = 131072
    socket.connection.setup.timeout.max.ms = 30000
    socket.connection.setup.timeout.ms = 10000
    ssl.cipher.suites = null
    ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
    ssl.endpoint.identification.algorithm = https
    ssl.engine.factory.class = null
    ssl.key.password = null
    ssl.keymanager.algorithm = SunX509
    ssl.keystore.certificate.chain = null
    ssl.keystore.key = null
    ssl.keystore.location = null
    ssl.keystore.password = null
    ssl.keystore.type = JKS
    ssl.protocol = TLSv1.3
    ssl.provider = null
    ssl.secure.random.implementation = null
    ssl.trustmanager.algorithm = PKIX
    ssl.truststore.certificates = null
    ssl.truststore.location = null
    ssl.truststore.password = null
    ssl.truststore.type = JKS
 (org.apache.kafka.clients.admin.AdminClientConfig) [connector-thread-my-source-cluster-west->my-target-cluster-east.MirrorSourceConnector]
2023-02-21 07:26:16,654 INFO Kafka version: 3.0.0 (org.apache.kafka.common.utils.AppInfoParser) [connector-thread-my-source-cluster-west->my-target-cluster-east.MirrorSourceConnector]
2023-02-21 07:26:16,654 INFO Kafka commitId: 8cb0a5e9d3441962 (org.apache.kafka.common.utils.AppInfoParser) [connector-thread-my-source-cluster-west->my-target-cluster-east.MirrorSourceConnector]
2023-02-21 07:26:16,654 INFO Kafka startTimeMs: 1676964376654 (org.apache.kafka.common.utils.AppInfoParser) [connector-thread-my-source-cluster-west->my-target-cluster-east.MirrorSourceConnector]
2023-02-21 07:26:16,664 INFO AdminClientConfig values: 
    bootstrap.servers = [versa-kafka-gke-w-kafka-bootstrap:9092]
    client.dns.lookup = use_all_dns_ips
    client.id = 
    connections.max.idle.ms = 300000
    default.api.timeout.ms = 60000
    metadata.max.age.ms = 300000
    metric.reporters = []
    metrics.num.samples = 2
    metrics.recording.level = INFO
    metrics.sample.window.ms = 30000
    receive.buffer.bytes = 65536
    reconnect.backoff.max.ms = 1000
    reconnect.backoff.ms = 50
    request.timeout.ms = 30000
    retries = 2147483647
    retry.backoff.ms = 100
    sasl.client.callback.handler.class = null
    sasl.jaas.config = null
    sasl.kerberos.kinit.cmd = /usr/bin/kinit
    sasl.kerberos.min.time.before.relogin = 60000
    sasl.kerberos.service.name = null
    sasl.kerberos.ticket.renew.jitter = 0.05
    sasl.kerberos.ticket.renew.window.factor = 0.8
    sasl.login.callback.handler.class = null
    sasl.login.class = null
    sasl.login.refresh.buffer.seconds = 300
    sasl.login.refresh.min.period.seconds = 60
    sasl.login.refresh.window.factor = 0.8
    sasl.login.refresh.window.jitter = 0.05
    sasl.mechanism = GSSAPI
    security.protocol = PLAINTEXT
    security.providers = null
    send.buffer.bytes = 131072
    socket.connection.setup.timeout.max.ms = 30000
    socket.connection.setup.timeout.ms = 10000
    ssl.cipher.suites = null
    ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
    ssl.endpoint.identification.algorithm = https
    ssl.engine.factory.class = null
    ssl.key.password = null
    ssl.keymanager.algorithm = SunX509
    ssl.keystore.certificate.chain = null
    ssl.keystore.key = null
    ssl.keystore.location = null
    ssl.keystore.password = null
    ssl.keystore.type = JKS
    ssl.protocol = TLSv1.3
    ssl.provider = null
    ssl.secure.random.implementation = null
    ssl.trustmanager.algorithm = PKIX
    ssl.truststore.certificates = null
    ssl.truststore.location = null
    ssl.truststore.password = null
    ssl.truststore.type = JKS
 (org.apache.kafka.clients.admin.AdminClientConfig) [Scheduler for MirrorSourceConnector-creating upstream offset-syncs topic]
2023-02-21 07:26:16,724 INFO Kafka version: 3.0.0 (org.apache.kafka.common.utils.AppInfoParser) [Scheduler for MirrorSourceConnector-creating upstream offset-syncs topic]
2023-02-21 07:26:16,724 INFO Kafka commitId: 8cb0a5e9d3441962 (org.apache.kafka.common.utils.AppInfoParser) [Scheduler for MirrorSourceConnector-creating upstream offset-syncs topic]
2023-02-21 07:26:16,724 INFO Kafka startTimeMs: 1676964376724 (org.apache.kafka.common.utils.AppInfoParser) [Scheduler for MirrorSourceConnector-creating upstream offset-syncs topic]
2023-02-21 07:26:16,788 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Connector my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector config updated (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [KafkaBasedLog Work Thread - mirrormaker2-cluster-configs]
2023-02-21 07:26:16,817 INFO App info kafka.admin.client for adminclient-26 unregistered (org.apache.kafka.common.utils.AppInfoParser) [kafka-admin-client-thread | adminclient-26]
2023-02-21 07:26:16,823 INFO Metrics scheduler closed (org.apache.kafka.common.metrics.Metrics) [kafka-admin-client-thread | adminclient-26]
2023-02-21 07:26:16,823 INFO Closing reporter org.apache.kafka.common.metrics.JmxReporter (org.apache.kafka.common.metrics.Metrics) [kafka-admin-client-thread | adminclient-26]
2023-02-21 07:26:16,823 INFO Metrics reporters closed (org.apache.kafka.common.metrics.Metrics) [kafka-admin-client-thread | adminclient-26]
2023-02-21 07:26:16,823 INFO creating upstream offset-syncs topic took 161 ms (org.apache.kafka.connect.mirror.Scheduler) [Scheduler for MirrorSourceConnector-creating upstream offset-syncs topic]
2023-02-21 07:26:17,283 INFO loading initial set of topic-partitions took 460 ms (org.apache.kafka.connect.mirror.Scheduler) [Scheduler for MirrorSourceConnector-loading initial set of topic-partitions]
2023-02-21 07:26:17,284 INFO creating downstream topic-partitions took 0 ms (org.apache.kafka.connect.mirror.Scheduler) [Scheduler for MirrorSourceConnector-creating downstream topic-partitions]
2023-02-21 07:26:17,311 WARN [Consumer clientId=consumer-mirrormaker2-cluster-2, groupId=mirrormaker2-cluster] Connection to node 1 (/34.23.176.129:9094) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient) [KafkaBasedLog Work Thread - mirrormaker2-cluster-status]
2023-02-21 07:26:17,311 INFO [Consumer clientId=consumer-mirrormaker2-cluster-2, groupId=mirrormaker2-cluster] Error sending fetch request (sessionId=2007621533, epoch=INITIAL) to node 1: (org.apache.kafka.clients.FetchSessionHandler) [KafkaBasedLog Work Thread - mirrormaker2-cluster-status]
org.apache.kafka.common.errors.DisconnectException
2023-02-21 07:26:17,397 INFO WorkerSourceTask{id=my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask) [SourceTaskOffsetCommitter-1]
2023-02-21 07:26:17,465 WARN [Producer clientId=producer-1] Got error produce response with correlation id 1438 on topic-partition mirrormaker2-cluster-offsets-22, retrying (2147483646 attempts left). Error: NETWORK_EXCEPTION. Error Message: Disconnected from node 1 (org.apache.kafka.clients.producer.internals.Sender) [kafka-producer-network-thread | producer-1]
2023-02-21 07:26:17,466 WARN [Producer clientId=producer-1] Received invalid metadata error in produce request on partition mirrormaker2-cluster-offsets-22 due to org.apache.kafka.common.errors.NetworkException: Disconnected from node 1. Going to request metadata update now (org.apache.kafka.clients.producer.internals.Sender) [kafka-producer-network-thread | producer-1]
2023-02-21 07:26:17,485 INFO refreshing known target topics took 201 ms (org.apache.kafka.connect.mirror.Scheduler) [Scheduler for MirrorSourceConnector-refreshing known target topics]
2023-02-21 07:26:17,487 INFO Started my-source-cluster-west->my-target-cluster-east.MirrorSourceConnector with 3 topic-partitions. (org.apache.kafka.connect.mirror.MirrorSourceConnector) [connector-thread-my-source-cluster-west->my-target-cluster-east.MirrorSourceConnector]
2023-02-21 07:26:17,487 INFO Starting my-source-cluster-west->my-target-cluster-east.MirrorSourceConnector took 857 ms. (org.apache.kafka.connect.mirror.MirrorSourceConnector) [connector-thread-my-source-cluster-west->my-target-cluster-east.MirrorSourceConnector]
2023-02-21 07:26:17,488 INFO SourceConnectorConfig values: 
    config.action.reload = restart
    connector.class = org.apache.kafka.connect.mirror.MirrorSourceConnector
    errors.log.enable = false
    errors.log.include.messages = false
    errors.retry.delay.max.ms = 60000
    errors.retry.timeout = 0
    errors.tolerance = none
    header.converter = null
    key.converter = null
    name = my-source-cluster-west->my-target-cluster-east.MirrorSourceConnector
    predicates = []
    tasks.max = 1
    topic.creation.groups = []
    transforms = []
    value.converter = null
 (org.apache.kafka.connect.runtime.SourceConnectorConfig) [DistributedHerder-connect-1-1]
2023-02-21 07:26:17,489 INFO EnrichedConnectorConfig values: 
    config.action.reload = restart
    connector.class = org.apache.kafka.connect.mirror.MirrorSourceConnector
    errors.log.enable = false
    errors.log.include.messages = false
    errors.retry.delay.max.ms = 60000
    errors.retry.timeout = 0
    errors.tolerance = none
    header.converter = null
    key.converter = null
    name = my-source-cluster-west->my-target-cluster-east.MirrorSourceConnector
    predicates = []
    tasks.max = 1
    topic.creation.groups = []
    transforms = []
    value.converter = null
 (org.apache.kafka.connect.runtime.ConnectorConfig$EnrichedConnectorConfig) [DistributedHerder-connect-1-1]
2023-02-21 07:26:17,518 ERROR Scheduler for MirrorSourceConnector caught exception in scheduled task: syncing topic ACLs (org.apache.kafka.connect.mirror.Scheduler) [Scheduler for MirrorSourceConnector-syncing topic ACLs]
java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.SecurityDisabledException: No Authorizer is configured on the broker
    at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:395)
    at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1999)
    at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:165)
    at org.apache.kafka.connect.mirror.MirrorSourceConnector.listTopicAclBindings(MirrorSourceConnector.java:397)
    at org.apache.kafka.connect.mirror.MirrorSourceConnector.syncTopicAcls(MirrorSourceConnector.java:288)
    at org.apache.kafka.connect.mirror.Scheduler.run(Scheduler.java:93)
    at org.apache.kafka.connect.mirror.Scheduler.executeThread(Scheduler.java:112)
    at org.apache.kafka.connect.mirror.Scheduler.lambda$scheduleRepeating$0(Scheduler.java:50)
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
    at java.base/java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305)
    at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: org.apache.kafka.common.errors.SecurityDisabledException: No Authorizer is configured on the broker
2023-02-21 07:26:17,530 INFO syncing topic configs took 10 ms (org.apache.kafka.connect.mirror.Scheduler) [Scheduler for MirrorSourceConnector-syncing topic configs]
2023-02-21 07:26:17,623 WARN [Consumer clientId=consumer-mirrormaker2-cluster-1, groupId=mirrormaker2-cluster] Connection to node 1 (/node_ip:9094) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient) [KafkaBasedLog Work Thread - mirrormaker2-cluster-offsets]
2023-02-21 07:26:17,626 INFO [Consumer clientId=consumer-mirrormaker2-cluster-1, groupId=mirrormaker2-cluster] Error sending fetch request (sessionId=351526999, epoch=INITIAL) to node 1: (org.apache.kafka.clients.FetchSessionHandler) [KafkaBasedLog Work Thread - mirrormaker2-cluster-offsets]
org.apache.kafka.common.errors.DisconnectException
2023-02-21 07:26:17,632 WARN [Producer clientId=producer-1] Connection to node 1 (/node_ip:9094) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient) [kafka-producer-network-thread | producer-1]
2023-02-21 07:26:18,351 WARN [Consumer clientId=consumer-mirrormaker2-cluster-2, groupId=mirrormaker2-cluster] Connection to node 1 (/node_ip:9094) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient) [KafkaBasedLog Work Thread - mirrormaker2-cluster-status]
2023-02-21 07:26:18,352 INFO [Consumer clientId=consumer-mirrormaker2-cluster-2, groupId=mirrormaker2-cluster] Error sending fetch request (sessionId=2007621533, epoch=INITIAL) to node 1: (org.apache.kafka.clients.FetchSessionHandler) [KafkaBasedLog Work Thread - mirrormaker2-cluster-status]
org.apache.kafka.common.errors.DisconnectException
2023-02-21 07:26:18,757 WARN [Consumer clientId=consumer-mirrormaker2-cluster-1, groupId=mirrormaker2-cluster] Connection to node 1 (/node_ip:9094) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient) [KafkaBasedLog Work Thread - mirrormaker2-cluster-offsets]
2023-02-21 07:26:18,757 INFO [Consumer clientId=consumer-mirrormaker2-cluster-1, groupId=mirrormaker2-cluster] Error sending fetch request (sessionId=351526999, epoch=INITIAL) to node 1: (org.apache.kafka.clients.FetchSessionHandler) [KafkaBasedLog Work Thread - mirrormaker2-cluster-offsets]
org.apache.kafka.common.errors.DisconnectException
2023-02-21 07:26:19,749 WARN [Consumer clientId=consumer-mirrormaker2-cluster-1, groupId=mirrormaker2-cluster] Connection to node 1 (/node_ip:9094) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient) [KafkaBasedLog Work Thread - mirrormaker2-cluster-offsets]
2023-02-21 07:26:19,749 INFO [Consumer clientId=consumer-mirrormaker2-cluster-1, groupId=mirrormaker2-cluster] Error sending fetch request (sessionId=351526999, epoch=INITIAL) to node 1: (org.apache.kafka.clients.FetchSessionHandler) [KafkaBasedLog Work Thread - mirrormaker2-cluster-offsets]
org.apache.kafka.common.errors.DisconnectException
2023-02-21 07:26:19,947 WARN [Consumer clientId=consumer-mirrormaker2-cluster-2, groupId=mirrormaker2-cluster] Connection to node 1 (/node_ip:9094) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient) [KafkaBasedLog Work Thread - mirrormaker2-cluster-status]
2023-02-21 07:26:19,947 INFO [Consumer clientId=consumer-mirrormaker2-cluster-2, groupId=mirrormaker2-cluster] Error sending fetch request (sessionId=2007621533, epoch=INITIAL) to node 1: (org.apache.kafka.clients.FetchSessionHandler) [KafkaBasedLog Work Thread - mirrormaker2-cluster-status]
org.apache.kafka.common.errors.DisconnectException
2023-02-21 07:26:20,983 WARN [Consumer clientId=consumer-mirrormaker2-cluster-1, groupId=mirrormaker2-cluster] Connection to node 1 (/node_ip:9094) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient) [KafkaBasedLog Work Thread - mirrormaker2-cluster-offsets]
2023-02-21 07:26:20,984 INFO [Consumer clientId=consumer-mirrormaker2-cluster-1, groupId=mirrormaker2-cluster] Error sending fetch request (sessionId=351526999, epoch=INITIAL) to node 1: (org.apache.kafka.clients.FetchSessionHandler) [KafkaBasedLog Work Thread - mirrormaker2-cluster-offsets]
org.apache.kafka.common.errors.DisconnectException

From the error, it is clear that KafkaMirrorMaker2 is unable to access the target loadbalancer_IP:9094.

Few questions on the KafkaMirrorMaker2 setup :

1. Currently, kafkamirrormaker2 is setup in the src cluster - i presume that is ok ?
2. I dont have authorization setup in the cluster, is that required for KafkaMirrorMaker2 ? (per my understanding, it is not required)

What do i need to ensure that the target kafka cluster bootstrap server is accessible ?

tia!

[scholzj]

Well, you enabled the ACL rules synchronization. In that case, you have to use ACLs. So you either need to disable it in the MM2 custom resource or add it to your Kafka clusters.

[karanalang]

Well, you enabled the ACL rules synchronization. In that case, you have to use ACLs. So you either need to disable it in the MM2 custom resource or add it to your Kafka clusters.

@scholzj - i was able to resolve this issue by setting sync.topic.acls.enabled: "false" in the KafkaMirrorMaker2 yaml

however, the MirrorMaker is not yet transferring data from the source to target clusters.

Here are the yamls & the logs:

KafkaMirrorMaker2 yaml:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: my-mirror-maker-2
spec:
  version: 3.0.0
  replicas: 3
  connectCluster: "my-target-cluster-east"
  clusters:
  - alias: "my-source-cluster-west"
    bootstrapServers: lb_ip_west:9094
  - alias: "my-target-cluster-east"
    bootstrapServers: lb_ip_east:9094
    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-west"
    targetCluster: "my-target-cluster-east"
    sourceConnector:
      config:
        replication.factor: 3
        offset-syncs.topic.replication.factor: 3
        sync.topic.acls.enabled: "false"
    heartbeatConnector:
      config:
        heartbeats.topic.replication.factor: 3
    checkpointConnector:
      config:
        checkpoints.topic.replication.factor: 3
    topicsPattern: ".*"
    groupsPattern: ".*"

Pls note : I'm using the loadbalancer IP for the source &
When i use the Internal bootstrap url (eg. nossl-w-kafka-bootstrap:9092 & nossl-e-kafka-bootstrap:9092), it is giving me exception - org.apache.kafka.common.errors.DisconnectException

Any ideas why ?

Logs from the mirrormaker service:

2023-02-23 06:41:01,697 INFO MirrorTaskConfig values: 
	admin.timeout.ms = 60000
	checkpoints.topic.replication.factor = 3
	config.action.reload = restart
	config.properties.blacklist = null
	config.properties.exclude = [follower\.replication\.throttled\.replicas, leader\.replication\.throttled\.replicas, message\.timestamp\.difference\.max\.ms, message\.timestamp\.type, unclean\.leader\.election\.enable, min\.insync\.replicas]
	config.property.filter.class = class org.apache.kafka.connect.mirror.DefaultConfigPropertyFilter
	connector.class = org.apache.kafka.connect.mirror.MirrorHeartbeatConnector
	consumer.poll.timeout.ms = 1000
	emit.checkpoints.enabled = true
	emit.checkpoints.interval.seconds = 60
	emit.heartbeats.enabled = true
	emit.heartbeats.interval.seconds = 1
	enabled = true
	errors.log.enable = false
	errors.log.include.messages = false
	errors.retry.delay.max.ms = 60000
	errors.retry.timeout = 0
	errors.tolerance = none
	group.filter.class = class org.apache.kafka.connect.mirror.DefaultGroupFilter
	groups = [.*]
	groups.blacklist = null
	groups.exclude = [console-consumer-.*, connect-.*, __.*]
	header.converter = null
	heartbeats.topic.replication.factor = 3
	key.converter = null
	metric.reporters = null
	name = my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector
	offset-syncs.topic.location = source
	offset-syncs.topic.replication.factor = 3
	offset.lag.max = 100
	predicates = []
	refresh.groups.enabled = true
	refresh.groups.interval.seconds = 600
	refresh.topics.enabled = true
	refresh.topics.interval.seconds = 600
	replication.factor = 2
	replication.policy.class = class org.apache.kafka.connect.mirror.DefaultReplicationPolicy
	replication.policy.separator = .
	sasl.client.callback.handler.class = null
	sasl.jaas.config = null
	sasl.kerberos.kinit.cmd = /usr/bin/kinit
	sasl.kerberos.min.time.before.relogin = 60000
	sasl.kerberos.service.name = null
	sasl.kerberos.ticket.renew.jitter = 0.05
	sasl.kerberos.ticket.renew.window.factor = 0.8
	sasl.login.callback.handler.class = null
	sasl.login.class = null
	sasl.login.refresh.buffer.seconds = 300
	sasl.login.refresh.min.period.seconds = 60
	sasl.login.refresh.window.factor = 0.8
	sasl.login.refresh.window.jitter = 0.05
	sasl.mechanism = GSSAPI
	security.protocol = PLAINTEXT
	source.cluster.alias = my-source-cluster-west
	ssl.cipher.suites = null
	ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
	ssl.endpoint.identification.algorithm = https
	ssl.engine.factory.class = null
	ssl.key.password = null
	ssl.keymanager.algorithm = SunX509
	ssl.keystore.certificate.chain = null
	ssl.keystore.key = null
	ssl.keystore.location = null
	ssl.keystore.password = null
	ssl.keystore.type = JKS
	ssl.protocol = TLSv1.3
	ssl.provider = null
	ssl.secure.random.implementation = null
	ssl.trustmanager.algorithm = PKIX
	ssl.truststore.certificates = null
	ssl.truststore.location = null
	ssl.truststore.password = null
	ssl.truststore.type = JKS
	sync.group.offsets.enabled = false
	sync.group.offsets.interval.seconds = 60
	sync.topic.acls.enabled = true
	sync.topic.acls.interval.seconds = 600
	sync.topic.configs.enabled = true
	sync.topic.configs.interval.seconds = 600
	target.cluster.alias = my-target-cluster-east
	task.assigned.groups = null
	task.assigned.partitions = null
	tasks.max = 1
	topic.filter.class = class org.apache.kafka.connect.mirror.DefaultTopicFilter
	topics = [.*]
	topics.blacklist = null
	topics.exclude = [.*[\-\.]internal, .*\.replica, __.*]
	transforms = []
	value.converter = null
 (org.apache.kafka.connect.mirror.MirrorTaskConfig) [task-thread-my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector-0]
2023-02-23 06:41:01,699 INFO WorkerSourceTask{id=my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector-0} Source task finished initialization and start (org.apache.kafka.connect.runtime.WorkerSourceTask) [task-thread-my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector-0]
2023-02-23 06:41:01,823 INFO [Producer clientId=connector-producer-my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector-0] Cluster ID: Hx_hcGamRHWhBWCFsMV2Rg (org.apache.kafka.clients.Metadata) [kafka-producer-network-thread | connector-producer-my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector-0]
2023-02-23 06:41:37,562 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Rebalance started (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator) [DistributedHerder-connect-1-1]
2023-02-23 06:41:37,562 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] (Re-)joining group (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator) [DistributedHerder-connect-1-1]
2023-02-23 06:41:37,656 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Successfully joined group with generation Generation{generationId=4, memberId='connect-1-d19859d9-0bb0-4885-ad5e-b83c97186ba7', protocol='sessioned'} (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator) [DistributedHerder-connect-1-1]
2023-02-23 06:41:37,733 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Successfully synced group in generation Generation{generationId=4, memberId='connect-1-d19859d9-0bb0-4885-ad5e-b83c97186ba7', protocol='sessioned'} (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator) [kafka-coordinator-heartbeat-thread | mirrormaker2-cluster]
2023-02-23 06:41:37,733 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Joined group at generation 4 with protocol version 2 and got assignment: Assignment{error=0, leader='connect-1-d19859d9-0bb0-4885-ad5e-b83c97186ba7', leaderUrl='http://10.98.129.133:8083/', offset=4, connectorIds=[my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector], taskIds=[my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector-0], revokedConnectorIds=[], revokedTaskIds=[], delay=0} with rebalance delay: 0 (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [DistributedHerder-connect-1-1]
2023-02-23 06:41:37,734 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Starting connectors and tasks using config offset 4 (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [DistributedHerder-connect-1-1]
2023-02-23 06:41:37,734 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Finished starting connectors and tasks (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [DistributedHerder-connect-1-1]
2023-02-23 06:42:01,689 INFO WorkerSourceTask{id=my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask) [SourceTaskOffsetCommitter-1]
2023-02-23 06:42:52,740 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Rebalance started (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator) [DistributedHerder-connect-1-1]
2023-02-23 06:42:52,740 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] (Re-)joining group (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator) [DistributedHerder-connect-1-1]
2023-02-23 06:42:52,865 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Successfully joined group with generation Generation{generationId=5, memberId='connect-1-d19859d9-0bb0-4885-ad5e-b83c97186ba7', protocol='sessioned'} (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator) [DistributedHerder-connect-1-1]
2023-02-23 06:42:52,939 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Successfully synced group in generation Generation{generationId=5, memberId='connect-1-d19859d9-0bb0-4885-ad5e-b83c97186ba7', protocol='sessioned'} (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator) [DistributedHerder-connect-1-1]
2023-02-23 06:42:52,940 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Joined group at generation 5 with protocol version 2 and got assignment: Assignment{error=0, leader='connect-1-d19859d9-0bb0-4885-ad5e-b83c97186ba7', leaderUrl='http://10.98.129.133:8083/', offset=4, connectorIds=[my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector], taskIds=[my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector-0], revokedConnectorIds=[], revokedTaskIds=[], delay=0} with rebalance delay: 0 (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [DistributedHerder-connect-1-1]
2023-02-23 06:42:52,940 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Starting connectors and tasks using config offset 4 (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [DistributedHerder-connect-1-1]
2023-02-23 06:42:52,940 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Finished starting connectors and tasks (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [DistributedHerder-connect-1-1]
2023-02-23 06:43:02,097 INFO WorkerSourceTask{id=my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask) [SourceTaskOffsetCommitter-1]
2023-02-23 06:43:23,320 INFO AbstractConfig values: 
 (org.apache.kafka.common.config.AbstractConfig) [pool-2-thread-2]
2023-02-23 06:43:23,523 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Connector my-source-cluster-west->my-target-cluster-east.MirrorCheckpointConnector config updated (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [KafkaBasedLog Work Thread - mirrormaker2-cluster-configs]
2023-02-23 06:43:23,524 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Rebalance started (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator) [DistributedHerder-connect-1-1]
2023-02-23 06:43:23,524 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] (Re-)joining group (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator) [DistributedHerder-connect-1-1]
2023-02-23 06:43:23,594 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Successfully joined group with generation Generation{generationId=6, memberId='connect-1-d19859d9-0bb0-4885-ad5e-b83c97186ba7', protocol='sessioned'} (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator) [DistributedHerder-connect-1-1]
2023-02-23 06:43:23,665 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Successfully synced group in generation Generation{generationId=6, memberId='connect-1-d19859d9-0bb0-4885-ad5e-b83c97186ba7', protocol='sessioned'} (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator) [DistributedHerder-connect-1-1]
2023-02-23 06:43:23,666 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Joined group at generation 6 with protocol version 2 and got assignment: Assignment{error=0, leader='connect-1-d19859d9-0bb0-4885-ad5e-b83c97186ba7', leaderUrl='http://10.98.129.133:8083/', offset=5, connectorIds=[my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector], taskIds=[my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector-0], revokedConnectorIds=[], revokedTaskIds=[], delay=0} with rebalance delay: 0 (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [DistributedHerder-connect-1-1]
2023-02-23 06:43:23,666 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Starting connectors and tasks using config offset 5 (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [DistributedHerder-connect-1-1]
2023-02-23 06:43:23,666 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Finished starting connectors and tasks (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [DistributedHerder-connect-1-1]
2023-02-23 06:43:23,668 INFO AbstractConfig values: 
 (org.apache.kafka.common.config.AbstractConfig) [pool-2-thread-2]
2023-02-23 06:43:23,873 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Connector my-source-cluster-west->my-target-cluster-east.MirrorSourceConnector config updated (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [KafkaBasedLog Work Thread - mirrormaker2-cluster-configs]
2023-02-23 06:43:23,873 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Rebalance started (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator) [DistributedHerder-connect-1-1]
2023-02-23 06:43:23,873 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] (Re-)joining group (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator) [DistributedHerder-connect-1-1]
2023-02-23 06:43:23,942 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Successfully joined group with generation Generation{generationId=7, memberId='connect-1-d19859d9-0bb0-4885-ad5e-b83c97186ba7', protocol='sessioned'} (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator) [DistributedHerder-connect-1-1]
2023-02-23 06:43:24,066 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Successfully synced group in generation Generation{generationId=7, memberId='connect-1-d19859d9-0bb0-4885-ad5e-b83c97186ba7', protocol='sessioned'} (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator) [DistributedHerder-connect-1-1]
2023-02-23 06:43:24,066 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Joined group at generation 7 with protocol version 2 and got assignment: Assignment{error=0, leader='connect-1-d19859d9-0bb0-4885-ad5e-b83c97186ba7', leaderUrl='http://10.98.129.133:8083/', offset=6, connectorIds=[my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector], taskIds=[my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector-0], revokedConnectorIds=[], revokedTaskIds=[], delay=0} with rebalance delay: 0 (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [DistributedHerder-connect-1-1]
2023-02-23 06:43:24,067 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Starting connectors and tasks using config offset 6 (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [DistributedHerder-connect-1-1]
2023-02-23 06:43:24,067 INFO [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Finished starting connectors and tasks (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [DistributedHerder-connect-1-1]
2023-02-23 06:44:02,166 INFO WorkerSourceTask{id=my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask) [SourceTaskOffsetCommitter-1]
2023-02-23 06:45:02,236 INFO WorkerSourceTask{id=my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask) [SourceTaskOffsetCommitter-1]
2023-02-23 06:46:02,305 INFO WorkerSourceTask{id=my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask) [SourceTaskOffsetCommitter-1]
2023-02-23 06:47:02,374 INFO WorkerSourceTask{id=my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask) [SourceTaskOffsetCommitter-1]
2023-02-23 06:48:02,443 INFO WorkerSourceTask{id=my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask) [SourceTaskOffsetCommitter-1]
2023-02-23 06:49:02,511 INFO WorkerSourceTask{id=my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask) [SourceTaskOffsetCommitter-1]
2023-02-23 06:50:02,580 INFO WorkerSourceTask{id=my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask) [SourceTaskOffsetCommitter-1]
2023-02-23 06:51:02,648 INFO WorkerSourceTask{id=my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask) [SourceTaskOffsetCommitter-1]
2023-02-23 06:52:02,716 INFO WorkerSourceTask{id=my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask) [SourceTaskOffsetCommitter-1]
2023-02-23 06:53:02,784 INFO WorkerSourceTask{id=my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask) [SourceTaskOffsetCommitter-1]
2023-02-23 06:54:02,851 INFO WorkerSourceTask{id=my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask) [SourceTaskOffsetCommitter-1]
2023-02-23 06:55:02,919 INFO WorkerSourceTask{id=my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask) [SourceTaskOffsetCommitter-1]
2023-02-23 06:56:02,987 INFO WorkerSourceTask{id=my-source-cluster-west->my-target-cluster-east.MirrorHeartbeatConnector-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask) [SourceTaskOffsetCommitter-1]

This does not seem to be showing any outstanding messages, and the messages I'm pushing into the source cluster (using kafka command-line) are not showing up in the target.

Pls advise on what needs to be done to fix this ?
How do I debug this ?

tia !
Pls note - this is critical functionality for us, hence your help is appreciated

[karanalang]

Another observation --

In the topic 'mirrormaker2-cluster-status', i see many tasks the status of which are shown below :

many have status - 'RUNNING'
{"state":"RUNNING","trace":null,"worker_id":"10.98.128.200:8083","generation":14}

some have status - 'UNASSIGNED'
{"state":"UNASSIGNED","trace":null,"worker_id":"10.98.129.130:8083","generation":18}

[scholzj]

'you enabled the ACL rules synchronization', where is that enabled

This is something you added to the Mirror Maker 2 custom resource: sync.topic.acls.enabled: "true" => so you told it to sync the ACLs. For that, you need to have the ACLs enabled (and to do that, you need to have authentication, set the ACLs for all the compoenents etc.). BTW, that anyway does not work with enabled User Operator. So you will need to disable it as well.

Or you can of course disable the ACL sync again as we have it by default.

[karanalang]

@scholzj - yes, i was able to get past the acl sync issue

Any ideas why the mirrormaker is not working as expected ? i.e. it is not transferring data from source to destination .. the logs (pasted above) don't show any errors !

[scholzj]

I'm not sure I understand what the other issue is as it is confusing what is related to the ACLs and what is not. Maybe you should strt form the beginning about what is the current issue.