Having issues replicating topics / CGs from Confluent Kafka managed cluster to Strimzi Kafka cluster

[irmiller22]

Background

I have 2 Kafka clusters - one is a managed Confluent Kafka cluster (dev-gcp-us-west1-yak), and the other is a Strimzi Kafka cluster (kafka-broker-1) that I have running locally via minikube. The MirrorMaker2 workload is also running on minikube, and this implementation is attempting to replicate messages from the dev-gcp-us-west1-yak cluster to the kafka-broker-1 cluster.

From what I understand, the Confluent Kafka cluster is using the following configs:

security.protocol: PLAINTEXT
sasl.mechanism: PLAIN

I'm not exactly sure why that is, and I'll ask Confluent more about this when I can, but for now I'm trying to understand why this isn't working.

Strimzi Kafka Operator configs

kafka-broker-1 cluster configs

---
apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: kafka-broker-1
spec:
  kafka:
    version: 3.4.0
    replicas: 1
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: tls
    authorization:
      type: simple
    config:
      offsets.topic.replication.factor: 1
      transaction.state.log.replication.factor: 1
      transaction.state.log.min.isr: 1
      default.replication.factor: 1
      min.insync.replicas: 1
      inter.broker.protocol.version: "3.4"
    storage:
      type: jbod
      volumes:
      - id: 0
        type: persistent-claim
        size: 5Gi
        deleteClaim: false
  zookeeper:
    replicas: 1
    storage:
      type: persistent-claim
      size: 5Gi
      deleteClaim: false
  entityOperator:
    topicOperator: {}
    userOperator: {}

user configs for kafka-broker-1 cluster

---
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: admin-user-1
  labels:
    strimzi.io/cluster: kafka-broker-1
spec:
  authentication:
    type: tls
  authorization:
    type: simple
    acls:
      - resource:
          type: topic
          name: "*"
          patternType: literal
        operations:
          - Create
          - Describe
          - Read
          - Write
      - resource:
          type: group
          name: "*"
          patternType: literal
        operations:
          - Read

mirrormaker2 configs

---
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: kafka-mirror
spec:
  version: 3.4.0
  replicas: 1
  # The cluster alias for Kafka Connect
  connectCluster: "kafka-broker-1"
  clusters:
  ############### SOURCE CLUSTER (Active) ################
  - alias: "dev-gcp-us-west1-yak"
    bootstrapServers: "<redacted>.us-west1.gcp.confluent.cloud:9092"
