[BUG] Spring Cloud Stream Azure Kafka with Managed Identity credential refresh not working

[jPrest]

Describe the bug
When using spring-cloud-stream with Event Hub Kafka API and the Azure identity support described here credential is not refreshing and eventually the kafka-client stops consuming data.

Exception or Stack Trace

Log leading up to the problem.

// log direction new to old

// consumer stopping due to missing authentication
2022-08-30T14:24:03.941Z ERROR org.springframework.kafka.listener.KafkaMessageListenerContainer Error while stopping the container: 
2022-08-30T14:24:03.937Z ERROR org.apache.kafka.clients.consumer.KafkaConsumer [Consumer clientId=REDACTED, groupId=REDACTED] Failed to close coordinator
2022-08-30T14:24:03.936Z ERROR org.apache.kafka.clients.NetworkClient [Consumer clientId=REDACTED, groupId=REDACTED] Connection to node 0 (REDACTED) failed authentication due to: An error: (java.security.PrivilegedActionException: javax.security.sasl.SaslException: No OAuth Bearer tokens in Subject's private credentials [Caused by java.io.IOException: No OAuth Bearer tokens in Subject's private credentials]) occurred when evaluating SASL token received from the Kafka Broker. Kafka Client will go to AUTHENTICATION_FAILED state.
2022-08-30T14:24:03.774Z ERROR org.springframework.kafka.listener.KafkaMessageListenerContainer Fatal consumer exception; stopping container
2022-08-30T14:24:03.772Z ERROR org.springframework.kafka.listener.KafkaMessageListenerContainer Authentication/Authorization Exception and no authExceptionRetryInterval set
2022-08-30T14:24:03.772Z ERROR org.apache.kafka.clients.NetworkClient [REDACTED, groupId=REDACTED] Connection to node 0 (REDACTED) failed authentication due to: An error: (java.security.PrivilegedActionException: javax.security.sasl.SaslException: No OAuth Bearer tokens in Subject's private credentials [Caused by java.io.IOException: No OAuth Bearer tokens in Subject's private credentials]) occurred when evaluating SASL token received from the Kafka Broker. Kafka Client will go to AUTHENTICATION_FAILED state.
2022-08-30T14:24:00.333Z ERROR org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin No Expiring Credential after a supposedly-successful re-login

// errors indicating refresh did not properly work
2022-08-30T10:21:46.249Z ERROR org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin No Expiring Credential after a supposedly-successful re-login
2022-08-30T10:20:36.247Z ERROR org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin No Expiring Credential after a supposedly-successful re-login
2022-08-30T10:05:22.559Z ERROR org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin No Expiring Credential after a supposedly-successful re-login
2022-08-30T09:54:37.529Z ERROR org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin No Expiring Credential after a supposedly-successful re-login

// loging working fine
2022-08-29T14:28:02.39Z INFO org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin [Principal=:null]: Expiring credential re-login sleeping until: Tue Aug 30 10:48:06 UTC 2022
2022-08-29T14:28:02.39Z INFO org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin [Principal=null]: Expiring credential valid from Mon Aug 29 14:19:54 UTC 2022 to Tue Aug 30 14:24:54 UTC 2022
2022-08-29T14:28:02.389Z INFO org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin [Principal=:null]: Expiring credential re-login thread started.
2022-08-29T14:28:02.389Z INFO org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin Successfully logged in.
2022-08-29T14:28:02.388Z INFO com.azure.identity.ChainedTokenCredential Azure Identity => Attempted credential ManagedIdentityCredential returns a token
2022-08-29T14:28:02.388Z INFO com.azure.identity.ManagedIdentityCredential Azure Identity => getToken() result for scopes [REDACTED]: SUCCESS
2022-08-29T14:28:02.388Z INFO com.azure.identity.ManagedIdentityCredential Azure Identity => Managed Identity environment: AZURE VM IMDS ENDPOINT
2022-08-29T14:28:02.377Z INFO com.azure.identity.ChainedTokenCredential Azure Identity => Attempted credential EnvironmentCredential is unavailable.

Stack Trace with Exception

