Kafka MirrorMaker2 cannot increase timeout

[Migueljfs]

Please use this to only for bug reports. For questions or when you need help, you can use the GitHub Discussions, our #strimzi Slack channel or out user mailing list.

Describe the bug
I've tried multiple ways to increase the consumer timeout for mirrormaker2, but the log keeps stating disconnections after 30000ms

To Reproduce
Deploy a strimzi kafka cluster, all default values. Deploy Mirrormaker2 with timeout values and other relevant configs. For example: consumer.max.poll.interval.ms: 150000
NOTE: I'm replicating from a non-strimzi cluster to a strimzi cluster (as I want to migrate). One sits in a GCP West region and the other in a Central region. (reported latency between regions per google is 40ms)

Expected behavior
I expect the timeout time to increase in accordance to the stated config

Environment (please complete the following information):

• Strimzi version: main
• Installation method: YAML files
• Kubernetes cluster: Kubernetes 1.23.12-gke.100
  --

- Infrastructure: GKE

YAML files and logs
cluster.yaml:

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: cluster
  namespace: kafka
spec:
  kafka:
    version: 3.2.3
    replicas: 6
    listeners:
      - name: inside
        port: 9092
        type: internal
        tls: false
      - name: outside
        port: 9094
        type: nodeport
        tls: false
        configuration:
          bootstrap:
            nodePort: 30002
          brokers:
          - broker: 0
            advertisedHost: my-kafka.com
            advertisedPort: 31000
            nodePort: 31000
          - broker: 1
            advertisedHost: my-kafka.com
            advertisedPort: 31001
            nodePort: 31001
          - broker: 2
            advertisedHost: my-kafka.com
            advertisedPort: 31002
            nodePort: 31002
          - broker: 3
            advertisedHost: my-kafka.com
            advertisedPort: 31003
            nodePort: 31003
          - broker: 4
            advertisedHost: my-kafka.com
            advertisedPort: 31004
            nodePort: 31004
          - broker: 5
            advertisedHost: my-kafka.com
            advertisedPort: 31005
            nodePort: 31005
    config:
      auto.create.topics.enable: true
      default.replication.factor: 3
      log.cleaner.threads: 1
      log.segment.bytes: 104857600
      log.retention.hours: 72
      log.retention.check.interval.ms: 60000
      log.flush.interval.ms: 5000
      socket.request.max.bytes: 1048576000
      min.insync.replicas: 2
      num.io.threads: 12
      num.network.threads: 3
      num.replica.fetchers: 64
      replica.fetch.min.bytes: 1
      replica.fetch.wait.max.ms: 500
      offsets.retention.minutes: 10080
      offsets.topic.replication.factor: 3
      num.recovery.threads.per.data.dir: 8
      socket.receive.buffer.bytes: 16777216
      socket.send.buffer.bytes: 16777216
      replica.socket.receive.buffer.bytes: 16777216
      max.incremental.fetch.session.cache.slots: 5000
      leader.imbalance.check.interval.seconds: 60
      transactional.id.expiration.ms: 2147483647
      transaction.state.log.replication.factor: 3
      log.roll.hours: 24
      transaction.state.log.min.isr: 2
      zookeeper.session.timeout.ms: 18000
      replica.lag.time.max.ms: 30000
      cruise.control.metrics.reporter.kubernetes.mode: true
      inter.broker.protocol.version: "3.2"

    logging:
      type: inline
      loggers:
        kafka.root.logger.level: "DEBUG"
    storage:
      type: jbod
      volumes:
      - id: 0
        type: persistent-claim
        size: 300Gi
        deleteClaim: false
        class: premium-rwo
    metricsConfig:
      type: jmxPrometheusExporter
      valueFrom:
        configMapKeyRef:
          name: kafka-metrics
          key: kafka-metrics-config.yml
  zookeeper:
    replicas: 5
    storage:
      type: persistent-claim
      size: 64Gi
      deleteClaim: false
      class: premium-rwo
    metricsConfig:
      type: jmxPrometheusExporter
      valueFrom:
        configMapKeyRef:
          name: kafka-metrics
          key: zookeeper-metrics-config.yml
  entityOperator:
    topicOperator: {}
    userOperator: {}
  kafkaExporter:
    topicRegex: ".*"
    groupRegex: ".*"
  cruiseControl:
    metricsConfig:
      type: jmxPrometheusExporter
      valueFrom:
        configMapKeyRef:
          name: cruise-control-metrics
          key: metrics-config.yml

mm2.yaml:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: kafka
  namespace: kafka
