Help Configuring SASL_SSL for KafkaMirrorMaker2 in Strimzi

[emimifleur]

Hello :)
 
I am experiencing issues configuring KafkaMirrorMaker2 with SASL_SSL authentication. While SASL_SSL is successfully configured in both primary and target clusters, and services are operating as expected, configuring SASL_SSL for MirrorMaker2 has become a stumbling block.
 
The primary and target Kafka clusters are using SCRAM-SHA-512 for authentication, and TLS for encryption with trusted certificates (the user operator created tls users, and I created keystore and truststore secrets from the tls user secrets). The services within the clusters are communicating without issues.
 
However, when setting up KafkaMirrorMaker2, I encounter an SSL handshake error:
 

javax.net.ssl.SSLHandshakeException: No subject alternative DNS name matching target-cluster-bootstrap-server:9093 found

Metadata update failed due to authentication error.
 
Failed to connect to and describe kafka cluster. Check broker connection and security properties.

 
Mote: MM2 was working perfectly with SASL_PLAINTEXT config.
 
 
Below is the KafkaMirrorMaker2 configuration I am using:
 

apiVersion: "kafka.strimzi.io/v1beta2"
kind: "KafkaMirrorMaker2"
metadata:
  name: mm2
spec:
  version: "3.4.0"
  replicas: 1
  connectCluster: "lo-target-cluster"
  clusters:
    - alias: "lo-source-cluster"
      bootstrapServers: kafka-cluster-kafka-external-bootstrap.svc.9095
      authentication:
        type: "scram-sha-512"
        username: "connect-user-source"
        passwordSecret:
          secretName: "connect-user-source"
          password: "password"
      tls:
        trustedCertificates:
          - secretName: "kafka-cluster-cluster-ca-cert-source"
            certificate: "ca.crt"
      config:
        "config.storage.replication.factor": "3"
        "offset.storage.replication.factor": "3"
        "status.storage.replication.factor": "3"
    - alias: "lo-target-cluster"
      bootstrapServers: target-kafka-external-bootstrap.svc.9095
      authentication:
        type: "scram-sha-512"
        username: "connect-user"
        passwordSecret:
          secretName: "connect-user"
          password: "target-secret-password"
      tls:
        trustedCertificates:
          - secretName: "kafka-cluster-us-east-cluster-ca-cert"
            certificate: "ca.crt"
      config:
        "config.storage.replication.factor": "3"
        "offset.storage.replication.factor": "3"
        "status.storage.replication.factor": "3"
  mirrors:
    - sourceCluster: "lo-source-cluster"
      targetCluster: "lo-target-cluster"
      sourceConnector:
        config:
          "replication.factor": "3"
          "offset-syncs.topic.replication.factor": "3"
          "sync.topic.acls.enabled": "true"
          "replication.policy.separator": "-"
          "replication.policy.class": "io.strimzi.kafka.connect.mirror.IdentityReplicationPolicy"
      heartbeatConnector:
        config:
          "heartbeats.topic.replication.factor": "3"
          "replication.policy.class": "io.strimzi.kafka.connect.mirror.IdentityReplicationPolicy"
      checkpointConnector:
        config:
          "checkpoints.topic.replication.factor": "3"
          "sync.group.offsets.enabled": "true"
          "replication.policy.separator": "-"
          "replication.policy.class": "io.strimzi.kafka.connect.mirror.IdentityReplicationPolicy"
      topicsPattern: ".*"
      groupsPattern: ".*"
  logging:
    type: inline
    loggers:
      connect.root.logger.level: "INFO"
  readinessProbe:
    initialDelaySeconds: 25
    timeoutSeconds: 25
  livelinessProbe:
    initialDelaySeconds: 25
    timeoutSeconds: 25
  resources:
    requests:
      cpu: "1"
      memory: "1Gi"
    limits:
      cpu: "2"
      memory: "2Gi"
  template:
    pod:
      affinity:
        podAntiAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            nodeSelectorTerms:
              - matchExpressions:
                  - key: "strimzi.io/cluster"
                    operator: "In"
                    values: ["mm2-example"]

 
I have configured the tls.trustedCertificates to use ca.crt from the source and target cluster secrets. I suspect I might be missing keystore configuration for mTLS or additional TLS settings for MirrorMaker2.
 
Could you please provide guidance on configuring the keystore for MirrorMaker2, or if there are additional configurations or considerations I should be aware of for SASL_SSL with MirrorMaker2?