# First suggestion - TLS configs have been removed    
#    tls: 
#      trustedCertificates:
#      - secretName: dev-gcp-us-west1-yak
#        certificate: ca.crt
    authentication: 
      type: plain
      username: JV2YK27A4MKMFKLR
      passwordSecret:
        secretName: dev-gcp-us-west1-yak
        password: user.password
    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   
      # avro serialization / deserialization
      key.converter: org.apache.kafka.connect.storage.StringConverter
      value.converter: io.confluent.connect.avro.AvroConverter
      value.converter.schema.registry.url: https://<redacted>.australia-southeast1.gcp.confluent.cloud:8081
      value.converter.schema.registry.basic.auth.user.info: <username_redacted>:<password_redacted>
      # securty
      security.protocol: PLAINTEXT
      sasl.mechanism: PLAIN
      producer.security.protocol: PLAINTEXT
      consumer.security.protocol: PLAINTEXT
      admin.security.protocol: PLAINTEXT
  ############### TARGET CLUSTER (Passive) ################        
  - alias: "kafka-broker-1"
    bootstrapServers: kafka-broker-1-kafka-bootstrap:9093
    tls: 
      trustedCertificates:
      - secretName: kafka-broker-1-cluster-ca-cert
        certificate: ca.crt
    authentication: 
      type: tls
      certificateAndKey:
        secretName: admin-user-1
        certificate: user.crt
        key: user.key
    config:
      # The following KafkaConnect topics are created by default. You can change the default
      # replication factors (it is 3) and their names here.
      #
      # Kafka topic that stores connector and task status configurations
      # Replication factor for internal topic 'mirrormaker2-cluster-configs'
      config.storage.replication.factor: 1
      config.storage.topic: mirrormaker2-cluster-configs
      #
      # Kafka topic that stores connector offsets.
      # Replication factor for internal topic 'mirrormaker2-cluster-offsets'
      offset.storage.replication.factor: 1
      offset.storage.topic: mirrormaker2-cluster-offsets
      #
      # Kafka topic that stores connector and task status updates.
      # Replication factor for internal topic 'mirrormaker2-cluster-status'
      status.storage.replication.factor: 1
      status.storage.topic: mirrormaker2-cluster-status
      #
      #ssl.cipher.suites: "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384"
      #ssl.enabled.protocols: "TLSv1.2"
      #ssl.protocol: "TLSv1.2"
      #compression.type: gzip
      #batch.size: 8192      
  mirrors:
  #######################
  # Replication flows
  #######################
  - sourceCluster: "dev-gcp-us-west1-yak"
    targetCluster: "kafka-broker-1"
    sourceConnector:
      tasksMax: 1
      config:
        # The replication factor for mirrored topics created at the target cluster
        replication.factor: 1
        # The replication factor for the MirrorSourceConnector 'offset-syncs' internal topic
        # that maps the offsets of the source and target clusters.
        offset-syncs.topic.replication.factor: 1
        sync.topic.acls.enabled: "false"
        # Defines the separator used for the renaming of remote topics (useful when mirroring ACTIVE / ACTIVE clusters)
        replication.policy.separator: ""
        # Override automatic renaming (use this policy for ACTIVE / PASSIVE use cases)
        # Instead of prepending the name with the name of the source cluster, the topic retains its original name
        replication.policy.class: "io.strimzi.kafka.connect.mirror.IdentityReplicationPolicy"
        # Refresh interval to check new topics created in source cluster
        refresh.topics.interval.seconds: 60
    heartbeatConnector:
      config:
        # Internal topic 'heartbeats' created at the target cluster
        heartbeats.topic.replication.factor: 3
    checkpointConnector:
      config:
        # Internal topic '<source-cluster-alias>.checkpoints.internal' created at the target cluster
        checkpoints.topic.replication.factor: 3
        # Override automatic renaming (use this policy for ACTIVE / PASSIVE use cases)
        # Instead of prepending the name with the name of the source cluster, the topic retains its original name
        replication.policy.class: "io.strimzi.kafka.connect.mirror.IdentityReplicationPolicy"        
        # Offset synchronization periodically transfers the consumer offsets for the consumer groups
        # of a source cluster into the consumer offsets topic of a target cluster.
        sync.group.offsets.enabled: true
        # Interval period to sync data offsets from source cluster
        #sync.group.offsets.interval.seconds: 60
        #emit.checkpoints.interval.seconds: 60
        # Interval period to refresh consumer groups statuses (Default: 600)
        refresh.groups.interval.seconds: 60
    topicsPattern: ".*"
    groupsPattern: ".*"
  logging:
    type: inline
    loggers:
      connect.root.logger.level: "INFO"
  readinessProbe:
    initialDelaySeconds: 45
    timeoutSeconds: 5
  livenessProbe:
    initialDelaySeconds: 45
    timeoutSeconds: 5

Error Logs

These are the following errors logs that I see when I spin mirrormaker up:

