Configuring KafkaConnect against OAuth protected Kafka Broker

[firdaustahir]

Hi,

I've followed Strimzi OAuth documentation steps just to get familiarized with the configuration. Based on that steps, I managed to configured streamer application to access OAuth protected broker.
However, so far I don't have any successful attempt to configured Kafka Connect against OAuth Kafka broker.

Similar to the strimzi steps, I'm using KeyCloak as the IDP/Token provider. Here is the Kafka config yaml file :

cert_manager:
  enabled: true
  config:
  - name: plain
    port: 9092
    type: internal
    tls: false
  - name: tls
    port: 9093
    type: internal
    tls: true
    authentication:
      type: oauth
      clientId: kafka-broker
      clientSecret:
        key: secret
        secretName: broker-oauth-secret
      validIssuerUri: https://keycloak.keycloak:8443/auth/realms/master
      jwksEndpointUri: https://keycloak.keycloak:8443/auth/realms/master/protocol/openid-connect/certs
      userNameClaim: preferred_username
      tlsTrustedCertificates:
      - secretName: ca-truststore
        certificate: ca.crt

kafka:
  replicas: 1
  config:
    offsets.topic.replication.factor: 1
    transaction.state.log.replication.factor: 1
    transaction.state.log.min.isr: 1
        
zookeeper:
  replicas: 1

The KafkaConnect config yaml file is as follows :

name: odf-connect-cluster
version: 3.5.1
replicas: 1
logging:
  type: inline
  loggers:      
    connect.root.logger.level: "DEBUG"
bootstrapServers: odf-cluster-kafka-bootstrap:9093
authentication:
    type: oauth
    clientId: kafka-broker
    clientSecret:
      key: secret
      secretName: broker-oauth-secret
    tokenEndpointUri: https://keycloak.keycloak:8443/auth/realms/master/protocol/openid-connect/token
    tlsTrustedCertificates:
    - secretName: ca-truststore
      certificate: ca.crt
connectContainer:
  env:
    - name : KAFKA_OPTS
      value : "-Djavax.net.ssl.trustStore=/opt/kafka/external-configuration/truststore/truststore.p12 -Djavax.net.ssl.trustStorePassword=truststorepassword"
initContainer:
  env:
    - name : KAFKA_OPTS
      value : "-Djavax.net.ssl.trustStore=/opt/kafka/external-configuration/truststore/truststore.p12 -Djavax.net.ssl.trustStorePassword=truststorepassword"
storage:
  type: ephemeral
config:
  group.id: connect-cluster
  offset.storage.topic: connect-cluster-offsets
  config.storage.topic: connect-cluster-configs
  status.storage.topic: connect-cluster-status
  value.converter: io.confluent.connect.avro.AvroConverter  
  config.storage.replication.factor: 1
  offset.storage.replication.factor: 1
  status.storage.replication.factor: 1
  connector.client.config.override.policy: All
  exactly.once.source.support: disabled
  # ssl.truststore.location: /opt/kafka/external-configuration/truststore/truststore.p12
  # ssl.truststore.password: truststorepassword
externalConfiguration:
  volumes:
    - name: truststore
      secret:
        secretName: sink-truststore
    - name: truststore-password
      secret:
        secretName: sink-truststore-password
    - name: input-value-schemas
      configMap:
        name: input-value-schemas
    - name: input-key-schemas
      configMap:
        name: input-key-schemas
    - name: source-hookpoints
      configMap:
        name: source-hookpoints
    - name: sink-hookpoints
      configMap:
        name: sink-hookpoints
    - name: odf-connect-customfiles
      configMap:
        name: odf-connect-customfiles
        optional: true

When I deployed this Kafka Connect, it failed to connect to broker 9093. The following logs show the messages from kafka connect log :

