Mirror between two external Kafka clusters using MirrorMaker2

[kmandal-volvo]

Strimzi: 0.20.X
Kafka: 2.6.0
KafkaConnect: 2.6.0
KafkaMirrorMaker: 2.6.0

PROBLEM DESCRIPTION:

I have encountered SslAuthenticationException: Failed to process post-handshake messages problems when I have tried to deploy KafkaMirrorMaker2 instance to replicate data between 2 Kafka clusters.

• vcc-internal-esp-sandbox is in vcc-internal-esp namespace in one AKS clusters

• vcc-internal-esp is in vcc-internal-esp namespace in another AKS clusters

Now I want to mirror data from vcc-internal-esp to vcc-internal-esp-sandbox using the external routes in aks clusters. I have copied and depolyed vcc-internal-esp-sandbox (target cluster where mirror maker deployed ) certificate and secret to vcc-internal-esp cluser (kafka cluster topic which I need to access from vcc-internal-esp-sandbox) .

vcc-internal-esp-sandbox broker bootstap url is 10.40.205.45:9094 and kafkaconnect bootstap url is vcc-internal-esp-sandbox-kafka-bootstrap:9093

vcc-internal-esp broker bootstap url is kafka.test.volvocars.net:9094 and kafkaconnect bootstap url is vcc-internal-esp-kafka-bootstrap:9093

MirrorMaker2 yaml:

apiVersion: kafka.strimzi.io/v1alpha1
kind: KafkaMirrorMaker2
metadata:
  name: sandbox-mirror-maker-2
  namespace: vcc-internal-esp
spec:
  version: 2.6.0
  replicas: 1
  connectCluster: "vcc-internal-esp-sandbox"
  clusters:
  - alias: "vcc-internal-esp"
    bootstrapServers: vcc-internal-esp-kafka-bootstrap.vcc-internal-esp.svc.cluster.local:9093
    tls:
      trustedCertificates:
        - secretName: vcc-internal-esp-cluster-ca-cert
          certificate: ca.crt
  - alias: "vcc-internal-esp-sandbox"
    bootstrapServers: vcc-internal-esp-sandbox-kafka-bootstrap.vcc-internal-esp.svc.cluster.local:9093
    tls:
      trustedCertificates:
        - secretName: vcc-internal-esp-sandbox-cluster-ca-cert
          certificate: ca.crt
    config:
      config.storage.replication.factor: 1
      offset.storage.replication.factor: 1
      status.storage.replication.factor: 1
  mirrors:
  - sourceCluster: "vcc-internal-esp"
    targetCluster: "vcc-internal-esp-sandbox"
    sourceConnector:
      config:
        replication.factor: 1
        offset-syncs.topic.replication.factor: 1
        sync.topic.acls.enabled: "false"
    heartbeatConnector:
      config:
        heartbeats.topic.replication.factor: 1
    checkpointConnector:
      config:
        checkpoints.topic.replication.factor: 1
    topicsPattern: ".*"
    groupsPattern: ".*" 

Logfile Attached

sandbox-mirror-maker-2-mirrormaker2-7b7fd7697-z5ttc.log

[scholzj]

I do not know what Tag mismatch means in the logs. But you say that each Kafka cluster is in different AKS cluster. But the bootstrap addresses seem ot be both local. So are they coreect? I would expect one of them to be use external address.

[kmandal-volvo]

One Kafka cluster deployed on sandbox Azure Kubernetes cluster and another Kafka deployed on test Azure Kubernetes. They are under the same network. Their bootstrap URL for a broker has the same series Ip as both the cluster is in the same network (Internal inside firewall) In mirror maker bootstrapServers I used Kafka connect bootstrap URL instead of Kafka broker bootstrap URL.

Kafka connect config for sandbox

apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaConnect
metadata:
  name: vcc-esp-connect-private-sandbox
  annotations:
    strimzi.io/use-connector-resources: "true"
