Cannot complete request because of a conflicting operation (e.g. worker rebalance)

[areebemumba]

I have deployed Strimzi Operator and Kafka Mirror Maker 2 on the Kubernetes cluster. I have a use case where I want to mirror messages from one topic to another topic. Both the Kafka clusters are accessible from MM2.

Error:

2024-03-27 08:53:11,124 ERROR Request 'PUT /connectors/a_kafka->b_kafka.MirrorCheckpointConnector/config' failed because it couldn't find the target Connect worker within two hops (between workers). (org.apache.kafka.connect.runtime.rest.HerderRequestHandler) [qtp1268701597-20]
2024-03-27 08:53:11,128 INFO 10.68.151.126 - - [27/Mar/2024:08:53:11 +0000] "PUT /connectors/a_kafka-%b_kafka.MirrorCheckpointConnector/config?forward=false HTTP/1.1" 409 113 "-" "kafka-connect" 15 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp1268701597-20]
2024-03-27 08:53:11,128 ERROR Error forwarding REST request (org.apache.kafka.connect.runtime.rest.RestClient) [qtp1268701597-21]
org.apache.kafka.connect.runtime.rest.errors.ConnectRestException: Cannot complete request because of a conflicting operation (e.g. worker rebalance)
	at org.apache.kafka.connect.runtime.rest.RestClient.httpRequest(RestClient.java:162)
	at org.apache.kafka.connect.runtime.rest.RestClient.httpRequest(RestClient.java:116)
	at org.apache.kafka.connect.runtime.rest.RestClient.httpRequest(RestClient.java:80)
	at org.apache.kafka.connect.runtime.rest.HerderRequestHandler.completeOrForwardRequest(HerderRequestHandler.java:112)
	at org.apache.kafka.connect.runtime.rest.HerderRequestHandler.completeOrForwardRequest(HerderRequestHandler.java:129)
	at org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource.putConnectorConfig(ConnectorsResource.java:230)

Helm template:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: my-mirror-maker-2
  namespace: kafka
spec:
  version: 3.6.1
  replicas: 1
  connectCluster: "b_kafka"
  clusters:
  - alias: "a_kafka" # Source cluster
    bootstrapServers: {{ .Values.a_kafka }}
  - alias: "b_kafka" # Target cluster
    bootstrapServers: {{ .Values.b_kafka }}
    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: "a_kafka"
    targetCluster: "b_kafka"
    sourceConnector:
      tasksMax: 1
      config:
        # -1 means it will use the default replication factor configured in the broker
        replication.factor: -1
        offset-syncs.topic.replication.factor: -1
        sync.topic.acls.enabled: "false"
        replication.policy.class: "org.apache.kafka.connect.mirror.IdentityReplicationPolicy"
        refresh.topics.interval.seconds: 600
    checkpointConnector:
      tasksMax: 1
      config:
        # -1 means it will use the default replication factor configured in the broker
        checkpoints.topic.replication.factor: -1
        sync.group.offsets.enabled: "false"
        replication.policy.class: "org.apache.kafka.connect.mirror.IdentityReplicationPolicy"
        refresh.groups.interval.seconds: 600
    topicsPattern: {{ .Values.topicsPattern }}
    groupsPattern: {{ .Values.groupsPattern }}