2024-01-17 02:24:02,530 DEBUG loginWithClientSecret() - tokenEndpointUrl: https://keycloak.keycloak:8443/auth/realms/master/protocol/openid-connect/token, clientId: kafka-broker, clientSecret: K9y5**27CD,      scope: null, audience: null, connectTimeout: 60, readTimeout: 60 (io.strimzi.kafka.oauth.common.OAuthAuthenticator) [main]
2024-01-17 02:24:02,815 DEBUG CallbackHandler io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler does not support SASL extensions. No extensions will be added (org.apache.kafka.common.securi     ty.oauthbearer.OAuthBearerLoginModule) [main]
2024-01-17 02:24:02,819 DEBUG Login succeeded; invoke commit() to commit it; current committed token count=0 (org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule) [main]
2024-01-17 02:24:02,819 DEBUG Done committing my token; committed token count is now 1 (org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule) [main]
2024-01-17 02:24:02,819 INFO Successfully logged in. (org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin) [main]
2024-01-17 02:24:02,819 DEBUG Found expiring credential with principal '2ff1f418-b363-4d8a-b37a-4e506eb2c783'. (org.apache.kafka.common.security.oauthbearer.internals.OAuthBearerRefreshingLogin) [main]
2024-01-17 02:24:02,820 DEBUG [Principal=:2ff1f418-b363-4d8a-b37a-4e506eb2c783]: It is an expiring credential (org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLo     gin) [main]
2024-01-17 02:24:02,844 INFO [Principal=:2ff1f418-b363-4d8a-b37a-4e506eb2c783]: Expiring credential re-login thread started. (org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin) [kafka-expiring-relogin-thread-2ff1f418-b363-4d8a-b37a-4e506eb2c783]
2024-01-17 02:24:02,857 INFO [Principal=2ff1f418-b363-4d8a-b37a-4e506eb2c783]: Expiring credential valid from Wed Jan 17 02:24:02 GMT 2024 to Wed Jan 17 08:24:02 GMT 2024 (org.apache.kafka.common.security.     oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin) [kafka-expiring-relogin-thread-2ff1f418-b363-4d8a-b37a-4e506eb2c783]
2024-01-17 02:24:02,857 INFO [Principal=:2ff1f418-b363-4d8a-b37a-4e506eb2c783]: Expiring credential re-login sleeping until: Wed Jan 17 07:12:14 GMT 2024 (org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin) [kafka-expiring-relogin-thread-2ff1f418-b363-4d8a-b37a-4e506eb2c783]
2024-01-17 02:24:02,870 WARN These configurations '[producer.sasl.jaas.config, group.id, rest.advertised.port, plugin.path, consumer.sasl.mechanism, admin.security.protocol, producer.sasl.login.callback.handler.class, status.storage.replication.factor, offset.storage.topic, consumer.security.protocol, admin.sasl.mechanism, value.converter, key.converter, consumer.client.rack, consumer.sasl.jaas.config, conf     ig.storage.topic, connector.client.config.override.policy, producer.security.protocol, rest.advertised.host.name, status.storage.topic, exactly.once.source.support, producer.sasl.mechanism, rest.port, conf     ig.storage.replication.factor, value.converter.schema.registry.url, consumer.sasl.login.callback.handler.class, admin.sasl.login.callback.handler.class, offset.storage.replication.factor, admin.sasl.jaas.config]' were supplied but are not used yet. (org.apache.kafka.clients.admin.AdminClientConfig) [main]
2024-01-17 02:24:02,870 INFO Kafka version: 3.3.1 (org.apache.kafka.common.utils.AppInfoParser) [main]
2024-01-17 02:24:02,871 INFO Kafka commitId: e23c59d00e687ff5 (org.apache.kafka.common.utils.AppInfoParser) [main]
2024-01-17 02:24:02,871 INFO Kafka startTimeMs: 1705458242870 (org.apache.kafka.common.utils.AppInfoParser) [main]
2024-01-17 02:24:02,873 DEBUG [AdminClient clientId=adminclient-1] Kafka admin client initialized (org.apache.kafka.clients.admin.KafkaAdminClient) [main]
2024-01-17 02:24:02,873 DEBUG Looking up Kafka cluster ID (org.apache.kafka.connect.util.ConnectUtils) [main]
2024-01-17 02:24:02,873 DEBUG [AdminClient clientId=adminclient-1] Thread starting (org.apache.kafka.clients.admin.KafkaAdminClient) [kafka-admin-client-thread | adminclient-1]
2024-01-17 02:24:02,875 DEBUG [AdminClient clientId=adminclient-1] Queueing Call(callName=listNodes, deadlineMs=1705458302874, tries=0, nextAllowedTryMs=0) with a timeout 30000 ms from now. (org.apache.kafka.clients.admin.KafkaAdminClient) [main]
2024-01-17 02:24:02,875 DEBUG Fetching Kafka cluster ID (org.apache.kafka.connect.util.ConnectUtils) [main]
2024-01-17 02:24:02,876 DEBUG Resolved host odf-cluster-kafka-bootstrap as 10.43.99.169 (org.apache.kafka.clients.ClientUtils) [kafka-admin-client-thread | adminclient-1]
2024-01-17 02:24:02,876 DEBUG [AdminClient clientId=adminclient-1] Initiating connection to node odf-cluster-kafka-bootstrap:9093 (id: -1 rack: null) using address odf-cluster-kafka-bootstrap/10.43.99.169      (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-1]
2024-01-17 02:24:02,886 DEBUG [AdminClient clientId=adminclient-1] Set SASL client state to SEND_APIVERSIONS_REQUEST (org.apache.kafka.common.security.authenticator.SaslClientAuthenticator) [kafka-admin-client-thread | adminclient-1]
2024-01-17 02:24:02,887 DEBUG [AdminClient clientId=adminclient-1] Creating SaslClient: client=null;service=kafka;serviceHostname=odf-cluster-kafka-bootstrap;mechs=[OAUTHBEARER] (org.apache.kafka.common.security.authenticator.SaslClientAuthenticator) [kafka-admin-client-thread | adminclient-1]
2024-01-17 02:24:02,890 DEBUG Setting SASL/OAUTHBEARER client state to SEND_CLIENT_FIRST_MESSAGE (org.apache.kafka.common.security.oauthbearer.internals.OAuthBearerSaslClient) [kafka-admin-client-thread | adminclient-1]
2024-01-17 02:24:02,893 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-01-17 02:24:03,103 DEBUG [AdminClient clientId=adminclient-1] Set SASL client state to RECEIVE_APIVERSIONS_RESPONSE (org.apache.kafka.common.security.authenticator.SaslClientAuthenticator) [kafka-admin-client-thread | adminclient-1]
2024-01-17 02:24:03,104 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-01-17 02:24:03,206 DEBUG [AdminClient clientId=adminclient-1] Connection with odf-cluster-kafka-bootstrap/10.43.99.169 (channelId=-1) 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:120)
        at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.receiveResponseOrToken(SaslClientAuthenticator.java:475)
        at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.receiveKafkaResponse(SaslClientAuthenticator.java:572)
        at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.authenticate(SaslClientAuthenticator.java:250)
        at org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:181)
        at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:543)
        at org.apache.kafka.common.network.Selector.poll(Selector.java:481)
        at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:560)
        at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.processRequests(KafkaAdminClient.java:1415)
        at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1346)
        at java.base/java.lang.Thread.run(Thread.java:829)
