[Bug]: Configure kafka connect connection to an existing kafka cluster

[AntonioDiTuri]

Bug Description

I am trying to configure a kafkaConnect instance to connect to an existing cluster:

Steps to reproduce

This is my configuration:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-kafka-connect
spec:
  authentication:
    type: plain
    username: user
    passwordSecret:
      secretName: my-connect-secret
      password: my-password-field

  replicas: 1
  bootstrapServers: "bootstrap-server"
  image: strimzi/kafka-connect:latest
  livenessProbe:
    initialDelaySeconds: 120
  readinessProbe:
    initialDelaySeconds: 120
  config:
    config.storage.replication.factor: 1
    offset.storage.replication.factor: 1
    status.storage.replication.factor: 1
  logging:
    type: inline
    loggers:
      log4j.rootLogger: DEBUG

---
apiVersion: v1
kind: Secret
metadata:
  name: my-connect-secret
type: Opaque
data:
  my-password-field: "base64 encoded secret"

But this is not working, the error I am getting is:

2024-05-15 13:07:03,378 DEBUG Set SASL client state to RECEIVE_APIVERSIONS_RESPONSE (org.apache.kafka.common.security.authenticator.SaslClientAuthenticator) [kafka-admin-client-thread | adminclient-1]
2024-05-15 13:07:03,378 DEBUG [AdminClient clientId=adminclient-1] Completed connection to node -1. Fetching API versions. (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-1]
2024-05-15 13:07:03,508 DEBUG [AdminClient clientId=adminclient-1] Connection with bootstrap-server disconnected (org.apache.kafka.common.network.Selector) [kafka-admin-client-thread | adminclient-1]
java.io.EOFException
    at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:119)
    at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.receiveResponseOrToken(SaslClientAuthenticator.java:306)
    at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.receiveKafkaResponse(SaslClientAuthenticator.java:394)
    at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.authenticate(SaslClientAuthenticator.java:180)
    at org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:127)
    at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:489)
    at org.apache.kafka.common.network.Selector.poll(Selector.java:427)
    at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:510)
    at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1116)
    at java.lang.Thread.run(Thread.java:748)
2024-05-15 13:07:03,508 DEBUG [AdminClient clientId=adminclient-1] Node -1 disconnected. (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-1]
2024-05-15 13:07:03,508 WARN [AdminClient clientId=adminclient-1] Connection to node -1 terminated during authentication. This may indicate that authentication failed due to invalid credentials. (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-1]
2024-05-15 13:07:04,510 DEBUG [AdminClient clientId=adminclient-1] Initiating connection to node kafka-psv.stage1.client-k8s.dlp-ingest-cons.aws.de.pri.o2.com:9092 (id: -1 rack: null) (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-1]
2024-05-15 13:07:04,510 DEBUG Set SASL client state to SEND_APIVERSIONS_REQUEST (org.apache.kafka.common.security.authenticator.SaslClientAuthenticator) [kafka-admin-client-thread | adminclient-1]
2024-05-15 13:07:04,510 DEBUG Creating SaslClient: client=null;service=kafka;serviceHostname=bootstrap-server;mechs=[SCRAM-SHA-512] (org.apache.kafka.common.security.authenticator.SaslClientAuthenticator) [kafka-admin-client-thread | adminclient-1]
2024-05-15 13:07:04,511 DEBUG Setting SASL/SCRAM_SHA_512 client state to SEND_CLIENT_FIRST_MESSAGE (org.apache.kafka.common.security.scram.internals.ScramSaslClient) [kafka-admin-client-thread | adminclient-1]
2024-05-15 13:07:04,515 DEBUG [AdminClient clientId=adminclient-1] Created socket with SO_RCVBUF = 65536, SO_SNDBUF = 131072, SO_TIMEOUT = 0 to node -1 (org.apache.kafka.common.network.Selector) [kafka-admin-client-thread | adminclient-1]
2024-05-15 13:07:04,515 DEBUG Set SASL client state to RECEIVE_APIVERSIONS_RESPONSE (org.apache.kafka.common.security.authenticator.SaslClientAuthenticator) [kafka-admin-client-thread | adminclient-1]
2024-05-15 13:07:04,516 DEBUG [AdminClient clientId=adminclient-1] Completed connection to node -1. Fetching API versions. (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-1]
2024-05-15 13:07:04,630 DEBUG [AdminClient clientId=adminclient-1] Connection with bootstrap-server disconnected (org.apache.kafka.common.network.Selector) [kafka-admin-client-thread | adminclient-1]
java.io.EOFException
    at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:119)
    at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.receiveResponseOrToken(SaslClientAuthenticator.java:306)
    at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.receiveKafkaResponse(SaslClientAuthenticator.java:394)
    at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.authenticate(SaslClientAuthenticator.java:180)
    at org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:127)
    at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:489)
    at org.apache.kafka.common.network.Selector.poll(Selector.java:427)
    at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:510)
    at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1116)
    at java.lang.Thread.run(Thread.java:748)

