Kafka Mirrormaker2 disconnect timeout

[Migueljfs]

Hello,

I setup kafka mirrormaker2 to replicate my old cluster as I want the strimzi kafka cluster to by new one, but I'm seeing weird logs in mirrormaker:

2022-11-21 11:40:51,461 INFO [source-cluster->target-cluster.MirrorSourceConnector|task-0] [Consumer clientId=consumer-null-5, 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-11-21 11:41:04,703 INFO [source-cluster->target-cluster.MirrorSourceConnector|task-0] [Consumer clientId=consumer-null-5, 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.TimeoutException: Failed to send request after 30000 ms.
2022-11-21 11:41:05,881 INFO [source-cluster->target-cluster.MirrorSourceConnector|task-0] [Consumer clientId=consumer-null-5, groupId=null] Disconnecting from node 12 due to socket connection setup timeout. The timeout value is 31181 ms. (org.apache.kafka.clients.NetworkClient) [task-thread-source-cluster->target-cluster.MirrorSourceConnector-0]
2022-11-21 11:41:05,881 INFO [source-cluster->target-cluster.MirrorSourceConnector|task-0] [Consumer clientId=consumer-null-5, 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

He keeps saying it's disconnecting due to timeout and I can't seem to increase the timeout value to be bigger than 30000 ms.

This is my 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
        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:
        heartbeats.topic.replication.factor: 3
    checkpointConnector:
      config:
        checkpoints.topic.replication.factor: 3
    topicsPattern: ".*"
    groupsPattern: ".*"
  metricsConfig:
    type: jmxPrometheusExporter
    valueFrom:
      configMapKeyRef:
        name: mirror-maker-2-metrics
        key: metrics-config.yml

I have tried adding consumer.request.timeout.ms / producer.request.timeout.ms, etc and did not seem to increase the timeout

[karanalang]

Hello,

I setup kafka mirrormaker2 to replicate my old cluster as I want the strimzi kafka cluster to by new one, but I'm seeing weird logs in mirrormaker:

2022-11-21 11:40:51,461 INFO [source-cluster->target-cluster.MirrorSourceConnector|task-0] [Consumer clientId=consumer-null-5, 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-11-21 11:41:04,703 INFO [source-cluster->target-cluster.MirrorSourceConnector|task-0] [Consumer clientId=consumer-null-5, 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.TimeoutException: Failed to send request after 30000 ms.
2022-11-21 11:41:05,881 INFO [source-cluster->target-cluster.MirrorSourceConnector|task-0] [Consumer clientId=consumer-null-5, groupId=null] Disconnecting from node 12 due to socket connection setup timeout. The timeout value is 31181 ms. (org.apache.kafka.clients.NetworkClient) [task-thread-source-cluster->target-cluster.MirrorSourceConnector-0]
2022-11-21 11:41:05,881 INFO [source-cluster->target-cluster.MirrorSourceConnector|task-0] [Consumer clientId=consumer-null-5, 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

He keeps saying it's disconnecting due to timeout and I can't seem to increase the timeout value to be bigger than 30000 ms.

This is my 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
        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:
        heartbeats.topic.replication.factor: 3
    checkpointConnector:
      config:
        checkpoints.topic.replication.factor: 3
    topicsPattern: ".*"
    groupsPattern: ".*"
  metricsConfig:
    type: jmxPrometheusExporter
    valueFrom:
      configMapKeyRef:
        name: mirror-maker-2-metrics
        key: metrics-config.yml

I have tried adding consumer.request.timeout.ms / producer.request.timeout.ms, etc and did not seem to increase the timeout

Hello -
were you able top resolve this issue ?
I'm getting similar issue, added - producer.request.timeout.ms in the yaml, but that is not increasing the timeout

[Migueljfs]

Nope, still waiting on an answer...

[karanalang]

@Migueljfs - i was able to get this working by adding the producer.request.timeout.ms in the config section

sourceConnector:
      config:
        replication.factor: 1
        offset-syncs.topic.replication.factor: 1
        sync.topic.acls.enabled: "false"
        producer.override.batch.size: 327680
        producer.override.linger.ms: 100
        producer.request.timeout.ms: 60000
        consumer.request.timeout.ms: 60000
        consumer.fetch.max.bytes: 52428800
        replication.policy.separator: ""
        replication.policy.class: "io.strimzi.kafka.connect.mirror.IdentityReplicationPolicy"