Error in configuring MirrorMaker2 between two external clusters on separate VMs with TLS enabled

[SaifulHasan3000]

I am trying to establish tls connection between two kafka clusters deployed on two different VMs using kafka Mirrormaker2.
I have made the following deployments:

VM1 (Source kafka cluster) (10.195.29.2)
Kafka-source-cluster.yaml:

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: kafka-source
  namespace: kafka
spec:
  kafka:
    version: 3.9.0
    replicas: 3
    listeners:
      - name: plain
        port: 9092
        tls: false
        type: nodeport
      - authentication:
          type: tls
        name: tls
        port: 9093
        tls: true
        type: nodeport
    config:
      offsets.topic.replication.factor: 2
      transaction.state.log.replication.factor: 2
      transaction.state.log.min.isr: 2
      log.message.format.version: "2.8"
      auto.create.topics.enable: "true"
    storage:
      type: jbod
      volumes:
      - id: 0
        type: persistent-claim
        size: 1Gi
        deleteClaim: false
    template:
      pod:
        affinity:
          podAntiAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              - labelSelector:
                  matchExpressions:
                    - key: strimzi.io/name
                      operator: In
                      values:
                        - kafka-source
                topologyKey: "kubernetes.io/hostname"
  zookeeper:
    replicas: 3
    storage:
      type: persistent-claim
      size: 10Gi
      deleteClaim: false
  entityOperator:
    topicOperator: {}
    userOperator: {}

VM2 (Target kafka cluster) (10.195.18.193)
Kafka-target-cluster.yaml:
Same as above, just replace source by target

Kafka-tls-mirrormaker2.yaml: (deployed on the target VM i.e. VM2)

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: my-mirror-maker-2
  namespace: kafka
spec:
  version: 3.9.0
  replicas: 1
  connectCluster: "kafka-target" # Must be the target custer
  clusters:
  - alias: "kafka-source" # Source cluster
    bootstrapServers: 10.195.29.2:30909
    tls:
      trustedCertificates:
        - secretName: kafka-source-cluster-ca-cert
          pattern: "ca.crt"
    config:
      ssl.endpoint.identification.algorithm: ""
  - alias: "kafka-target" # Target cluster
    bootstrapServers: kafka-target-kafka-tls-bootstrap:9093
    tls:
      trustedCertificates:
        - secretName: kafka-target-cluster-ca-cert
          pattern: "ca.crt"
    config:
      ssl.endpoint.identification.algorithm: ""
      config.storage.replication.factor: 1
      offset.storage.replication.factor: 1
      status.storage.replication.factor: 1
  mirrors:
  - sourceCluster: "kafka-source"
    targetCluster: "kafka-target"
    sourceConnector:
      config:
        replication.factor: 1
        offset-syncs.topic.replication.factor: 1
        sync.topic.acls.enabled: "false"
        replication.policy.separator: ""
        replication.policy.class: "io.strimzi.kafka.connect.mirror.IdentityReplicationPolicy"
    heartbeatConnector:
      config:
        heartbeats.topic.replication.factor: 1
    checkpointConnector:
      config:
        checkpoints.topic.replication.factor: 1
        replication.policy.separator: ""
        replication.policy.class: "io.strimzi.kafka.connect.mirror.IdentityReplicationPolicy"
    topicsPattern: "topic-source"
    groupsPattern: " .* "
  logging:
    type: inline
    loggers:
      connect.root.logger.level: "INFO"

The port 30909 on which tls port of source cluster is forwarded.
For the secret: kafka-source-cluster-ca-cert, I have extracted the yaml file of this secret and used this yaml file to deploy the secret on the VM2 (since mirrormaker requires both the certificates as secrets in the same namespace of the VM2 in which it is deployed.)