#  annotations:
#  # use-connector-resources configures this KafkaConnect
#  # to use KafkaConnector resources to avoid
#  # needing to call the Connect REST API directly
#    strimzi.io/use-connector-resources: "true"
spec:
  image: vccai.azurecr.io/custom-kafka-connect-image:v9.0
  imagePullPolicy: Always
  #version: 2.6.0
  replicas: 2
  bootstrapServers: vcc-internal-esp-sandbox-kafka-bootstrap:9093
  imagePullSecrets:
    - name: azure-docker-registry
  tls:
    trustedCertificates:
      - secretName: vcc-internal-esp-sandbox-cluster-ca-cert
        certificate: ca.crt
  authentication:
    type: tls
    certificateAndKey:
      secretName: icoe-int-super-user
      certificate: user.crt
      key: user.key
  config:
    group.id: connect-cluster
    offset.storage.topic: connect-cluster-offsets
    config.storage.topic: connect-cluster-configs
    status.storage.topic: connect-cluster-status
  metrics:
    # Inspired by kafka-connect rules
    # https://github.com/prometheus/jmx_exporter/blob/master/example_configs/kafka-connect.yml
    lowercaseOutputName: true
    lowercaseOutputLabelNames: true
    rules:
      #kafka.connect:type=app-info,client-id="{clientid}"
      #kafka.consumer:type=app-info,client-id="{clientid}"
      #kafka.producer:type=app-info,client-id="{clientid}"
      - pattern: 'kafka.(.+)<type=app-info, client-id=(.+)><>start-time-ms'
        name: kafka_$1_start_time_seconds
        labels:
          clientId: "$2"
        help: "Kafka $1 JMX metric start time seconds"
        type: GAUGE
        valueFactor: 0.001
      - pattern: 'kafka.(.+)<type=app-info, client-id=(.+)><>(commit-id|version): (.+)'
        name: kafka_$1_$3_info
        value: 1
        labels:
          clientId: "$2"
          $3: "$4"
        help: "Kafka $1 JMX metric info version and commit-id"
        type: GAUGE

      #kafka.producer:type=producer-topic-metrics,client-id="{clientid}",topic="{topic}"", partition="{partition}"
      #kafka.consumer:type=consumer-fetch-manager-metrics,client-id="{clientid}",topic="{topic}"", partition="{partition}"
      - pattern: kafka.(.+)<type=(.+)-metrics, client-id=(.+), topic=(.+), partition=(.+)><>(.+-total|compression-rate|.+-avg|.+-replica|.+-lag|.+-lead)
        name: kafka_$2_$6
        labels:
          clientId: "$3"
          topic: "$4"
          partition: "$5"
        help: "Kafka $1 JMX metric type $2"
        type: GAUGE

      #kafka.producer:type=producer-topic-metrics,client-id="{clientid}",topic="{topic}"
      #kafka.consumer:type=consumer-fetch-manager-metrics,client-id="{clientid}",topic="{topic}"", partition="{partition}"
      - pattern: kafka.(.+)<type=(.+)-metrics, client-id=(.+), topic=(.+)><>(.+-total|compression-rate|.+-avg)
        name: kafka_$2_$5
        labels:
          clientId: "$3"
          topic: "$4"
        help: "Kafka $1 JMX metric type $2"
        type: GAUGE

      #kafka.connect:type=connect-node-metrics,client-id="{clientid}",node-id="{nodeid}"
      #kafka.consumer:type=consumer-node-metrics,client-id=consumer-1,node-id="{nodeid}"
      - pattern: kafka.(.+)<type=(.+)-metrics, client-id=(.+), node-id=(.+)><>(.+-total|.+-avg)
        name: kafka_$2_$5
        labels:
          clientId: "$3"
          nodeId: "$4"
        help: "Kafka $1 JMX metric type $2"
        type: UNTYPED

      #kafka.connect:type=kafka-metrics-count,client-id="{clientid}"
      #kafka.consumer:type=consumer-fetch-manager-metrics,client-id="{clientid}"
      #kafka.consumer:type=consumer-coordinator-metrics,client-id="{clientid}"
      #kafka.consumer:type=consumer-metrics,client-id="{clientid}"
      - pattern: kafka.(.+)<type=(.+)-metrics, client-id=(.*)><>(.+-total|.+-avg|.+-bytes|.+-count|.+-ratio|.+-age|.+-flight|.+-threads|.+-connectors|.+-tasks|.+-ago)
        name: kafka_$2_$4
        labels:
          clientId: "$3"
        help: "Kafka $1 JMX metric type $2"
        type: GAUGE

      #kafka.connect:type=connector-task-metrics,connector="{connector}",task="{task}<> status"
      - pattern: 'kafka.connect<type=connector-task-metrics, connector=(.+), task=(.+)><>status: ([a-z-]+)'
        name: kafka_connect_connector_status
        value: 1
        labels:
          connector: "$1"
          task: "$2"
          status: "$3"
        help: "Kafka Connect JMX Connector status"
        type: GAUGE

      #kafka.connect:type=task-error-metrics,connector="{connector}",task="{task}"
      #kafka.connect:type=source-task-metrics,connector="{connector}",task="{task}"
      #kafka.connect:type=sink-task-metrics,connector="{connector}",task="{task}"
      #kafka.connect:type=connector-task-metrics,connector="{connector}",task="{task}"
      - pattern: kafka.connect<type=(.+)-metrics, connector=(.+), task=(.+)><>(.+-total|.+-count|.+-ms|.+-ratio|.+-avg|.+-failures|.+-requests|.+-timestamp|.+-logged|.+-errors|.+-retries|.+-skipped)
        name: kafka_connect_$1_$4
        labels:
          connector: "$2"
          task: "$3"
        help: "Kafka Connect JMX metric type $1"
        type: GAUGE

      #kafka.connect:type=connector-metrics,connector="{connector}"
      #kafka.connect:type=connect-worker-metrics,connector="{connector}"
      - pattern: kafka.connect<type=connect-worker-metrics, connector=(.+)><>([a-z-]+)
        name: kafka_connect_worker_$2
        labels:
          connector: "$1"
        help: "Kafka Connect JMX metric $1"
        type: GAUGE

      #kafka.connect:type=connect-worker-metrics
      - pattern: kafka.connect<type=connect-worker-metrics><>([a-z-]+)
        name: kafka_connect_worker_$1
        help: "Kafka Connect JMX metric worker"
        type: GAUGE

      #kafka.connect:type=connect-worker-rebalance-metrics
      - pattern: kafka.connect<type=connect-worker-rebalance-metrics><>([a-z-]+)
        name: kafka_connect_worker_rebalance_$1
        help: "Kafka Connect JMX metric rebalance information"
        type: GAUGE
  http:
    port: 8083