[scholzj]

The error says that you are tryin to connect to target-cluster-external-bootstrap.namespace.svc but this hostname is not in the TLS certificate. But this does not seem to correspond to the KafkaMirrorMaker2 resource you shared which does not seem to configure target-cluster-external-bootstrap.namespace.svc. So these two do not seem to correspond. Seeing the Kafka resources for the Kafka clusters and understanding how you got the certificate secrets would be also important to see what the issue might be.

[emimifleur]

I used dummy values which clearly made it confusing. Here is the actual MM2 file:

{{- with .Values.mirrorMaker }
{{- if .enabled }}}
apiVersion: "kafka.strimzi.io/v1beta2"
kind: "KafkaMirrorMaker2"
metadata:
  name: "{{ .name }}"
spec:
  version: "{{ $.Values.kafka.version }}"
  replicas: {{ .replicas }}
  connectCluster: lo-target-cluster
  Clusters:
    - alias: lo-source-cluster
      bootstrapServers: kafka-cluster-kafka-external-bootstrap.svc.9095
      authentication:
        type: scram-sha-512
        username: connect-user-source
        passwordSecret:
          secretName: "connect-user-source"
          password: "password"
      tls:
        trustedCertificates:
          - secretName: kafka-cluster-cluster-ca-cert-source
            certificate: ca.crt
      config:
        "config.storage.replication.factor": -1
        "offset.storage.replication.factor": -1
        "status.storage.replication.factor": -1
    - alias: lo-target-cluster
      bootstrapServers: target-cluster-external-bootstrap.svc.9095
      authentication:
        type: scram-sha-512
        username: connect-user
        passwordSecret:
          secretName: "connect-user"
          password: "password"
      tls:
        trustedCertificates:
          - secretName: kafka-cluster-us-east-cluster-ca-cert
            certificate: ca.crt
      config:
        "config.storage.replication.factor":  {{ .mirror.sourceCluster.storageReplicationFactor }}
        "offset.storage.replication.factor":  {{ .mirror.sourceCluster.storageReplicationFactor }}
        "status.storage.replication.factor":  {{ .mirror.sourceCluster.storageReplicationFactor }}
  mirrors:
    - sourceCluster: lo-source-cluster
      targetCluster: lo-target-cluster
      sourceConnector:
        config:
          "replication.factor": {{ .mirror.sourceConnector.replicationFactor }}
          "offset-syncs.topic.replication.factor": {{ .mirror.sourceConnector.offsetSyncReplicationFactor }}
          "sync.topic.acls.enabled": {{ .mirror.sourceConnector.syncAclsEnabled}}
          "replication.policy.separator": {{ .mirror.sourceConnector.policySeparator }}
          "replication.policy.class": {{ .mirror.sourceConnector.policyClass}}
      heartbeatConnector:
        config:
          "heartbeats.topic.replication.factor": {{ .mirror.heartbeatConnector.replicationFactor }}
          "replication.policy.class": {{ .mirror.heartbeatConnector.policyClass}}
      checkpointConnector:
        config:
          "checkpoints.topic.replication.factor": {{ .mirror.checkpointConnector.replicationFactor }}
          "sync.group.offsets.enabled": true
          "replication.policy.separator": {{ .mirror.checkpointConnector.policySeparator }}
          "replication.policy.class": {{ .mirror.checkpointConnector.policyClass}}
          {{- end }}
      topicsPattern: {{ .mirror.topicsPattern }}
      groupsPattern: {{ .mirror.groupsPattern }}
  logging:
    type: inline
    loggers:
      connect.root.logger.level: "INFO"
  readinessProbe:
    initialDelaySeconds: 25
    timeoutSeconds: 25
  livelinessProbe:
    initialDelaySeconds: 25
    timeoutSeconds: 25
  resources:
{{ toYaml .mirror.resources | indent 4 }}
  template:
    pod:
      affinity:
        podAntiAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            nodeSelectorTerms:
              - matchExpressions:
                  - key: purpose
                    operator: In
                    values: ["kafka"]
{{- end}}
{{- end}}

Note that the connect-user were created by the user operator.

The kafka-cluster-cluster-ca-cert-source is the kafka-cluster-cluster-ca-cert from the source cluster.
The kafka-cluster-us-east-cluster-ca-cert is the kafka-cluster-cluster-ca-cert from the target cluster.

[scholzj]

That is not useful because that is some kind of template. You need to share the actual resources. Also, you shared the KafkaMirrorMaker2 resource before. But not the Kafka resources.

