Error "Cancelled in-flight API_VERSIONS request" when using OAUTHBEARER

[rubenski]

I am having an issue where I can't connect to the Strimzi brokers using OAUTHBEARER protocol.

We use a custom Stimzi image based on kafka:0.35.1-kafka-3.4.0. The custom image allows us to include our custom JAR, see below.

Sorry for the lengthy post. I tried to provide all the relevant details. In the code samples and config below I replaced our domain by example.com.

I use both OAUTHBEARER and SCRAM-SHA-512 under the same custom listener on port 9092. I can connect with Apache (Java) KafkaAdminClient using SCRAM-SHA-512 and username/password, create a topic and I can create a KafkaProducer and KafkaCconsumer and produce and consume messages on this topic.

With OAUTHBEARER however, I keep running into this error:

Cancelled in-flight API_VERSIONS request with correlation id 36 due to node -1 being disconnected

This is the listener config from the Strimzi manifest:

listeners:
  - authentication:
      listenerConfig:
        oauthbearer.connections.max.reauth.ms: 3600000
        oauthbearer.sasl.jaas.config: >
          org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule
          required \
          jwks_uri="https://example.com/openid/connect/jwks.json" \
          unsecuredLoginStringClaim_sub="admin";
        oauthbearer.sasl.server.callback.handler.class: >-
          com.example.TokenValidatorCallbackHandler
        sasl.enabled.mechanisms: 'OAUTHBEARER,SCRAM-SHA-512'
        scram-sha-512.sasl.jaas.config: org.apache.kafka.common.security.scram.ScramLoginModule required;
      sasl: true
      type: custom
    configuration:
      bootstrap:
        annotations:
          external-dns.alpha.kubernetes.io/hostname: kafka-bootstrap.strimzi11-3d9y.example.com
        labels:
          protocol: tcp
      brokerCertChainAndKey:
        certificate: tls.crt
        key: tls.key
        secretName: broker-cert
      brokers:
        - advertisedHost: broker-0.strimzi11-3d9y.example.com
          annotations:
            external-dns.alpha.kubernetes.io/hostname: broker-0.strimzi11-3d9y.example.com
            metallb.universe.tf/address-pool: metallb-ip-address-pool
          broker: 0
          labels:
            protocol: tcp
        - advertisedHost: broker-1.strimzi11-3d9y.example.com
          annotations:
            external-dns.alpha.kubernetes.io/hostname: broker-1.strimzi11-3d9y.example.com
            metallb.universe.tf/address-pool: metallb-ip-address-pool
          broker: 1
          labels:
            protocol: tcp
        - advertisedHost: broker-2.strimzi11-3d9y.example.com
          annotations:
            external-dns.alpha.kubernetes.io/hostname: broker-2.strimzi11-3d9y.example.com
            metallb.universe.tf/address-pool: metallb-ip-address-pool
          broker: 2
          labels:
            protocol: tcp
    name: sdsext
    port: 9092
    tls: true
    type: loadbalancer

The class TokenValidatorCallbackHandler is our custom JAR. It is an implementation of AuthenticateCallbackHandler. The JAR validates the incoming token that our client obtains from our identity provider. It does this using public keys it obtains from the same identity provider. (I am aware this scenario is supported out of the box by Strimzi, but for reasons beyond my control this is currently our solution).

