Mirror Maker 2: Connection to node terminated during authentication

[alexpollock8]

I am having a connection issue. I am trying to setup active/passive connection between self managed cluster on AWS and Confluent Cloud cluster. Here is the error I am getting when applying mirrormaker2 config yaml to source Kafka cluster. Both are using plaintext.

WARN [AdminClient clientId=adminclient-1] Connection to node -1 terminated during authentication. This may happen due to any of the following reasons: (1) Authentication failed due to invalid credentials with brokers older than 1.0.0, (2) Firewall blocking Kafka TLS traffic (eg it may only allow HTTPS traffic), (3) Transient network issue. (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-1]

Here is my KafkaMirrorMaker2 config yaml

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: kafka-mirror-maker-2
  namespace: kafka
spec:
  version: 3.0.0
  replicas: 1
  connectCluster: "poc-sapcc-basic-1"
  clusters:
  - alias: "cxm-dev-cluster"
    bootstrapServers: kafka-kafka-bootstrap.kafka.svc.cluster.local:9092
  - alias: "poc-sapcc-basic-1"
    bootstrapServers: pkc-pgq85.us-west-2.aws.confluent.cloud:9092
    authentication:
      type: plain
      username: 5AFH5TOHQRTXHOKS
      passwordSecret:
        secretName: poc-confluent-cloud-plain
        password: secretKey
    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: "cxm-dev-cluster"
    targetCluster: "poc-sapcc-basic-1"
    sourceConnector:
      config:
        replication.factor: 1
        offset-syncs.topic.replication.factor: 1
        sync.topic.acls.enabled: "false"
        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: "*"
    groupsPattern: "*"
  logging: 
    type: inline
    loggers:
      connect.root.logger.level: "INFO"
  readinessProbe: 
    initialDelaySeconds: 25
    timeoutSeconds: 25
  livenessProbe:
    initialDelaySeconds: 25
    timeoutSeconds: 25

Here is my source Kafka cluster config:

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: kafka
  namespace: kafka
spec:
  kafka:
    version: 3.0.0
    replicas: 3
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: tls
        port: 9093
        type: internal
        tls: true
    config:
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2
      log.message.format.version: "3.0"
      message.max.bytes: 5242880
      log.retention.hours: 360
      default.replication.factor: 3
      min.insync.replicas: 2
      auto.create.topics.enable: false
    resources:
      requests:
        cpu: "1.0"
        memory: 4Gi
    rack:
      topologyKey: topology.kubernetes.io/zone
    storage:
      type: jbod
      volumes:
      - id: 0
        type: persistent-claim
        size: 120Gi
        deleteClaim: false       
    metricsConfig:
      type: jmxPrometheusExporter
      valueFrom:
        configMapKeyRef:
          name: kafka-metrics
          key: kafka-metrics-config.yaml
  zookeeper:
    replicas: 3
    resources:
      requests:
        cpu: 0.2
        memory: 512M
    storage:
      type: persistent-claim
      size: 100Gi
      deleteClaim: false
    metricsConfig:
      type: jmxPrometheusExporter
      valueFrom:
        configMapKeyRef:
          name: kafka-metrics
          key: zookeeper-metrics-config.yaml
  entityOperator:
    topicOperator: {}
    userOperator: {}

MirrorMaker2 Pod logs:

Preparing MirrorMaker 2.0 cluster truststores and keystores
Preparing MirrorMaker 2.0 truststores and keystores for cluster cxm-dev-cluster
  with trusted certs 
  with tls auth certs 
  with tls auth keys 
Preparing MirrorMaker 2.0 truststores and keystores for cluster poc-sapcc-basic-1
  with trusted certs 
  with tls auth certs 
  with tls auth keys 
Preparing MirrorMaker 2.0 cluster truststores is complete
Creating connector configuration:


# SASL
poc-sapcc-basic-1.sasl.password=[hidden]