[emimifleur]

Here is the kafka resource:

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: kafka-cluster
spec:
  kafka:
    version: 3.4.0
    replicas: 3
    image: quay.io/strimzi/kafka:latest-kafka-3.4.0
    rack:
      topologyKey: failure-domain.beta.kubernetes.io/zone
    authorization:
      type: simple
      superUsers:
        - CN=kafka-user
        - CN=connect-user
        - CN=schema-registry-user
        - CN=ksqldb-user
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
        authentication:
          type: scram-sha-512
      - name: external
        port: 9095
        type: loadbalancer
        tls: true
        authentication:
          type: scram-sha-512
        configuration:
          loadBalancerSourceRanges:
            - 172.17.0.0/16
          externalTrafficPolicy: Cluster
          bootstrap:
            LoadBalancerIP: 192.xxx.10.100
            annotations:
              networking.gke.io/load-balancer-type: Internal
              networking.gke.io/internal-load-balancer-allow-global-access: "true"
          brokers:
            - broker: 0
              loadBalancerIP: 192.xxx.10.101
              annotations:
                networking.gke.io/load-balancer-type: Internal
                networking.gke.io/internal-load-balancer-allow-global-access: "true"
            - broker: 1
              loadBalancerIP: 192.xxx.10.102
              annotations:
                networking.gke.io/load-balancer-type: Internal
                networking.gke.io/internal-load-balancer-allow-global-access: "true"
            - broker: 2
              loadBalancerIP: 192.xxx.10.103
              annotations:
                networking.gke.io/load-balancer-type: Internal
                networking.gke.io/internal-load-balancer-allow-global-access: "true"
    logging:
      type: inline
      loggers:
        kafka.root.logger.level: "INFO"
    config:
      auto.create.topics.enable: false
      num.partitions: 3
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2
      default.replication.factor: 3
      min.insync.replicas: 2
      inter.broker.protocol.version: "3.5"
      super.users: "User:kafka-user;User:connect-user"
  resources:
    requests:
      cpu: "1"
      memory: 2Gi
    limits:
      cpu: "2"
      memory: 4Gi
  storage:
    type: persistent-claim
    size: "100Gi"
    deleteClaim: false
  template:
    pod:
      terminationGracePeriodSeconds: 30
      affinity:
        nodeAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            nodeSelectorTerms:
            - matchExpressions:
              - key: purpose
                operator: In
                values: [ "kafka" ]
        podAntiAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
          - labelSelector:
              matchExpressions:
                - key: strimzi.io/name
                  operator: In
                values: 
                 - kafka-cluster
            topologyKey: "kubernetes.io/hostname"
  zookeeper:
    replicas: 3
    resources:
      requests:
        cpu: "0.5"
        memory: "1Gi"
      limits:
        cpu: "1"
        memory: "2Gi"
    storage:
      type: persistent-claim
      size: "50Gi"
      deleteClaim: false
  entityOperator:
    topicOperator: {}
    userOperator: {}

[scholzj]

The problem is, that your resources are inconsistent. Your KafkaMirrorMaker2 connects to target-cluster-external-bootstrap:9093, your error is about target-cluster-external-bootstrap.namespace.svc and your Kafka cluster is named my-cluster and does not seem to match either of those and has no listener on port 9093 where your Mirror Maker is supposed to connect.

If I should guess, you seem to be using an external listener service for access internally. That is wrong. If it is external listener, you have to provide the external address - e.g. the load balancer address. So either use that, or use an internal listener. The correct bootstrap names for the different services is always in the status section of the Kafka custom resource.

[emimifleur]

I updated the original post with the exact values, the external listener uses port 9095. I used the exact configuration MINUS the tls block for SASL_PLAINTEXT, and it was working as expected.

[Neustradamus]

SASL_PLAINTEXT is not secure, password is in plain text.

Have you progressed on SCRAM?

[emimifleur]

Hello, yes.
I was able to resolve the issue and successfully implement SASL_SSL for MirrorMaker2. The problem occurred because both the source and target clusters used the same Kafka user and secret names.

When configuring MirrorMaker 2, I had to define credentials that allow the source cluster to authenticate to the target cluster, and vice versa. Since the same Kafka user existed in both clusters, I added suffixes (for example, kafka-user-source and kafka-user-target) to differentiate the secrets. This ensured that MirrorMaker2 could correctly identify which credentials to use for each cluster.