[Bug]: MirrorMaker2 doesn't copy all the data

[apolienko]

Bug Description

I have 2 different clusters deployed in 2 namespaces. (kafka and kafka1)

I have 1 strimzi-operator with env STRIMZI_NAMESPACE: *

So I had empry second cluster as replica. Then I deployed mirrorMaker.
But now main cluster is 12.13GB and replication that was filled by MM is 99MB

Replica factor is 1 for both clusters.

And I cant see any progress. This is final version of replication.

But then I opened MirrorMaker2 state and found error with max bytes of message.

      name: rdss->rdss1.MirrorSourceConnector
      tasks:
        - id: 0
          state: FAILED
          trace: "org.apache.kafka.connect.errors.ConnectException: Unrecoverable exception from producer send callback\n\tat org.apache.kafka.connect.runtime.WorkerSourceTask.maybeThrowProducerSendException(WorkerSourceTask.java:340)\n\tat org.apache.kafka.connect.runtime.WorkerSourceTask.prepareToSendRecord(WorkerSourceTask.java:133)\n\tat org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.sendRecords(AbstractWorkerSourceTask.java:411)\n\tat org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.execute(AbstractWorkerSourceTask.java:367)\n\tat org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:204)\n\tat org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:259)\n\tat org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.run(AbstractWorkerSourceTask.java:77)\n\tat org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:236)\n\tat java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)\n\tat java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)\n\tat java.base/java.lang.Thread.run(Thread.java:840)\nCaused by: org.apache.kafka.common.errors.RecordTooLargeException: The message is 1243853 bytes when serialized which is larger than 1048576, which is the value of the max.request.size configuration.\n"
          worker_id: >-
            mirror-maker-kafka-cluster-rdss-mirrormaker2-0.mirror-maker-kafka-cluster-rdss-mirrormaker2.kafka1.svc:8083

I didnt find anything in docs where I can specify maximum bytes of message
And can it be the reason why data is not copied?

Steps to reproduce

No response

Expected behavior

Ability to specify max bytes of message that must be replicated

Strimzi version

0.40.0

Kubernetes version

1.30.6-gke.1596000

Installation method

Helm chart

Infrastructure

GKE

Configuration files and logs

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  annotations:
    meta.helm.sh/release-name: rdss
    meta.helm.sh/release-namespace: kafka1
  creationTimestamp: '2025-01-14T19:23:58Z'
  generation: 1
  labels:
    app.kubernetes.io/managed-by: Helm
  name: mirror-maker-kafka-cluster-rdss
  namespace: kafka1
  resourceVersion: '279250110'
  uid: e1baf2c1-3f03-4dca-9246-30570e699409
  selfLink: >-
    /apis/kafka.strimzi.io/v1beta2/namespaces/kafka1/kafkamirrormaker2s/mirror-maker-kafka-cluster-rdss
status:
  conditions:
    - lastTransitionTime: '2025-01-14T19:28:34.658252835Z'
      status: 'True'
      type: Ready
  connectors:
    - connector:
        state: RUNNING
        worker_id: >-
          mirror-maker-kafka-cluster-rdss-mirrormaker2-0.mirror-maker-kafka-cluster-rdss-mirrormaker2.kafka1.svc:8083
      name: rdss->rdss1.MirrorCheckpointConnector
      tasks:
        - id: 0
          state: RUNNING
          worker_id: >-
            mirror-maker-kafka-cluster-rdss-mirrormaker2-0.mirror-maker-kafka-cluster-rdss-mirrormaker2.kafka1.svc:8083
      type: source
    - connector:
        state: RUNNING
        worker_id: >-
          mirror-maker-kafka-cluster-rdss-mirrormaker2-0.mirror-maker-kafka-cluster-rdss-mirrormaker2.kafka1.svc:8083
      name: rdss->rdss1.MirrorSourceConnector
      tasks:
        - id: 0
          state: FAILED
          trace: "org.apache.kafka.connect.errors.ConnectException: Unrecoverable exception from producer send callback\n\tat org.apache.kafka.connect.runtime.WorkerSourceTask.maybeThrowProducerSendException(WorkerSourceTask.java:340)\n\tat org.apache.kafka.connect.runtime.WorkerSourceTask.prepareToSendRecord(WorkerSourceTask.java:133)\n\tat org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.sendRecords(AbstractWorkerSourceTask.java:411)\n\tat org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.execute(AbstractWorkerSourceTask.java:367)\n\tat org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:204)\n\tat org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:259)\n\tat org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.run(AbstractWorkerSourceTask.java:77)\n\tat org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:236)\n\tat java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)\n\tat java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)\n\tat java.base/java.lang.Thread.run(Thread.java:840)\nCaused by: org.apache.kafka.common.errors.RecordTooLargeException: The message is 1243853 bytes when serialized which is larger than 1048576, which is the value of the max.request.size configuration.\n"
          worker_id: >-
            mirror-maker-kafka-cluster-rdss-mirrormaker2-0.mirror-maker-kafka-cluster-rdss-mirrormaker2.kafka1.svc:8083
      type: source
  labelSelector: >-
    strimzi.io/cluster=mirror-maker-kafka-cluster-rdss,strimzi.io/name=mirror-maker-kafka-cluster-rdss-mirrormaker2,strimzi.io/kind=KafkaMirrorMaker2
  observedGeneration: 1
  replicas: 1
  url: http://mirror-maker-kafka-cluster-rdss-mirrormaker2-api.kafka1.svc:8083