Preparing truststore
Preparing truststore is complete
Starting Kafka Connect with configuration:
# Bootstrap servers
bootstrap.servers=pkc-pgq85.us-west-2.aws.confluent.cloud:9092
# REST Listeners
rest.port=8083
rest.advertised.host.name=192.168.73.226 
rest.advertised.port=8083
# Plugins
plugin.path=/opt/kafka/plugins
# Provided configuration
config.storage.topic=mirrormaker2-cluster-configs
group.id=mirrormaker2-cluster
status.storage.topic=mirrormaker2-cluster-status
config.providers.file.class=org.apache.kafka.common.config.provider.FileConfigProvider
offset.storage.topic=mirrormaker2-cluster-offsets
config.providers=file
value.converter=org.apache.kafka.connect.converters.ByteArrayConverter
key.converter=org.apache.kafka.connect.converters.ByteArrayConverter
header.converter=org.apache.kafka.connect.converters.ByteArrayConverter
config.storage.replication.factor=-1
offset.storage.replication.factor=-1
status.storage.replication.factor=-1


security.protocol=SASL_PLAINTEXT
producer.security.protocol=SASL_PLAINTEXT
consumer.security.protocol=SASL_PLAINTEXT
admin.security.protocol=SASL_PLAINTEXT


sasl.mechanism=PLAIN
sasl.jaas.config=[hidden]

producer.sasl.mechanism=PLAIN
producer.sasl.jaas.config=[hidden]

consumer.sasl.mechanism=PLAIN
consumer.sasl.jaas.config=[hidden]

admin.sasl.mechanism=PLAIN
admin.sasl.jaas.config=[hidden]

# Additional configuration
client.rack=