I also tried to set:

admin.override.sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required username="user" password="secret""
admin.override.security.protocol: SASL_SSL
admin.override.sasl.mechanism: PLAIN

Since I saw in the logs that the admin is not using sasl_ssl but mechs=[SCRAM-SHA-512]

Since the

Expected behavior

No response

Strimzi version

main

Kubernetes version

1.27

Installation method

Yaml files

Infrastructure

Amazon EKS

Configuration files and logs

No response

Additional context

Any advice on how to solve?

[scholzj]

I think the error suggests some connectivity issue - either it cannot reach the broker, or there is some firewall or something.

[AntonioDiTuri]

Other clients connect in the same way without any problem.
I am not sure also why the admin client has registered SASL/SCRAM_SHA_512 mechanism which is not working of course.
Any idea how to double check?

[scholzj]

I don't follow the comment, sorry. I think this is not the error you would normally get when you have wrong SASL confguration. I also don't see anywhere anything about SCRAM-SHA-512. But in general, you have to know what authentication does your Kafka broker support and whether it has any error related to it.

[AntonioDiTuri]

We use SASL_Plain to authenticate.
With my comment I meant:
2024-05-15 13:07:04,510 DEBUG Creating SaslClient: client=null;service=kafka;serviceHostname=bootstrap-server;mechs=[SCRAM-SHA-512] (org.apache.kafka.common.security.authenticator.SaslClientAuthenticator) [kafka-admin-client-thread | adminclient-1]
Specifically the mecs. If the admin tries to authenticate with scram-sha-512 it would not work

[AntonioDiTuri]

I also tried to add the tls configuration with the proper certificate:

tls:
  trustedCertificates:
   - secretName:  cert-secret
     certificate: certificate-key

And got this error, any ideas?