I am somewhat confident our TokenValidatorCallbackHandler JAR is doing what it is supposed to. Its handle() method is below.
The jwkProvider is built earlier with the jwks URL. I am confident the jwkProvider is created correctly, so not including its creation.

  @Override
  public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
      Arrays.stream(callbacks)
              .filter(cb -> cb instanceof OAuthBearerValidatorCallback)
              .map(cb -> (OAuthBearerValidatorCallback) cb)
              .forEach(cb -> {
                  try {
                      var token = new OurCustomOAuthBearerToken(cb.tokenValue());
                      var jwk = jwkProvider.get(token.keyId());
                      var verifier = JWT.require(getAlgorithm(jwk))
                              .acceptExpiresAt(60) // 60 seconds = 1 minute
                              .acceptIssuedAt(60)
                              .build();
                      var result = verifier.verify(cb.tokenValue());
                      LOGGER.info("Token verified: {}", result.getToken());
                      // Return token to callback?
                      cb.token(token);
                  } catch (Exception e) {
                      LOGGER.error("Could not validate token", e);
                      cb.error("TokenValidationError", String.format("Could not validate token. Reason: %s", e.getMessage()), null);
                  }
              });
  }

When I run our Java admin / producer code (see below), in the log I see no exceptions and I see the "Token verified" LOG statement, which would not happen if the token could not be verified.

2023-09-12 08:36:00,848 INFO Token verified: eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1Ni.....

I also check the logged token and it is indeed coming from our IDP. I am quite confident the fetching of the token by the client (code below) and the verification of the token by our custom JAR work OK.

I hope I am doing the right thing by returning the new token at cb.token(token); at the end. This is still a question mark for me.

If indeed I am supposed to return the token to the callback, then I think the JAR is working as designed and should not be the cause of the problem.

The sdsext / 9092 custom listener config from /tmp/strimzi.properties on the broker looks like this:

##########
# Listener configuration: SDSEXT-9092
##########
listener.name.sdsext-9092.oauthbearer.connections.max.reauth.ms=3600000
listener.name.sdsext-9092.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
jwks_uri="https://example.com/openid/connect/jwks.json" \
unsecuredLoginStringClaim_sub="admin";

listener.name.sdsext-9092.oauthbearer.sasl.server.callback.handler.class=com.example.TokenValidatorCallbackHandler
listener.name.sdsext-9092.sasl.enabled.mechanisms=OAUTHBEARER,SCRAM-SHA-512
listener.name.sdsext-9092.scram-sha-512.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required;
listener.name.sdsext-9092.ssl.keystore.location=/tmp/kafka/custom-sdsext-9092.keystore.p12
listener.name.sdsext-9092.ssl.keystore.password=edited
listener.name.sdsext-9092.ssl.keystore.type=PKCS12

The common listener config:

##########
# Common listener configuration
##########
listeners=CONTROLPLANE-9090://0.0.0.0:9090,REPLICATION-9091://0.0.0.0:9091,SDSEXT-9092://0.0.0.0:9092,PLAIN-9095://0.0.0.0:9095,TLS-9093://0.0.0.0:9093,OAUTHPLAIN-9094://0.0.0.0:9094
advertised.listeners=CONTROLPLANE-9090://my-cluster-kafka-0.my-cluster-kafka-brokers.strimzi11-3d9y.svc:9090,REPLICATION-9091://my-cluster-kafka-0.my-cluster-kafka-brokers.strimzi11-3d9y.svc:9091,SDSEXT-9092://broker-0.strimzi11-3d9y.example.com:9092,PLAIN-9095://my-cluster-kafka-0.my-cluster-kafka-brokers.strimzi11-3d9y.svc.local:9095,TLS-9093://my-cluster-kafka-0.my-cluster-kafka-brokers.strimzi11-3d9y.svc.local:9093,OAUTHPLAIN-9094://my-cluster-kafka-0.my-cluster-kafka-brokers.strimzi11-3d9y.svc.local:9094
listener.security.protocol.map=CONTROLPLANE-9090:SSL,REPLICATION-9091:SSL,SDSEXT-9092:SASL_SSL,PLAIN-9095:SSL,TLS-9093:SASL_SSL,OAUTHPLAIN-9094:SASL_SSL
control.plane.listener.name=CONTROLPLANE-9090
inter.broker.listener.name=REPLICATION-9091
sasl.enabled.mechanisms=
ssl.endpoint.identification.algorithm=HTTPS