---
apiVersion: networking.k8s.io/v1beta1
kind: Ingress
metadata:
  name: vcc-esp-connect-route
  namespace: vcc-internal-esp
  annotations:
    kubernetes.io/ingress.class: nginx
    nginx.ingress.kubernetes.io/ssl-redirect: "false"
    nginx.ingress.kubernetes.io/use-regex: "true"
#    nginx.ingress.kubernetes.io/rewrite-target: /$1
spec:
  rules:
    - http:
        paths:
          - backend:
              serviceName: vcc-esp-connect-connect-api
              servicePort: 8083
            path: /connect
            pathType: Prefix

Kafka connect test cluster configuration

apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaConnect
metadata:
  name: vcc-esp-connect-private-test
  annotations:
    strimzi.io/use-connector-resources: "true"
#  annotations:
#  # use-connector-resources configures this KafkaConnect
#  # to use KafkaConnector resources to avoid
#  # needing to call the Connect REST API directly
#    strimzi.io/use-connector-resources: "true"
spec:
  image: vccai.azurecr.io/custom-kafka-connect-image:v9.0
  imagePullPolicy: Always
  #version: 2.6.0
  replicas: 2
  bootstrapServers: vcc-internal-esp-kafka-bootstrap:9093
  imagePullSecrets:
    - name: azure-docker-registry
  tls:
    trustedCertificates:
      - secretName: vcc-internal-esp-cluster-ca-cert
        certificate: ca.crt
  authentication:
    type: tls
    certificateAndKey:
      secretName: icoe-int-super-user
      certificate: user.crt
      key: user.key
  config:
    group.id: connect-cluster
    offset.storage.topic: connect-cluster-offsets
    config.storage.topic: connect-cluster-configs
    status.storage.topic: connect-cluster-status
  metrics:
    # Inspired by kafka-connect rules
    # https://github.com/prometheus/jmx_exporter/blob/master/example_configs/kafka-connect.yml
    lowercaseOutputName: true
    lowercaseOutputLabelNames: true
    rules:
      #kafka.connect:type=app-info,client-id="{clientid}"
      #kafka.consumer:type=app-info,client-id="{clientid}"
      #kafka.producer:type=app-info,client-id="{clientid}"
      - pattern: 'kafka.(.+)<type=app-info, client-id=(.+)><>start-time-ms'
        name: kafka_$1_start_time_seconds
        labels:
          clientId: "$2"
        help: "Kafka $1 JMX metric start time seconds"
        type: GAUGE
        valueFactor: 0.001
      - pattern: 'kafka.(.+)<type=app-info, client-id=(.+)><>(commit-id|version): (.+)'
        name: kafka_$1_$3_info
        value: 1
        labels:
          clientId: "$2"
          $3: "$4"
        help: "Kafka $1 JMX metric info version and commit-id"
        type: GAUGE

      #kafka.producer:type=producer-topic-metrics,client-id="{clientid}",topic="{topic}"", partition="{partition}"
      #kafka.consumer:type=consumer-fetch-manager-metrics,client-id="{clientid}",topic="{topic}"", partition="{partition}"
      - pattern: kafka.(.+)<type=(.+)-metrics, client-id=(.+), topic=(.+), partition=(.+)><>(.+-total|compression-rate|.+-avg|.+-replica|.+-lag|.+-lead)
        name: kafka_$2_$6
        labels:
          clientId: "$3"
          topic: "$4"
          partition: "$5"
        help: "Kafka $1 JMX metric type $2"
        type: GAUGE

      #kafka.producer:type=producer-topic-metrics,client-id="{clientid}",topic="{topic}"
      #kafka.consumer:type=consumer-fetch-manager-metrics,client-id="{clientid}",topic="{topic}"", partition="{partition}"
      - pattern: kafka.(.+)<type=(.+)-metrics, client-id=(.+), topic=(.+)><>(.+-total|compression-rate|.+-avg)
        name: kafka_$2_$5
        labels:
          clientId: "$3"
          topic: "$4"
        help: "Kafka $1 JMX metric type $2"
        type: GAUGE

      #kafka.connect:type=connect-node-metrics,client-id="{clientid}",node-id="{nodeid}"
      #kafka.consumer:type=consumer-node-metrics,client-id=consumer-1,node-id="{nodeid}"
      - pattern: kafka.(.+)<type=(.+)-metrics, client-id=(.+), node-id=(.+)><>(.+-total|.+-avg)
        name: kafka_$2_$5
        labels:
          clientId: "$3"
          nodeId: "$4"
        help: "Kafka $1 JMX metric type $2"
        type: UNTYPED

      #kafka.connect:type=kafka-metrics-count,client-id="{clientid}"
      #kafka.consumer:type=consumer-fetch-manager-metrics,client-id="{clientid}"
      #kafka.consumer:type=consumer-coordinator-metrics,client-id="{clientid}"
      #kafka.consumer:type=consumer-metrics,client-id="{clientid}"
      - pattern: kafka.(.+)<type=(.+)-metrics, client-id=(.*)><>(.+-total|.+-avg|.+-bytes|.+-count|.+-ratio|.+-age|.+-flight|.+-threads|.+-connectors|.+-tasks|.+-ago)
        name: kafka_$2_$4
        labels:
          clientId: "$3"
        help: "Kafka $1 JMX metric type $2"
        type: GAUGE

      #kafka.connect:type=connector-task-metrics,connector="{connector}",task="{task}<> status"
      - pattern: 'kafka.connect<type=connector-task-metrics, connector=(.+), task=(.+)><>status: ([a-z-]+)'
        name: kafka_connect_connector_status
        value: 1
        labels:
          connector: "$1"
          task: "$2"
          status: "$3"
        help: "Kafka Connect JMX Connector status"
        type: GAUGE

      #kafka.connect:type=task-error-metrics,connector="{connector}",task="{task}"
      #kafka.connect:type=source-task-metrics,connector="{connector}",task="{task}"
      #kafka.connect:type=sink-task-metrics,connector="{connector}",task="{task}"
      #kafka.connect:type=connector-task-metrics,connector="{connector}",task="{task}"
      - pattern: kafka.connect<type=(.+)-metrics, connector=(.+), task=(.+)><>(.+-total|.+-count|.+-ms|.+-ratio|.+-avg|.+-failures|.+-requests|.+-timestamp|.+-logged|.+-errors|.+-retries|.+-skipped)
        name: kafka_connect_$1_$4
        labels:
          connector: "$2"
          task: "$3"
        help: "Kafka Connect JMX metric type $1"
        type: GAUGE

      #kafka.connect:type=connector-metrics,connector="{connector}"
      #kafka.connect:type=connect-worker-metrics,connector="{connector}"
      - pattern: kafka.connect<type=connect-worker-metrics, connector=(.+)><>([a-z-]+)
        name: kafka_connect_worker_$2
        labels:
          connector: "$1"
        help: "Kafka Connect JMX metric $1"
        type: GAUGE

      #kafka.connect:type=connect-worker-metrics
      - pattern: kafka.connect<type=connect-worker-metrics><>([a-z-]+)
        name: kafka_connect_worker_$1
        help: "Kafka Connect JMX metric worker"
        type: GAUGE

      #kafka.connect:type=connect-worker-rebalance-metrics
      - pattern: kafka.connect<type=connect-worker-rebalance-metrics><>([a-z-]+)
        name: kafka_connect_worker_rebalance_$1
        help: "Kafka Connect JMX metric rebalance information"
        type: GAUGE
  http:
    port: 8083