2023-07-13 18:24:44,720 INFO [dev-gcp-us-west1-yak->kafka-broker-1.MirrorCheckpointConnector|worker] [AdminClient clientId=adminclient-3] Metadata update failed (org.apache.kafka.clients.admin.internals.AdminMetadataManager) [kafka-admin-client-thread | adminclient-3]
org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: fetchMetadata
2023-07-13 18:24:44,723 INFO [dev-gcp-us-west1-yak->kafka-broker-1.MirrorSourceConnector|worker] [AdminClient clientId=adminclient-4] Metadata update failed (org.apache.kafka.clients.admin.internals.AdminMetadataManager) [kafka-admin-client-thread | adminclient-4]
org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: fetchMetadata
2023-07-13 18:24:44,750 INFO [dev-gcp-us-west1-yak->kafka-broker-1.MirrorSourceConnector|worker] [AdminClient clientId=adminclient-7] Metadata update failed (org.apache.kafka.clients.admin.internals.AdminMetadataManager) [kafka-admin-client-thread | adminclient-7]
org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: fetchMetadata
2023-07-13 18:24:44,752 ERROR [dev-gcp-us-west1-yak->kafka-broker-1.MirrorSourceConnector|worker] Scheduler for MirrorSourceConnector timed out running task: creating upstream offset-syncs topic (org.apache.kafka.connect.mirror.Scheduler) [connector-thread-dev-gcp-us-west1-yak->kafka-broker-1.MirrorSourceConnector]
2023-07-13 18:24:44,753 ERROR [dev-gcp-us-west1-yak->kafka-broker-1.MirrorSourceConnector|worker] Scheduler for MirrorSourceConnector caught exception in scheduled task: creating upstream offset-syncs topic (org.apache.kafka.connect.mirror.Scheduler) [Scheduler for MirrorSourceConnector-creating upstream offset-syncs topic]
org.apache.kafka.connect.errors.ConnectException: Timed out while checking for or creating topic 'mm2-offset-syncskafka-broker-1internal'. This could indicate a connectivity issue, unavailable topic partitions, or if this is your first use of the topic it may have taken too long to create.
	at org.apache.kafka.connect.mirror.MirrorUtils.createCompactedTopic(MirrorUtils.java:150)
	at org.apache.kafka.connect.mirror.MirrorUtils.createSinglePartitionCompactedTopic(MirrorUtils.java:160)
	at org.apache.kafka.connect.mirror.MirrorSourceConnector.createOffsetSyncsTopic(MirrorSourceConnector.java:312)
	at org.apache.kafka.connect.mirror.Scheduler.run(Scheduler.java:93)
	at org.apache.kafka.connect.mirror.Scheduler.executeThread(Scheduler.java:112)
	at org.apache.kafka.connect.mirror.Scheduler.lambda$execute$2(Scheduler.java:63)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: createTopics
2023-07-13 18:24:44,783 ERROR [dev-gcp-us-west1-yak->kafka-broker-1.MirrorCheckpointConnector|worker] Scheduler for MirrorCheckpointConnector timed out running task: loading initial consumer groups (org.apache.kafka.connect.mirror.Scheduler) [connector-thread-dev-gcp-us-west1-yak->kafka-broker-1.MirrorCheckpointConnector]
2023-07-13 18:24:44,785 INFO [dev-gcp-us-west1-yak->kafka-broker-1.MirrorCheckpointConnector|worker] loading initial consumer groups took 60002 ms (org.apache.kafka.connect.mirror.Scheduler) [Scheduler for MirrorCheckpointConnector-loading initial consumer groups]
2023-07-13 18:24:44,785 INFO [dev-gcp-us-west1-yak->kafka-broker-1.MirrorCheckpointConnector|worker] Started dev-gcp-us-west1-yak->kafka-broker-1.MirrorCheckpointConnector with 0 consumer groups. (org.apache.kafka.connect.mirror.MirrorCheckpointConnector) [connector-thread-dev-gcp-us-west1-yak->kafka-broker-1.MirrorCheckpointConnector]
2023-07-13 18:24:44,785 WARN [dev-gcp-us-west1-yak->kafka-broker-1.MirrorCheckpointConnector|worker] Scheduler for MirrorCheckpointConnector took too long (60002 ms) running task: loading initial consumer groups (org.apache.kafka.connect.mirror.Scheduler) [Scheduler for MirrorCheckpointConnector-loading initial consumer groups]
2023-07-13 18:24:44,787 INFO SourceConnectorConfig values:
	config.action.reload = restart
	connector.class = org.apache.kafka.connect.mirror.MirrorCheckpointConnector
	errors.log.enable = false
	errors.log.include.messages = false
	errors.retry.delay.max.ms = 60000
	errors.retry.timeout = 0
	errors.tolerance = none
	exactly.once.support = requested
	header.converter = null
	key.converter = null
	name = dev-gcp-us-west1-yak->kafka-broker-1.MirrorCheckpointConnector
	offsets.storage.topic = null
	predicates = []
	tasks.max = 1
	topic.creation.groups = []
	transaction.boundary = poll
	transaction.boundary.interval.ms = null
	transforms = []
	value.converter = null
 (org.apache.kafka.connect.runtime.SourceConnectorConfig) [DistributedHerder-connect-1-1]