2024-05-15 12:28:48,707 INFO AdminClientConfig values: 
    bootstrap.servers = [kafka-psv.stage1.client-k8s.dlp-ingest-cons.aws.de.pri.o2.com:9092]
    client.id = 
    connections.max.idle.ms = 300000
    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 = 120000
    retries = 5
    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 = SCRAM-SHA-512
    security.protocol = SASL_SSL
    send.buffer.bytes = 131072
    ssl.cipher.suites = null
    ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
    ssl.endpoint.identification.algorithm = https
    ssl.key.password = null
    ssl.keymanager.algorithm = SunX509
    ssl.keystore.location = null
    ssl.keystore.password = null
    ssl.keystore.type = JKS
    ssl.protocol = TLS
    ssl.provider = null
    ssl.secure.random.implementation = null
    ssl.trustmanager.algorithm = PKIX
    ssl.truststore.location = /tmp/kafka/cluster.truststore.p12
    ssl.truststore.password = [hidden]
    ssl.truststore.type = PKCS12
 (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2024-05-15 12:28:48,737 INFO Successfully logged in. (org.apache.kafka.common.security.authenticator.AbstractLogin) [main]
2024-05-15 12:28:48,813 ERROR Stopping due to error (org.apache.kafka.connect.cli.ConnectDistributed) [main]
org.apache.kafka.common.KafkaException: Failed create new KafkaAdminClient
    at org.apache.kafka.clients.admin.KafkaAdminClient.createInternal(KafkaAdminClient.java:370)
    at org.apache.kafka.clients.admin.AdminClient.create(AdminClient.java:62)
    at org.apache.kafka.connect.util.ConnectUtils.lookupKafkaClusterId(ConnectUtils.java:44)
    at org.apache.kafka.connect.cli.ConnectDistributed.main(ConnectDistributed.java:78)
Caused by: org.apache.kafka.common.KafkaException: org.apache.kafka.common.KafkaException: org.apache.kafka.common.KafkaException: Failed to load SSL keystore /tmp/kafka/cluster.truststore.p12 of type PKCS12
    at org.apache.kafka.common.network.SaslChannelBuilder.configure(SaslChannelBuilder.java:153)
    at org.apache.kafka.common.network.ChannelBuilders.create(ChannelBuilders.java:140)
    at org.apache.kafka.common.network.ChannelBuilders.clientChannelBuilder(ChannelBuilders.java:65)
    at org.apache.kafka.clients.ClientUtils.createChannelBuilder(ClientUtils.java:88)
    at org.apache.kafka.clients.admin.KafkaAdminClient.createInternal(KafkaAdminClient.java:346)
    ... 3 more
Caused by: org.apache.kafka.common.KafkaException: org.apache.kafka.common.KafkaException: Failed to load SSL keystore /tmp/kafka/cluster.truststore.p12 of type PKCS12
    at org.apache.kafka.common.security.ssl.SslFactory.configure(SslFactory.java:137)
    at org.apache.kafka.common.network.SaslChannelBuilder.configure(SaslChannelBuilder.java:149)
    ... 7 more
Caused by: org.apache.kafka.common.KafkaException: Failed to load SSL keystore /tmp/kafka/cluster.truststore.p12 of type PKCS12
    at org.apache.kafka.common.security.ssl.SslFactory$SecurityStore.load(SslFactory.java:330)
    at org.apache.kafka.common.security.ssl.SslFactory.createSSLContext(SslFactory.java:226)
    at org.apache.kafka.common.security.ssl.SslFactory.configure(SslFactory.java:135)
    ... 8 more
Caused by: java.io.IOException: keystore password was incorrect
    at sun.security.pkcs12.PKCS12KeyStore.engineLoad(PKCS12KeyStore.java:2059)
    at java.security.KeyStore.load(KeyStore.java:1445)
    at org.apache.kafka.common.security.ssl.SslFactory$SecurityStore.load(SslFactory.java:327)
    ... 10 more
Caused by: java.security.UnrecoverableKeyException: failed to decrypt safe contents entry: javax.crypto.BadPaddingException: Given final block not properly padded. Such issues can arise if a bad key is used during decryption.
    ... 13 more
Heap
 garbage-first heap   total 1048576K, used 95063K [0x0000000082000000, 0x0000000082400800, 0x0000000800000000)
  region size 4096K, 21 young (86016K), 5 survivors (20480K)
 Metaspace       used 19266K, capacity 19440K, committed 19712K, reserved 1067008K
  class space    used 2337K, capacity 2450K, committed 2560K, reserved 1048576K
Stream closed EOF for default/my-kafka-connect-connect-0 (my-kafka-connect-connect)

[scholzj]

keytool error: java.io.IOException: keystore password was incorrect normally suggests you are using a very old version of Strimzi or that you strimzi/kafka-connect:latest container image is based on an very old version of Strimzi. The operator log suggests that it is Strimzi 0.40.0. So likely the second option? How did you build the image? Or where did you get it?

[AntonioDiTuri]

HI! Many thanks for the comment! I was indeed using an old version of the kafka-connect pulled by the docker registry, did not notice that before. Where can I find newer kafka-connect images?

[AntonioDiTuri]

I found an image that connected, according to the docs also seems correct but the name appears to be misleading: quay.io/strimzi/kafka:latest-kafka-3.7.0 is this correct for kafka connect?

[scholzj]

The images are in the Quay.io repository. You should always use the image matching your Strimzi and Kafka versions. (i.e. not the latest images unless you use Strimzi built directly form the development branch)

[AntonioDiTuri]

Thanks!