The source-secret.yaml file used to deploy the kafka-source-cluster-ca-cert onto kafka namespace of VM2 looks like this:
(I've commented the ownerReference block as I was not able to deploy the secret on the Kafka cluster 2 with it in place, so thought that its unnecessary so removed it.)

apiVersion: v1
data:
  ca.crt: #the encoded cert here (I've not copied that here to save space0
  ca.p12:  #the .p12 related to the cert (I've not copied that here to save space0
  ca.password: UGxVU3YyUEJnTXhn
kind: Secret
metadata:
  annotations:
    strimzi.io/ca-cert-generation: "0"
  creationTimestamp: "2025-01-23T07:51:55Z"
  labels:
    app.kubernetes.io/instance: kafka-source
    app.kubernetes.io/managed-by: strimzi-cluster-operator
    app.kubernetes.io/name: certificate-authority
    app.kubernetes.io/part-of: strimzi-kafka-source
    strimzi.io/cluster: kafka-source
    strimzi.io/component-type: certificate-authority
    strimzi.io/kind: Kafka
    strimzi.io/name: strimzi
  name: kafka-source-cluster-ca-cert
  namespace: kafka
  # ownerReferences:
  # - apiVersion: kafka.strimzi.io/v1beta2
  #   blockOwnerDeletion: false
  #   controller: false
  #   kind: Kafka
  #   name: kafka-source
  #   uid: 78fd54f9-4fe7-475d-97fd-8f2c159469e3
  # resourceVersion: "61295984"
  # uid: 18257c39-461d-46d2-b1c7-939614eb0d84
type: Opaque

I have followed this: #2436 and have included he ssl.endpoint.identification.algorithm: "" in the config too, after that when all the things are deployed I am getting the following error in the logs of the deployed mirrormaker2:

org.apache.kafka.common.errors.TimeoutException: The AdminClient thread has exited. Call: fetchMetadata
2025-01-23 14:58:37,261 INFO [AdminClient clientId=adminclient-1] Timed out 2 remaining operation(s) during close. (org.apache.kafka.clients.admin.KafkaAdminClient) [kafka-admin-client-thread | adminclient-1]
2025-01-23 14:58:37,266 INFO Metrics scheduler closed (org.apache.kafka.common.metrics.Metrics) [kafka-admin-client-thread | adminclient-1]
2025-01-23 14:58:37,266 INFO Closing reporter org.apache.kafka.common.metrics.JmxReporter (org.apache.kafka.common.metrics.Metrics) [kafka-admin-client-thread | adminclient-1]
2025-01-23 14:58:37,266 INFO Metrics reporters closed (org.apache.kafka.common.metrics.Metrics) [kafka-admin-client-thread | adminclient-1]
2025-01-23 14:58:37,266 ERROR Stopping due to error (org.apache.kafka.connect.cli.AbstractConnectCli) [main]
org.apache.kafka.connect.errors.ConnectException: Failed to connect to and describe Kafka cluster. Check worker's broker connection and security properties.
        at org.apache.kafka.connect.runtime.WorkerConfig.lookupKafkaClusterId(WorkerConfig.java:303)
        at org.apache.kafka.connect.runtime.WorkerConfig.lookupKafkaClusterId(WorkerConfig.java:283)
        at org.apache.kafka.connect.runtime.WorkerConfig.kafkaClusterId(WorkerConfig.java:413)
        at org.apache.kafka.connect.cli.AbstractConnectCli.startConnect(AbstractConnectCli.java:124)
        at org.apache.kafka.connect.cli.AbstractConnectCli.run(AbstractConnectCli.java:95)
        at org.apache.kafka.connect.cli.ConnectDistributed.main(ConnectDistributed.java:112)
Caused by: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.SslAuthenticationException: Failed to process post-handshake messages
        at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:396)
        at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2073)
        at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:165)
        at org.apache.kafka.connect.runtime.WorkerConfig.lookupKafkaClusterId(WorkerConfig.java:297)
        ... 5 more
Caused by: org.apache.kafka.common.errors.SslAuthenticationException: Failed to process post-handshake messages
Caused by: javax.net.ssl.SSLHandshakeException: Received fatal alert: bad_certificate
        at java.base/sun.security.ssl.Alert.createSSLException(Alert.java:131)
        at java.base/sun.security.ssl.Alert.createSSLException(Alert.java:117)
        at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:370)
        at java.base/sun.security.ssl.Alert$AlertConsumer.consume(Alert.java:293)
        at java.base/sun.security.ssl.TransportContext.dispatch(TransportContext.java:209)
        at java.base/sun.security.ssl.SSLTransport.decode(SSLTransport.java:172)
        at java.base/sun.security.ssl.SSLEngineImpl.decode(SSLEngineImpl.java:736)
        at java.base/sun.security.ssl.SSLEngineImpl.readRecord(SSLEngineImpl.java:691)
        at java.base/sun.security.ssl.SSLEngineImpl.unwrap(SSLEngineImpl.java:506)
        at java.base/sun.security.ssl.SSLEngineImpl.unwrap(SSLEngineImpl.java:482)
        at java.base/javax.net.ssl.SSLEngine.unwrap(SSLEngine.java:679)
        at org.apache.kafka.common.network.SslTransportLayer.read(SslTransportLayer.java:586)
        at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:85)
        at org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:462)
        at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:412)
        at org.apache.kafka.common.network.Selector.attemptRead(Selector.java:694)
        at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:596)
        at org.apache.kafka.common.network.Selector.poll(Selector.java:501)
        at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:596)
        at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.processRequests(KafkaAdminClient.java:1542)
        at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1473)
        at java.base/java.lang.Thread.run(Thread.java:840)

I tried the whole setup without tls and it was working perfectly, so there is something with tls certs thats causing the issue.
Don’t know what to do, any help is appreciated as it is quite crucial for me!
Kindly reply if any further clarification is required.

@scholzj please help me out with this
Thanks in advance!

[scholzj]

You have TLS authentication configured on the Kafka clusters but not in Mirror Maker 2. So that would be start - to either remove it from the Kafka configuration or add it to the MM2. Please check the docs and examples on GitHub for more details.

[SaifulHasan3000]