2023-07-13 18:24:44,787 INFO EnrichedConnectorConfig values:
	config.action.reload = restart
	connector.class = org.apache.kafka.connect.mirror.MirrorCheckpointConnector
	errors.log.enable = false
	errors.log.include.messages = false
	errors.retry.delay.max.ms = 60000
	errors.retry.timeout = 0
	errors.tolerance = none
	exactly.once.support = requested
	header.converter = null
	key.converter = null
	name = dev-gcp-us-west1-yak->kafka-broker-1.MirrorCheckpointConnector
	offsets.storage.topic = null
	predicates = []
	tasks.max = 1
	topic.creation.groups = []
	transaction.boundary = poll
	transaction.boundary.interval.ms = null
	transforms = []
	value.converter = null
 (org.apache.kafka.connect.runtime.ConnectorConfig$EnrichedConnectorConfig) [DistributedHerder-connect-1-1]
2023-07-13 18:24:44,931 INFO [dev-gcp-us-west1-yak->kafka-broker-1.MirrorSourceConnector|worker] [AdminClient clientId=adminclient-4] Node -1 disconnected. (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-4]
2023-07-13 18:24:44,931 INFO [dev-gcp-us-west1-yak->kafka-broker-1.MirrorSourceConnector|worker] [AdminClient clientId=adminclient-4] Cancelled in-flight API_VERSIONS request with correlation id 51 due to node -1 being disconnected (elapsed time since creation: 89ms, elapsed time since send: 89ms, request timeout: 3600000ms) (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-4]
2023-07-13 18:24:45,129 INFO [dev-gcp-us-west1-yak->kafka-broker-1.MirrorSourceConnector|worker] [AdminClient clientId=adminclient-8] Node -1 disconnected. (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-8]
2023-07-13 18:24:45,129 INFO [dev-gcp-us-west1-yak->kafka-broker-1.MirrorSourceConnector|worker] [AdminClient clientId=adminclient-8] Cancelled in-flight API_VERSIONS request with correlation id 13 due to node -1 being disconnected (elapsed time since creation: 84ms, elapsed time since send: 84ms, request timeout: 3600000ms) (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-8]
2023-07-13 18:24:45,322 INFO [dev-gcp-us-west1-yak->kafka-broker-1.MirrorSourceConnector|worker] [AdminClient clientId=adminclient-7] Node -1 disconnected. (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-7]
2023-07-13 18:24:45,322 INFO [dev-gcp-us-west1-yak->kafka-broker-1.MirrorSourceConnector|worker] [AdminClient clientId=adminclient-7] Cancelled in-flight API_VERSIONS request with correlation id 52 due to node -1 being disconnected (elapsed time since creation: 84ms, elapsed time since send: 84ms, request timeout: 3600000ms) (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-7]
2023-07-13 18:24:45,353 INFO [dev-gcp-us-west1-yak->kafka-broker-1.MirrorSourceConnector|worker] [AdminClient clientId=adminclient-10] Node -1 disconnected. (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-10]
2023-07-13 18:24:45,353 INFO [dev-gcp-us-west1-yak->kafka-broker-1.MirrorSourceConnector|worker] [AdminClient clientId=adminclient-10] Cancelled in-flight API_VERSIONS request with correlation id 13 due to node -1 being disconnected (elapsed time since creation: 84ms, elapsed time since send: 84ms, request timeout: 3600000ms) (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-10]
2023-07-13 18:24:45,559 INFO [dev-gcp-us-west1-yak->kafka-broker-1.MirrorCheckpointConnector|worker] [AdminClient clientId=adminclient-3] Node -1 disconnected. (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-3]
2023-07-13 18:24:45,559 INFO [dev-gcp-us-west1-yak->kafka-broker-1.MirrorCheckpointConnector|worker] [AdminClient clientId=adminclient-3] Cancelled in-flight API_VERSIONS request with correlation id 51 due to node -1 being disconnected (elapsed time since creation: 84ms, elapsed time since send: 84ms, request timeout: 3600000ms) (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-3]
2023-07-13 18:24:46,208 INFO [dev-gcp-us-west1-yak->kafka-broker-1.MirrorSourceConnector|worker] [AdminClient clientId=adminclient-8] Node -1 disconnected. (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-8]
2023-07-13 18:24:46,209 INFO [dev-gcp-us-west1-yak->kafka-broker-1.MirrorSourceConnector|worker] [AdminClient clientId=adminclient-8] Cancelled in-flight API_VERSIONS request with correlation id 14 due to node -1 being disconnected (elapsed time since creation: 84ms, elapsed time since send: 84ms, request timeout: 3600000ms) (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-8]
2023-07-13 18:24:46,208 INFO [dev-gcp-us-west1-yak->kafka-broker-1.MirrorSourceConnector|worker] [AdminClient clientId=adminclient-4] Node -1 disconnected. (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-4]
2023-07-13 18:24:46,209 INFO [dev-gcp-us-west1-yak->kafka-broker-1.MirrorSourceConnector|worker] [AdminClient clientId=adminclient-4] Cancelled in-flight API_VERSIONS request with correlation id 52 due to node -1 being disconnected (elapsed time since creation: 85ms, elapsed time since send: 85ms, request timeout: 3600000ms) (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-4]
2023-07-13 18:24:46,408 INFO [dev-gcp-us-west1-yak->kafka-broker-1.MirrorSourceConnector|worker] [AdminClient clientId=adminclient-7] Node -1 disconnected. (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-7]
2023-07-13 18:24:46,408 INFO [dev-gcp-us-west1-yak->kafka-broker-1.MirrorSourceConnector|worker] [AdminClient clientId=adminclient-7] Cancelled in-flight API_VERSIONS request with correlation id 53 due to node -1 being disconnected (elapsed time since creation: 94ms, elapsed time since send: 94ms, request timeout: 3600000ms) (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-7]