spec:
  version: 3.2.3
  replicas: 3
  connectCluster: "target-cluster"
  clusters:
  - alias: "source-cluster"
    bootstrapServers: my-other-kafka.com:30002
  - alias: "target-cluster"
    bootstrapServers: cluster-kafka-bootstrap.kafka:9092
    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: "source-cluster"
    targetCluster: "target-cluster"
    sourceConnector:
      config:
        replication.factor: 3
        consumer.max.poll.interval.ms: 150000
        offset-syncs.topic.replication.factor: 3
        sync.topic.acls.enabled: "false"
        replication.policy.separator: ""
        replication.policy.class: "org.apache.kafka.connect.mirror.IdentityReplicationPolicy"
    heartbeatConnector:
      config:
        consumer.max.poll.interval.ms: 150000
        heartbeats.topic.replication.factor: 3
    checkpointConnector:
      config:
        consumer.max.poll.interval.ms: 150000
        checkpoints.topic.replication.factor: 3
    topicsPattern: ".*"
    groupsPattern: ".*"
  metricsConfig:
    type: jmxPrometheusExporter
    valueFrom:
      configMapKeyRef:
        name: mirror-maker-2-metrics
        key: metrics-config.yml

logs:

2022-12-09 18:48:42,275 ERROR [source-cluster->target-cluster.MirrorSourceConnector|task-0] Failure sending offset sync. (org.apache.kafka.connect.mirror.MirrorSourceTask) [kafka-producer-network-thread | producer-4]
org.apache.kafka.common.errors.TimeoutException: Expiring 10 record(s) for mm2-offset-syncstarget-clusterinternal-0:120001 ms has passed since batch creation
2022-12-09 18:48:42,275 ERROR [source-cluster->target-cluster.MirrorSourceConnector|task-0] Failure sending offset sync. (org.apache.kafka.connect.mirror.MirrorSourceTask) [kafka-producer-network-thread | producer-4]
org.apache.kafka.common.errors.TimeoutException: Expiring 10 record(s) for mm2-offset-syncstarget-clusterinternal-0:120001 ms has passed since batch creation
2022-12-09 18:48:42,275 ERROR [source-cluster->target-cluster.MirrorSourceConnector|task-0] Failure sending offset sync. (org.apache.kafka.connect.mirror.MirrorSourceTask) [kafka-producer-network-thread | producer-4]
org.apache.kafka.common.errors.TimeoutException: Expiring 10 record(s) for mm2-offset-syncstarget-clusterinternal-0:120001 ms has passed since batch creation
2022-12-09 18:48:59,326 INFO [source-cluster->target-cluster.MirrorSourceConnector|task-0] [Consumer clientId=consumer-null-4, groupId=null] Error sending fetch request (sessionId=INVALID, epoch=INITIAL) to node 13: (org.apache.kafka.clients.FetchSessionHandler) [task-thread-source-cluster->target-cluster.MirrorSourceConnector-0]
org.apache.kafka.common.errors.TimeoutException: Failed to send request after 30000 ms.
2022-12-09 18:49:03,964 INFO [source-cluster->target-cluster.MirrorSourceConnector|task-0] [Consumer clientId=consumer-null-4, groupId=null] Disconnecting from node 13 due to socket connection setup timeout. The timeout value is 34726 ms. (org.apache.kafka.clients.NetworkClient) [task-thread-source-cluster->target-cluster.MirrorSourceConnector-0]
2022-12-09 18:49:03,964 INFO [source-cluster->target-cluster.MirrorSourceConnector|task-0] [Consumer clientId=consumer-null-4, groupId=null] Error sending fetch request (sessionId=INVALID, epoch=INITIAL) to node 13: (org.apache.kafka.clients.FetchSessionHandler) [task-thread-source-cluster->target-cluster.MirrorSourceConnector-0]
org.apache.kafka.common.errors.DisconnectException
2022-12-09 18:49:06,113 INFO [source-cluster->target-cluster.MirrorSourceConnector|task-0] [Consumer clientId=consumer-null-4, groupId=null] Disconnecting from node 12 due to socket connection setup timeout. The timeout value is 28420 ms. (org.apache.kafka.clients.NetworkClient) [task-thread-source-cluster->target-cluster.MirrorSourceConnector-0]
2022-12-09 18:49:06,113 INFO [source-cluster->target-cluster.MirrorSourceConnector|task-0] [Consumer clientId=consumer-null-4, groupId=null] Error sending fetch request (sessionId=INVALID, epoch=INITIAL) to node 12: (org.apache.kafka.clients.FetchSessionHandler) [task-thread-source-cluster->target-cluster.MirrorSourceConnector-0]
org.apache.kafka.common.errors.DisconnectException
2022-12-09 18:49:11,770 INFO [source-cluster->target-cluster.MirrorSourceConnector|task-0] [Producer clientId=producer-4] Disconnecting from node 13 due to socket connection setup timeout. The timeout value is 29206 ms. (org.apache.kafka.clients.NetworkClient) [kafka-producer-network-thread | producer-4]

[scholzj]

The full Mirror Maker 2 logs will print the used configurations. So the start will be to look into them and see if the configurations are applied. TBH, I'm not sure the errors from the snippet have anything to do about this option.

[Migueljfs]

In the logs it appears correctly (as in, the value I configured), but I always have the same error no matter how many timeout configurations I change. It's always 30000ms timeout and I do not understand what I have to change to increase this value