MM2 unable to replicate messages larger than 1Mb

[JordanPawlett]

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!