Am I doing something wrong with the connection setup for Mirrormaker2? It seems like I am, but I'm not exactly sure how to address it.

[scholzj]

It is not really clear what is not working without a full log. But if your Confluent cluster has security.protocol: PLAINTEXT, that means it has not TLS encryption. So you should not configure it for it in Mirror Maker configuration.

[irmiller22]

@scholzj thanks for your response. I've updated the MirrorMaker2 configurations as per your answer regarding TLS. I've also updated the configuration in the description (tls config is now commented out). I'm still seeing similar connection errors.

I've attached the full set of logs from the mirrormaker2 pod.

mm2.txt

[scholzj]

Well, I guess you should check the permissions you gave to your users as you have several issues around that in the log and you seem to give it also lot less permissions then the Strimzi examples do. You should also doublecheck the configurations. Does your Confluent cluster really use PLAINTEXT with SASL PLAIN? That would be highly insecure, but I do not know how Confluent works really.

[irmiller22]

@scholzj yes - for our use case, we use Confluent Schema Registry to serialize messages, and in order to deserialize them properly, you'd need to be able to retrieve the associated schema metadata. I agree that this isn't a lot of protection though. That's something that I'll ask Confluent about. For their managed clusters, they only give the option to access port 9092 on the bootstrap server, which is typically non-TLS.

I'll double check on the users. The user for kafka-broker-1 already has CRUD access to all topics via a wildcard, and can also read all consumer groups as well. I'll try expanding the permissions. The PLAIN credentials that the dev-gcp-us-west1-yak user has is also using a global access key, so there shouldn't be an issue with access on that end as well.

I'll keep trying a couple of things, and report back.

[irmiller22]

Finally found my issue - it was because the tls setting wasn't defined for the target connector.

This gave me the answer I needed: #2825 (comment).

TL;DR I was unblocked by this config for the target connector:

    tls:
      trustedCertificates: []