2022-06-27 14:29:28,535 INFO DistributedConfig values: 
	access.control.allow.methods = 
	access.control.allow.origin = 
	admin.listeners = null
	bootstrap.servers = [pkc-pgq85.us-west-2.aws.confluent.cloud:9092]
	client.dns.lookup = use_all_dns_ips
	client.id = 
	config.providers = [file]
	config.storage.replication.factor = -1
	config.storage.topic = mirrormaker2-cluster-configs
	connect.protocol = sessioned
	connections.max.idle.ms = 540000
	connector.client.config.override.policy = All
	group.id = mirrormaker2-cluster
	header.converter = class org.apache.kafka.connect.converters.ByteArrayConverter
	heartbeat.interval.ms = 3000
	inter.worker.key.generation.algorithm = HmacSHA256
	inter.worker.key.size = null
	inter.worker.key.ttl.ms = 3600000
	inter.worker.signature.algorithm = HmacSHA256
	inter.worker.verification.algorithms = [HmacSHA256]
	key.converter = class org.apache.kafka.connect.converters.ByteArrayConverter
	listeners = [http://:8083]
	metadata.max.age.ms = 300000
	metric.reporters = []
	metrics.num.samples = 2
	metrics.recording.level = INFO
	metrics.sample.window.ms = 30000
	offset.flush.interval.ms = 60000
	offset.flush.timeout.ms = 5000
	offset.storage.partitions = 25
	offset.storage.replication.factor = -1
	offset.storage.topic = mirrormaker2-cluster-offsets
	plugin.path = [/opt/kafka/plugins]
	rebalance.timeout.ms = 60000
	receive.buffer.bytes = 32768
	reconnect.backoff.max.ms = 1000
	reconnect.backoff.ms = 50
	request.timeout.ms = 40000
	response.http.headers.config = 
	rest.advertised.host.name = 192.168.73.226
	rest.advertised.listener = null
	rest.advertised.port = 8083
	rest.extension.classes = []
	retry.backoff.ms = 100
	sasl.client.callback.handler.class = null
	sasl.jaas.config = [hidden]
	sasl.kerberos.kinit.cmd = /usr/bin/kinit
	sasl.kerberos.min.time.before.relogin = 60000
	sasl.kerberos.service.name = null
	sasl.kerberos.ticket.renew.jitter = 0.05
	sasl.kerberos.ticket.renew.window.factor = 0.8
	sasl.login.callback.handler.class = null
	sasl.login.class = null
	sasl.login.refresh.buffer.seconds = 300
	sasl.login.refresh.min.period.seconds = 60
	sasl.login.refresh.window.factor = 0.8
	sasl.login.refresh.window.jitter = 0.05
	sasl.mechanism = PLAIN
	scheduled.rebalance.max.delay.ms = 300000
	security.protocol = SASL_PLAINTEXT
	send.buffer.bytes = 131072
	session.timeout.ms = 10000
	socket.connection.setup.timeout.max.ms = 30000
	socket.connection.setup.timeout.ms = 10000
	ssl.cipher.suites = null
	ssl.client.auth = none
	ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
	ssl.endpoint.identification.algorithm = https
	ssl.engine.factory.class = null
	ssl.key.password = null
	ssl.keymanager.algorithm = SunX509
	ssl.keystore.certificate.chain = null
	ssl.keystore.key = null
	ssl.keystore.location = null
	ssl.keystore.password = null
	ssl.keystore.type = JKS
	ssl.protocol = TLSv1.3
	ssl.provider = null
	ssl.secure.random.implementation = null
	ssl.trustmanager.algorithm = PKIX
	ssl.truststore.certificates = null
	ssl.truststore.location = null
	ssl.truststore.password = null
	ssl.truststore.type = JKS
	status.storage.partitions = 5
	status.storage.replication.factor = -1
	status.storage.topic = mirrormaker2-cluster-status
	task.shutdown.graceful.timeout.ms = 5000
	topic.creation.enable = true
	topic.tracking.allow.reset = true
	topic.tracking.enable = true
	value.converter = class org.apache.kafka.connect.converters.ByteArrayConverter
	worker.sync.timeout.ms = 3000
	worker.unsync.backoff.ms = 300000
 (org.apache.kafka.connect.runtime.distributed.DistributedConfig) [main]
2022-06-27 14:29:28,545 INFO Creating Kafka admin client (org.apache.kafka.connect.util.ConnectUtils) [main]
2022-06-27 14:29:28,549 INFO AdminClientConfig values: 
	bootstrap.servers = [pkc-pgq85.us-west-2.aws.confluent.cloud:9092]
	client.dns.lookup = use_all_dns_ips
	client.id = 
	connections.max.idle.ms = 300000
	default.api.timeout.ms = 60000
	metadata.max.age.ms = 300000
	metric.reporters = []
	metrics.num.samples = 2
	metrics.recording.level = INFO
	metrics.sample.window.ms = 30000
	receive.buffer.bytes = 65536
	reconnect.backoff.max.ms = 1000
	reconnect.backoff.ms = 50
	request.timeout.ms = 30000
	retries = 2147483647
	retry.backoff.ms = 100
	sasl.client.callback.handler.class = null
	sasl.jaas.config = [hidden]
	sasl.kerberos.kinit.cmd = /usr/bin/kinit
	sasl.kerberos.min.time.before.relogin = 60000
	sasl.kerberos.service.name = null
	sasl.kerberos.ticket.renew.jitter = 0.05
	sasl.kerberos.ticket.renew.window.factor = 0.8
	sasl.login.callback.handler.class = null
	sasl.login.class = null
	sasl.login.refresh.buffer.seconds = 300
	sasl.login.refresh.min.period.seconds = 60
	sasl.login.refresh.window.factor = 0.8
	sasl.login.refresh.window.jitter = 0.05
	sasl.mechanism = PLAIN
	security.protocol = SASL_PLAINTEXT
	security.providers = null
	send.buffer.bytes = 131072
	socket.connection.setup.timeout.max.ms = 30000
	socket.connection.setup.timeout.ms = 10000
	ssl.cipher.suites = null
	ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
	ssl.endpoint.identification.algorithm = https
	ssl.engine.factory.class = null
	ssl.key.password = null
	ssl.keymanager.algorithm = SunX509
	ssl.keystore.certificate.chain = null
	ssl.keystore.key = null
	ssl.keystore.location = null
	ssl.keystore.password = null
	ssl.keystore.type = JKS
	ssl.protocol = TLSv1.3
	ssl.provider = null
	ssl.secure.random.implementation = null
	ssl.trustmanager.algorithm = PKIX
	ssl.truststore.certificates = null
	ssl.truststore.location = null
	ssl.truststore.password = null
	ssl.truststore.type = JKS
 (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2022-06-27 14:29:28,908 INFO Successfully logged in. (org.apache.kafka.common.security.authenticator.AbstractLogin) [main]
2022-06-27 14:29:28,983 WARN The configuration 'producer.sasl.jaas.config' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2022-06-27 14:29:28,983 WARN The configuration 'group.id' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2022-06-27 14:29:28,983 WARN The configuration 'rest.advertised.port' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2022-06-27 14:29:28,983 WARN The configuration 'plugin.path' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2022-06-27 14:29:28,983 WARN The configuration 'config.providers' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2022-06-27 14:29:28,983 WARN The configuration 'consumer.sasl.mechanism' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2022-06-27 14:29:28,983 WARN The configuration 'admin.security.protocol' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2022-06-27 14:29:28,983 WARN The configuration 'status.storage.replication.factor' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2022-06-27 14:29:28,983 WARN The configuration 'config.providers.file.class' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2022-06-27 14:29:28,983 WARN The configuration 'offset.storage.topic' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2022-06-27 14:29:28,983 WARN The configuration 'consumer.security.protocol' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2022-06-27 14:29:28,983 WARN The configuration 'admin.sasl.mechanism' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2022-06-27 14:29:28,983 WARN The configuration 'value.converter' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2022-06-27 14:29:28,984 WARN The configuration 'key.converter' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2022-06-27 14:29:28,984 WARN The configuration 'consumer.sasl.jaas.config' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2022-06-27 14:29:28,984 WARN The configuration 'config.storage.topic' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2022-06-27 14:29:28,984 WARN The configuration 'producer.security.protocol' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2022-06-27 14:29:28,984 WARN The configuration 'rest.advertised.host.name' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2022-06-27 14:29:28,984 WARN The configuration 'status.storage.topic' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2022-06-27 14:29:28,984 WARN The configuration 'client.rack' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2022-06-27 14:29:28,984 WARN The configuration 'header.converter' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2022-06-27 14:29:28,984 WARN The configuration 'producer.sasl.mechanism' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2022-06-27 14:29:28,984 WARN The configuration 'rest.port' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2022-06-27 14:29:28,984 WARN The configuration 'config.storage.replication.factor' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2022-06-27 14:29:28,984 WARN The configuration 'offset.storage.replication.factor' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2022-06-27 14:29:28,984 WARN The configuration 'admin.sasl.jaas.config' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2022-06-27 14:29:28,984 INFO Kafka version: 3.0.0 (org.apache.kafka.common.utils.AppInfoParser) [main]
2022-06-27 14:29:28,984 INFO Kafka commitId: 8cb0a5e9d3441962 (org.apache.kafka.common.utils.AppInfoParser) [main]
2022-06-27 14:29:28,984 INFO Kafka startTimeMs: 1656340168984 (org.apache.kafka.common.utils.AppInfoParser) [main]

[scholzj]

You did not configured any encryption for the Confluent Cloud cluster. Is it really using plaintext communication without TLS?

[alexpollock8]

How can I configure encryption for Confluent Cloud cluster and what should be added to Strimzi mirrormaker2 config file?

[scholzj]

I never used Confluent Cloud - so it is really just a guess based on the error. So I don't know. You would need to start by explaining how is your Confluent Cloud configured and what it needs.

[AAR256]

I'm also facing the same issue while trying to setup MM2 between Confluent Cloud and Self Managed Strimzi cluster. Last I checked in the cluster configuration, Confluent cluster didn't have TLS enabled. Is there any other configuration that we might have to provide to get the setup to working.