The sasl.enabled.mechanisms is empty in the common listener config. Is this by design? Could this be causing the issue? Can I set it via the Strimzi manifest somehow?

Then the main() method from my test client called ExampleProducer below. Our test client uses kafka-clients 3.4.0 (same Kafka version as our Strimzi)

  public static void main(String[] args) throws IOException, InterruptedException, ExecutionException {

      var properties = new Properties();
      try (var stream = ExampleProducer.class.getResourceAsStream("/oauth.properties")) {
          properties.load(stream);
      }

      try (var client = KafkaAdminClient.create(properties)) {
          // LINE BELOW FAILS
          var nodes = client.describeCluster()
                  .nodes()
                  .get();
          nodes.forEach(n -> LOGGER.info("Found node: {}", n.id()));
          LOGGER.info("Creating topic '{}'", TEST_TOPIC);
          var testTopic = new NewTopic(TEST_TOPIC, 3, (short) 3);
          client.createTopics(List.of(testTopic));
          client.listTopics().listings().get().forEach(n -> LOGGER.info("Found topic: {}", n));
      } catch (ExecutionException | InterruptedException e) {
          throw new RuntimeException(e);
      }
  }

This code fails on client.describeCluster().nodes().get(). Full log output from running this:

10:18:18.458 [main] INFO  org.apache.kafka.clients.admin.AdminClientConfig:376 - AdminClientConfig values: 
	auto.include.jmx.reporter = true
	bootstrap.servers = [kafka-bootstrap.strimzi11-3d9y.example.com: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 = class org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler
	sasl.login.class = null
	sasl.login.connect.timeout.ms = null
	sasl.login.read.timeout.ms = 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.login.retry.backoff.max.ms = 10000
	sasl.login.retry.backoff.ms = 100
	sasl.mechanism = OAUTHBEARER
	sasl.oauthbearer.clock.skew.seconds = 30
	sasl.oauthbearer.expected.audience = null
	sasl.oauthbearer.expected.issuer = null
	sasl.oauthbearer.jwks.endpoint.refresh.ms = 3600000
	sasl.oauthbearer.jwks.endpoint.retry.backoff.max.ms = 10000
	sasl.oauthbearer.jwks.endpoint.retry.backoff.ms = 100
	sasl.oauthbearer.jwks.endpoint.url = null
	sasl.oauthbearer.scope.claim.name = scope
	sasl.oauthbearer.sub.claim.name = client_id
	sasl.oauthbearer.token.endpoint.url = https://example.com/auth/oauth/v2/token
	security.protocol = SASL_SSL
	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 = ./src/main/resources/trust.jks
	ssl.truststore.password = [hidden]
	ssl.truststore.type = JKS

10:18:18.659 [main] INFO  org.apache.kafka.common.config.AbstractConfig:376 - AbstractConfig values: 
	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

10:18:19.160 [main] INFO  o.a.k.c.s.o.i.e.ExpiringCredentialRefreshingLogin:205 - Successfully logged in.
10:18:19.162 [kafka-expiring-relogin-thread-98c92df2-87a7-4dcf-b420-d5386e3f0690] INFO  o.a.k.c.s.o.i.e.ExpiringCredentialRefreshingLogin:72 - [Principal=:98c92df2-87a7-4dcf-b420-d5386e3f0690]: Expiring credential re-login thread started.
10:18:19.163 [kafka-expiring-relogin-thread-98c92df2-87a7-4dcf-b420-d5386e3f0690] INFO  o.a.k.c.s.o.i.e.ExpiringCredentialRefreshingLogin:318 - [Principal=98c92df2-87a7-4dcf-b420-d5386e3f0690]: Expiring credential valid from Tue Sep 12 10:18:19 CEST 2023 to Tue Sep 12 11:18:19 CEST 2023
10:18:19.169 [kafka-expiring-relogin-thread-98c92df2-87a7-4dcf-b420-d5386e3f0690] INFO  o.a.k.c.s.o.i.e.ExpiringCredentialRefreshingLogin:91 - [Principal=:98c92df2-87a7-4dcf-b420-d5386e3f0690]: Expiring credential re-login sleeping until: Tue Sep 12 11:08:49 CEST 2023
10:18:19.233 [main] INFO  org.apache.kafka.common.utils.AppInfoParser:119 - Kafka version: 3.4.0
10:18:19.233 [main] INFO  org.apache.kafka.common.utils.AppInfoParser:120 - Kafka commitId: 2e1947d240607d53
10:18:19.233 [main] INFO  org.apache.kafka.common.utils.AppInfoParser:121 - Kafka startTimeMs: 1694506699232
10:18:19.501 [kafka-admin-client-thread | adminclient-1] INFO  org.apache.kafka.clients.NetworkClient:937 - [AdminClient clientId=adminclient-1] Node -1 disconnected.
10:18:19.502 [kafka-admin-client-thread | adminclient-1] INFO  org.apache.kafka.clients.NetworkClient:341 - [AdminClient clientId=adminclient-1] Cancelled in-flight API_VERSIONS request with correlation id 0 due to node -1 being disconnected (elapsed time since creation: 14ms, elapsed time since send: 14ms, request timeout: 3600000ms)
10:18:19.682 [kafka-admin-client-thread | adminclient-1] INFO  org.apache.kafka.clients.NetworkClient:937 - [AdminClient clientId=adminclient-1] Node -1 disconnected.
10:18:19.682 [kafka-admin-client-thread | adminclient-1] INFO  org.apache.kafka.clients.NetworkClient:341 - [AdminClient clientId=adminclient-1] Cancelled in-flight API_VERSIONS request with correlation id 1 due to node -1 being disconnected (elapsed time since creation: 12ms, elapsed time since send: 12ms, request timeout: 3600000ms)
10:18:19.863 [kafka-admin-client-thread | adminclient-1] INFO  org.apache.kafka.clients.NetworkClient:937 - [AdminClient clientId=adminclient-1] Node -1 disconnected.
10:18:19.863 [kafka-admin-client-thread | adminclient-1] INFO  org.apache.kafka.clients.NetworkClient:341 - [AdminClient clientId=adminclient-1] Cancelled in-flight API_VERSIONS request with correlation id 2 due to node -1 being disconnected (elapsed time since creation: 11ms, elapsed time since send: 11ms, request timeout: 3600000ms)
10:18:20.250 [kafka-admin-client-thread | adminclient-1] INFO  org.apache.kafka.clients.NetworkClient:937 - [AdminClient clientId=adminclient-1] Node -1 disconnected.

Finally, the content of the oauth.properties file that the ExampleProducer loads properties from:

bootstrap.servers=kafka-bootstrap.strimzi11-3d9y.example.com:9092


# SASL OAUTHBEARER config
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
    scope='edited' \
    clientId='edited' \
    clientSecret='edited';

sasl.mechanism=OAUTHBEARER

sasl.oauthbearer.sub.claim.name=client_id
sasl.login.callback.handler.class=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler
sasl.oauthbearer.token.endpoint.url=https://example.com/auth/oauth/v2/token

#SSL configs
security.protocol=SASL_SSL

ssl.truststore.location=./src/main/resources/trust.jks
ssl.truststore.password=edited

Some things I already figured out:

• This is not a network issue. Brokers URLs are accessible.
• Switching to a broker URL or bootstrap URL produces the same result.
• The error messages logged by org.apache.kafka.clients.NetworkClient are coming from its processDisconnection() method. The diconnectState.state() is READY. So none of the switch options are matched. https://github.com/a0x8o/kafka/blob/54eff6af115ee647f60129f2ce6a044cb17215d0/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java#L760.

[scholzj]

The sasl.enabled.mechanisms is empty in the common listener config. Is this by design? Could this be causing the issue? Can I set it via the Strimzi manifest somehow?

This is by design as SASL is not used on the internal listeners. Your listener's SASL configuration is under the listener prefix.

[rubenski]

Thanks for the quick reply, @scholzj.

Trace logging on my Kafka client yielded a little bit more info.

14:51:15.474 [kafka-admin-client-thread | adminclient-1] TRACE org.apache.kafka.clients.admin.KafkaAdminClient:1080 - [AdminClient clientId=adminclient-1] Trying to choose nodes for [Call(callName=listNodes, deadlineMs=1694523126795, tries=0, nextAllowedTryMs=0), Call(callName=fetchMetadata, deadlineMs=1694523096793, tries=0, nextAllowedTryMs=0)] at 1694523075474
14:51:15.474 [kafka-admin-client-thread | adminclient-1] TRACE o.a.k.clients.admin.internals.AdminMetadataManager:155 - [AdminClient clientId=adminclient-1] Metadata is not ready: we have not fetched metadata from the bootstrap nodes yet.
14:51:15.475 [kafka-admin-client-thread | adminclient-1] TRACE org.apache.kafka.clients.admin.KafkaAdminClient:1109 - [AdminClient clientId=adminclient-1] Unable to assign Call(callName=listNodes, deadlineMs=1694523126795, tries=0, nextAllowedTryMs=0) to a node.
14:51:15.475 [kafka-admin-client-thread | adminclient-1] TRACE org.apache.kafka.clients.NetworkClient:705 - [AdminClient clientId=adminclient-1] Removing node broker-0.strimzi11-3d9y.example.com:9092 (id: -1 rack: null) from least loaded node selection since it is neither ready for sending or connecting
14:51:15.475 [kafka-admin-client-thread | adminclient-1] TRACE org.apache.kafka.clients.NetworkClient:722 - [AdminClient clientId=adminclient-1] Least loaded node selection failed to find an available node

The AdminClient is unable to fetch metadata from the cluster and ultimately times out trying, even though the nodes are accessible and requests arrive in our JAR.

[scholzj]

I'm afraid I replied to the only point I had some input on. And given you seem to be using a very custom setup, I have no idea what might be the cause of the issue. Sorry.

[rubenski]

I still don't know what caused this issue, but I managed to circumvent it by falling back to creating a listener with regular oauth configuration (type: oauth) and using Strimzi's JaasServerOauthValidatorCallbackHandler instead of our (unnecessary) custom class. Authentication now works and the client is able to fetch metadata from the cluster.

Authentication config of the listener (I provided the jwks Urls both in regular config and in JAAS config. Still need to remove one of those):

authentication:
  sasl: true
  type: oauth
  clientId: kafka-broker
  clientSecret:
    key: secret
    secretName: oauth-client-secret
  validIssuerUri: https://example.com
  jwksEndpointUri: https://example.com/path/to/jwks.json
  checkAccessTokenType: false
  userNameClaim: client_id
  tlsTrustedCertificates:
    - secretName: our-idp-ca
      certificate: our-idp-ca.pem
  listenerConfig:
    oauthbearer.sasl.server.callback.handler.class: io.strimzi.kafka.oauth.server.JaasServerOauthValidatorCallbackHandler
    oauthbearer.connections.max.reauth.ms: 3600000
    # For details about this JAAS config read https://github.com/strimzi/strimzi-kafka-oauth#configuring-the-jaas-login-module
    oauthbearer.sasl.jaas.config: |
      org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
      oauth.jwks.endpoint.uri="https://example.com/path/to/jwks.json" \
      oauth.valid.issuer.uri="https://example.com" \
      oauth.jwks.refresh.seconds=300 \
      oauth.jwks.expiry.seconds=360 \
      unsecuredLoginStringClaim_sub="unused";
    sasl.enabled.mechanisms: OAUTHBEARER