hmm, I've removed the tls authentication from both the kafka clusters and redeployed them, copied the new cert secret from the source cluster vm and deployed it on the target cluster vm in the kafka ns. The mirromaker is also redeployed too with the new port number (32000) for the source ip.

authentication:
          type: tls

Now the mirrormaker is getting deployed and is running, and in its logs initially some errors are flashed, but eventually I am getting the following logs:

2025-01-24 09:05:36,788 INFO 10.42.1.72 - - [24/Jan/2025:09:05:36 +0000] "GET /connectors/kafka-source->kafka-target.MirrorHeartbeatConnector/topics HTTP/1.1" 200 81 "-" "-" 2 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp │
│ 376660032-111]                                                                                                                                                                                                                          │
│ 2025-01-24 09:05:36,788 INFO 10.42.1.72 - - [24/Jan/2025:09:05:36 +0000] "GET /connectors/kafka-source->kafka-target.MirrorSourceConnector/topics HTTP/1.1" 200 80 "-" "-" 2 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp376 │
│ 660032-70]                                                                                                                                                                                                                              │
│ 2025-01-24 09:05:38,458 INFO 10.42.2.1 - - [24/Jan/2025:09:05:38 +0000] "GET / HTTP/1.1" 200 91 "-" "kube-probe/1.30" 2 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp376660032-61]                                            │
│ 2025-01-24 09:05:38,458 INFO 10.42.2.1 - - [24/Jan/2025:09:05:38 +0000] "GET / HTTP/1.1" 200 91 "-" "kube-probe/1.30" 2 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp376660032-139]                                           │
│ 2025-01-24 09:05:48,458 INFO 10.42.2.1 - - [24/Jan/2025:09:05:48 +0000] "GET / HTTP/1.1" 200 91 "-" "kube-probe/1.30" 2 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp376660032-144]                                           │
│ 2025-01-24 09:05:48,458 INFO 10.42.2.1 - - [24/Jan/2025:09:05:48 +0000] "GET / HTTP/1.1" 200 91 "-" "kube-probe/1.30" 2 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp376660032-117]      

This indicates to me, that the clusters are connected, but when I am trying to send some message from source cluster using cli based kafka producer pod on the topic: topic-source and try to receive it on the target cluster using a similar consumer pod, I am not receiving anything at the consumer's end.

However, when I deploy the mirrormaker without tls I am able to use the same producer and consumer and it is working fine there.
One more thing to point out, that in this without tls mirrormaker setup, whenever I am sending some messages from the source cluster, I get a log like the one given below:
Logs when a message is sent from source: (it is quite delayed but comes)

 2025-01-24 08:35:22,532 INFO [kafka-source->kafka-target.MirrorSourceConnector|task-0|offsets] WorkerSourceTask{id=kafka-source->kafka-target.MirrorSourceConnector-0} Committing offsets for 1 acknowledged messages (org.apache.kafka │
│ .connect.runtime.WorkerSourceTask) [SourceTaskOffsetCommitter-1]   

But I don't get such a log in the with tls mirrormaker setup.
rather I get some like this, randomly and not periodically:

2025-01-24 08:54:28,460 INFO 10.42.2.1 - - [24/Jan/2025:08:54:28 +0000] "GET / HTTP/1.1" 200 91 "-" "kube-probe/1.30" 4 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp376660032-66]                                            │
│ 2025-01-24 08:54:33,311 INFO [kafka-source->kafka-target.MirrorCheckpointConnector|worker] refreshing consumer groups took 29 ms (org.apache.kafka.connect.mirror.Scheduler) [Scheduler for MirrorCheckpointConnector: kafka-source->ka │
│ fka-target|kafka-source->kafka-target.MirrorCheckpointConnector-refreshing consumer groups]                                                                                                                                             │
│ 2025-01-24 08:54:33,369 INFO [kafka-source->kafka-target.MirrorSourceConnector|worker] syncing topic configs took 32 ms (org.apache.kafka.connect.mirror.Scheduler) [Scheduler for MirrorSourceConnector: kafka-source->kafka-target|ka │
│ fka-source->kafka-target.MirrorSourceConnector-syncing topic configs]                                                                                                                                                                   │
│ 2025-01-24 08:54:33,405 INFO [kafka-source->kafka-target.MirrorSourceConnector|worker] refreshing topics took 36 ms (org.apache.kafka.connect.mirror.Scheduler) [Scheduler for MirrorSourceConnector: kafka-source->kafka-target|kafka- │
│ source->kafka-target.MirrorSourceConnector-refreshing topics]                                                                                                                                                                           │
│ 2025-01-24 08:54:38,459 INFO 10.42.2.1 - - [24/Jan/2025:08:54:38 +0000] "GET / HTTP/1.1" 200 91 "-" "kube-probe/1.30" 3 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp376660032-70]                                            │
│ 2025-01-24 08:54:38,459 INFO 10.42.2.1 - - [24/Jan/2025:08:54:3

[scholzj]

It is hard to say what might be the issue without seeing the updated configs and full logs. The obvious step if everything else seems to work would be to simplify the configuration and e.g. try if removing the topic pattern helps etc.