MM2 unable to replicate messages larger than 1Mb

[JordanPawlett]

Discussed in #5932

^{Originally posted by JordanPawlett November 23, 2021}
Hello, I'm struggling to figure out how to properly define producer and consumer configuration for the KafkaMirrorMaker2 CRD, as i can't seem to find any relevant documentation.

I'm trying to replicate a few topics from our source cluster to the target cluster. The caveat is, that a few of the topics contain messages that are larger than 1MB in size.

The topics which contain messages that are smaller than 1MB replicate fine. However, the rest do not.

I found this error in the mirrormaker2-cluster-status topic.

{
"state":"FAILED"
"trace":"org.apache.kafka.connect.errors.ConnectException: Unrecoverable exception from producer send callback at org.apache.kafka.connect.runtime.WorkerSourceTask.maybeThrowProducerSendException(WorkerSourceTask.java:266) at org.apache.kafka.connect.runtime.WorkerSourceTask.sendRecords(WorkerSourceTask.java:320) at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:248) at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:182) at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:231) at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) 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.RecordTooLargeException: The message is 2698421 bytes when serialized which is larger than 1048576, which is the value of the max.request.size configuration. "
"worker_id":"ip-adress"
"generation":36
}

We consistently see these errors in the logs from each node

2021-11-23 17:03:46,456 INFO [Consumer clientId=consumer-null-14, groupId=null] Error sending fetch request (sessionId=2115750344, epoch=INITIAL) to node 3: (org.apache.kafka.clients.FetchSessionHandler) [task-thread-source->target.MirrorSourceConnector-0]
org.apache.kafka.common.errors.DisconnectException
2021-11-23 17:03:46,456 INFO [Consumer clientId=consumer-null-14, groupId=null] Error sending fetch request (sessionId=529322538, epoch=INITIAL) to node 6: (org.apache.kafka.clients.FetchSessionHandler) [task-thread-source->target.MirrorSourceConnector-0]
org.apache.kafka.common.errors.DisconnectException
2021-11-23 17:04:13,906 INFO WorkerSourceTask{id=source->target.MirrorSourceConnector-1} flushing 19921 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask) [SourceTaskOffsetCommitter-1]
2021-11-23 17:04:18,606 INFO [Consumer clientId=consumer-null-13, groupId=null] Error sending fetch request (sessionId=1000053246, epoch=2) to node 1: (org.apache.kafka.clients.FetchSessionHandler) [task-thread-source->target.MirrorSourceConnector-1]
org.apache.kafka.common.errors.DisconnectException

Here's our CRD configuration.

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: mirror-maker
  namespace: kafka
spec:
  version: 2.8.0
  replicas: 3
  connectCluster: target
  clusters:
  - alias: source
    bootstrapServers: strimzi-kafka-bootstrap:9093
    config:
      producer.max.request.size: 10485760
      producer.message.max.bytes: 83886080
    tls:
      trustedCertificates:
      - certificate: ca.crt
        secretName: strimzi-cluster-ca-cert
  - alias: target
    bootstrapServers: "external-endpoint:9094"
    config:
      config.storage.replication.factor: -1
      offset.storage.replication.factor: -1
      status.storage.replication.factor: -1
      producer.max.request.size: 10485760
      producer.message.max.bytes: 83886080
    tls:
      trustedCertificates:
      - certificate: ca.crt
        secretName: target-cluster-ca-cert
  mirrors:
  - sourceCluster: source
    targetCluster: target
    sourceConnector:
      tasksMax: 10
      config:
        replication.factor: 1
        offset-syncs.topic.replication.factor: 1
        sync.topic.acls.enabled: "false"
        consumer.max.partition.fetch.bytes: 10485760
        consumer.auto.offset.reset: latest
        max.partition.fetch.bytes: 10485760
        auto.offset.reset: latest
        producer.max.request.size: 10485760
        producer.message.max.bytes: 83886080
        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.class: "io.strimzi.kafka.connect.mirror.IdentityReplicationPolicy"
    topicsPattern: "*"
    # DO NOT MATCH ANY CONSUMER GROUPS. Only replicate topics messages
    # groupsPattern: ""
  resources:
    requests:
      cpu: "1"
      memory: 2Gi
    limits:
      cpu: "2"
      memory: 4Gi
  logging:
    type: inline
    loggers:
      connect.root.logger.level: "INFO"
  template:
    pod:
      affinity:
        podAntiAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            - labelSelector:
                matchExpressions:
                  - key: app.kubernetes.io/instance
                    operator: In
                    values:
                      - mirror-maker
              topologyKey: "kubernetes.io/hostname"
    connectContainer:
      env:
        - name: JAEGER_SERVICE_NAME
          value: kafka-mirror-maker
        - name: JAEGER_ENDPOINT
          value: "http://jaeger-collector.monitoring.svc:14268/api/traces?format=jaeger.thrift"
  tracing:
    type: jaeger

I've tried to define both producer and consumer configuration in a multitude of ways, with no prevail.
If anyone could offer any advice or point me in the right directly / documentation, that'd be great. Thank you!

[scholzj]

What is the value of asking the same question in multiple places? It just creates confusion because if someone replies somewhere with a solution, it will leave it open on many other places.

[JordanPawlett]

What is the value of asking the same question in multiple places? It just creates confusion because if someone replies somewhere with a solution, it will leave it open on many other places.

It's a feature of Github, when you create a discussion it prompts you to open an issue too.
The same question hasn't been asked twice, an issue has been opened that links directly to the discussion.

[scholzj]

Well, my understanding is that that is a feature if the discussion discovers some bug for example and you want to open an issue for it. Right now the question is asked twice ... two different URLs, two different search results:

• MM2 unable to replicate messages larger than 1Mb #5933
• MM2 unable to replicate messages larger than 1Mb #5932

[sergey-shuyskiy]

@JordanPawlett It's works for me
MirrorMaker2 CRD:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: kafka-main-to-dev
spec:
  version: 2.8.0
  connectCluster: "kafka-dev-cluster"
  replicas: 1
  clusters:
    - alias: "kafka-main-cluster"
      bootstrapServers: main-kafka-bootstrap.kafka-main.svc.cluster.local:9092
    - alias: "kafka-dev-cluster"
      bootstrapServers: dev-kafka-plain-bootstrap:9092
      config:
        config.storage.replication.factor: 1
        offset.storage.replication.factor: 1
        status.storage.replication.factor: 1
        producer.max.request.size: "104857600"
        producer.buffer.memory: "104857600"

Kafka brokers (or topics):

max.message.bytes: "104857600"

[sappusaketh]

@sergey-shuyskiy where do you set that producer.max.request.size is it supposed to be set under sourceConnector.spec something similar to this. And also max.request.size is a connector config right how do we decide where we should put this config?

[GoaKoala]

What fixed this problem for us is to add
producer.override.max.request.size: 40000000 # maxMessageSize
under sourceCluster config.

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
spec:
  mirrors:
    - sourceCluster:
        sourceConnector:
          config:
            producer.override.max.request.size: 40000000

Only the relevant config lines included