---
apiVersion: networking.k8s.io/v1beta1
kind: Ingress
metadata:
  name: vcc-esp-connect-route
  namespace: vcc-internal-esp
  annotations:
    kubernetes.io/ingress.class: nginx
    nginx.ingress.kubernetes.io/ssl-redirect: "false"
    nginx.ingress.kubernetes.io/use-regex: "true"
#    nginx.ingress.kubernetes.io/rewrite-target: /$1
spec:
  rules:
    - http:
        paths:
          - backend:
              serviceName: vcc-esp-connect-connect-api
              servicePort: 8083
            path: /connect
            pathType: Prefix

[scholzj]

One Kafka cluster deployed on sandbox Azure Kubernetes cluster and another Kafka deployed on test Azure Kubernetes. They are under the same network. Their bootstrap URL for a broker has the same series Ip as both the cluster is in the same network (Internal inside firewall)

I do not know your setup. Are you sure the clusters share the internal Kubernetes network and DNS?

In mirror maker bootstrapServers I used Kafka connect bootstrap URL instead of Kafka broker bootstrap URL.

I do not follow this. Mirror Maker connects to Kafka broker. Not to Kafka Connect. And Kafka Connect does not have any bootstrap address.

[kmandal-volvo]

Now certificate issue resolved. I don't find many issues in the log but the mirror maker pod is failed. Please find attached the log, waiting for your information.
sandbox-mirror-maker-2-mirrormaker2-54975f7f57-75nvx.log

[scholzj]

The log seems quite clear ... it seems to be connected, but not authorized:

2021-07-01 14:53:28,660 ERROR [Worker clientId=connect-1, groupId=mirrormaker2-cluster] Uncaught exception in herder work thread, exiting:  (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [DistributedHerder-connect-1-1]
org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [mirrormaker2-cluster-offsets]

So you need to give it the right ACLS rights.

[kmandal-volvo]

After fixing the TopicAuthorizationException it working fine and able to see the mirror topic. Thanks for your help.