2024-01-17 02:24:03,207 INFO [AdminClient clientId=adminclient-1] Node -1 disconnected. (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-1]
2024-01-17 02:24:03,209 WARN [AdminClient clientId=adminclient-1] Connection to node -1 (odf-cluster-kafka-bootstrap/10.43.99.169:9093) 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]

Further checking on Kafka broker log, it shows that the authentication failed due to SSL Handshake.

2024-01-18 02:49:34,352 INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Failed authentication with /10.42.2.18 (channelId=10.42.2.236:9093-10.42.2.18:53448-9115) (SSL handshake failed) (org.apache.kafka.common.network.Selector) [data-plane-kafka-network-thread-0-ListenerName(TLS-9093)-SASL_SSL-8]
2024-01-18 02:49:35,266 INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Failed authentication with /10.42.2.18 (channelId=10.42.2.236:9093-10.42.2.18:53456-9115) (SSL handshake failed) (org.apache.kafka.common.network.Selector) [data-plane-kafka-network-thread-0-ListenerName(TLS-9093)-SASL_SSL-9]

So far I'm suspecting that it failed the authentication because it is missing the required certificates. That's is what I'm trying to do in the kafka connect config yaml (specifying the truststore location/password as env variable). But so far it has not work yet. Is there other configuration that I should set instead?

By the way, the truststore.p12 contains these two certs :

1. ca.crt - used by keycloak
2. kafka.crt - I generated using the cluster-ca-cert.

Any suggestions are greatly appreciated.
Thanks.

[scholzj]

The YAMLs you share are not valid Strimzi custom resources. Maybe some kind of your private template? You should share the full Strimzi custom resources so that it is clear how you actually configured Strimzi..

[firdaustahir]

Thank you for the feedback @scholzj . I managed to make it work by adding the following:

tls:  
  trustedCertificates:
  - secretName: kafka-connect-secret
    certificate: kafka.crt

It is now able to connect to broker port 9093.