org.apache.kafka.common.errors.SaslAuthenticationException: An error: (java.security.PrivilegedActionException: javax.security.sasl.SaslException: No OAuth Bearer tokens in Subject's private credentials [Caused by java.io.IOException: No OAuth Bearer tokens in Subject's private credentials]) occurred when evaluating SASL token received from the Kafka Broker. Kafka Client will go to AUTHENTICATION_FAILED state. Caused by: javax.security.sasl.SaslException: No OAuth Bearer tokens in Subject's private credentials at org.apache.kafka.common.security.oauthbearer.internals.OAuthBearerSaslClient.evaluateChallenge(OAuthBearerSaslClient.java:120) at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.lambda$createSaslToken$1(SaslClientAuthenticator.java:534) at java.base/java.security.AccessController.doPrivileged(AccessController.java:712) at java.base/javax.security.auth.Subject.doAs(Subject.java:439) at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.createSaslToken(SaslClientAuthenticator.java:534) at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.sendSaslClientToken(SaslClientAuthenticator.java:433) at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.sendInitialToken(SaslClientAuthenticator.java:332) at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.authenticate(SaslClientAuthenticator.java:298) 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.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:265) at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:236) at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1297) at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1238) at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1211) at jdk.internal.reflect.GeneratedMethodAccessor98.invoke(Unknown Source) at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.base/java.lang.reflect.Method.invoke(Method.java:568) at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:344) at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:208) at jdk.proxy2/jdk.proxy2.$Proxy129.poll(Unknown Source) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollConsumer(KafkaMessageListenerContainer.java:1529) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doPoll(KafkaMessageListenerContainer.java:1519) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1343) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1255) 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.lang.Thread.run(Thread.java:833) Caused by: java.io.IOException: No OAuth Bearer tokens in Subject's private credentials at org.apache.kafka.common.security.oauthbearer.internals.OAuthBearerSaslClientCallbackHandler.handleCallback(OAuthBearerSaslClientCallbackHandler.java:104) at org.apache.kafka.common.security.oauthbearer.internals.OAuthBearerSaslClientCallbackHandler.handle(OAuthBearerSaslClientCallbackHandler.java:83) at org.apache.kafka.common.security.oauthbearer.internals.OAuthBearerSaslClient.evaluateChallenge(OAuthBearerSaslClient.java:92) ... 29 common frames omitted

To Reproduce
Deploy application according to https://docs.microsoft.com/en-us/azure/developer/java/spring-framework/configure-spring-cloud-stream-binder-java-app-kafka-azure-event-hub#update-configuration and leave it running for more than 24h.

Code Snippet
not applicable

Expected behavior
The credential refresh should work correctly and so the kafka-consumer can keep running for more than 24 hours.

Screenshots

The application keeps running, but the kafka consumer (and corresponding metrics) is gone after 24 hours.

Setup (please complete the following information):

• OS: [AKS with Dockerfile Base Image openjdk:17-alpine]
• IDE: [IntelliJ (for development)]
• Library/Libraries: [
  com.azure.spring:spring-cloud-azure-starter:4.3.0
  com.fasterxml.jackson.module:jackson-module-kotlin:2.13.3
  org.springframework.boot:spring-boot-starter-web:2.7.2
  org.springframework.cloud:spring-cloud-stream-binder-kafka:3.2.4
  org.springframework.cloud:spring-cloud-stream:3.2.4
  org.springframework.kafka:spring-kafka:2.8.8 (*)
  ]
• Java version: [17]
• App Server/Environment: [Spring Boot embedded Tomcat]
• Frameworks: [Spring Boot]

Additional context
The application is deployed in two separate AKS environments and I encounter the issue on both.

Information Checklist
Kindly make sure that you have added all the following information above and checkoff the required fields otherwise we will treat the issuer as an incomplete report

• Bug Description Added
• Repro Steps Added
• Setup information Added

[yiliuTo]

@jPrest thanks for your reporting, we will fix it soon.

[yiliuTo]

@jPrest the bug has been fixed in #30883, you can look forward to our next release which is planned in September. Before that, you can disable the pre-refresh to avoid this bug, with the below properties,

spring.cloud.stream.kafka.binder.configuration:
  sasl.login.refresh:
    window:
      factor: 1.0
      jitter: 0
    min.period.seconds: 0
    buffer.seconds: 0

[jPrest]

Hey @yiliuTo thank you very much for this very fast fix!
I'll be trying your proposed workaround and let you know about the results.

[yiliuTo]

Hi @jPrest the release of com.azure.spring:spring-cloud-azure-starter:4.4.0 which carries this fix has been released now, you can upgrade to the latest one. Feel free to reopen this issue if you have other concerns.