spec:
  clusters:
    - alias: rdss
      bootstrapServers: kafka-cluster-rdss-kafka-bootstrap.kafka.svc.cluster.local:9092
    - alias: rdss1
      bootstrapServers: kafka-cluster-rdss-kafka-bootstrap.kafka1.svc.cluster.local:9092
      config:
        config.storage.replication.factor: -1
        offset.storage.replication.factor: -1
        status.storage.replication.factor: -1
  connectCluster: rdss1
  mirrors:
    - checkpointConnector:
        config:
          checkpoints.topic.replication.factor: -1
          refresh.groups.interval.seconds: 600
          replication.policy.class: org.apache.kafka.connect.mirror.IdentityReplicationPolicy
          sync.group.offsets.enabled: 'false'
        tasksMax: 1
      groupsPattern: .*
      sourceCluster: rdss
      sourceConnector:
        config:
          offset-syncs.topic.replication.factor: -1
          refresh.topics.interval.seconds: 600
          replication.factor: -1
          replication.policy.class: org.apache.kafka.connect.mirror.IdentityReplicationPolicy
          sync.topic.acls.enabled: 'false'
        tasksMax: 1
      targetCluster: rdss1
      topicsPattern: .*
  replicas: 1
  version: 3.6.0

Additional context

No response

[scholzj]

The error seems pretty clear:

org.apache.kafka.common.errors.RecordTooLargeException: The message is 1243853 bytes when serialized which is larger than 1048576, which is the value of the max.request.size configuration

You have large messages in your Kafka cluster. That is a bad pattern, and it also does not work with Kafka's default configuration. You would need to configure the Kafka broker and the Mirror Maker to accept messages with larger size than 1048576.

Why are you opening a bug issue for something that is pretty clearly explained in the error you got and is clearly not a Strimzi bug?

[apolienko]

Jakub, sorry for that. Didn't check discussions. Thank you a lot!
But anyway I could set ability to accept messages larger, but would it lead to cluster recreation with this not default property ?

And also it's not clear how to set to mirrorMaker. Here is mentioned as least two ways and none of them is approved

[scholzj]

Jakub, sorry for that. Didn't check discussions. Thank you a lot!

The bug template has pretty clear instructions about where to ask questions and what to use the bug reports for.

But anyway I could set ability to accept messages larger, but would it lead to cluster recreation with this not default property ?

It should not lead to recreation of any cluster - at least ot with Strimzi. If you change the setting in the KafkaMirrorMaker2 and Kafka custom resources, the operator should roll out the change to the operands.

The value is set to 1MB by default because that is Kafka's default. You would need to ask there how they decided on the exact value. But I think it is because large messages are well recognized bad pattern but also having the defaults set for large messages would cause a lot of inefficiencies in the proper use cases.

And also it's not clear how to set to mirrorMaker. Here is mentioned as least two ways and none of them is approved

I'm afraid I do not know the config options out of my head, sorry. But I think it was discussed here, in issues and on Slack many times in the past. So I think you should be able to find it in one of them.

[apolienko]

Thank you!!

What the most easiest way to identrify these heavy messages in strimzi brokers ?
Kafka-ui shares only topic size, but not messages size.
Is there any tools that help to check this info with strimzi ? Thank you, appreciate it

[scholzj]

I don't think there are any tools for that. Kafka in general as a streaming platform does not put much focus on individual messages. So I think you would need to write your own tool for that, read the whole topic and check it message by message.