Describe the bug

After the upgrade to the Kafka clients 2.8.0 in my CI (my fork) it happens very often that the tests become stuck after this error:

23:13:09.362 [kafka-expiring-relogin-thread-undefined:org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule@318] ERROR org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule - java.lan>
java.io.IOException: java.lang.InterruptedException
at io.streamnative.pulsar.handlers.kop.security.oauth.ClientCredentialsFlow.authenticate(ClientCredentialsFlow.java:89) ~[oauth-client-2.10.0.8-SNAPSHOT.jar:?]
at io.streamnative.pulsar.handlers.kop.security.oauth.OauthLoginCallbackHandler.handleCallback(OauthLoginCallbackHandler.java:82) ~[oauth-client-2.10.0.8-SNAPSHOT.jar:?]
at io.streamnative.pulsar.handlers.kop.security.oauth.OauthLoginCallbackHandler.handle(OauthLoginCallbackHandler.java:67) ~[oauth-client-2.10.0.8-SNAPSHOT.jar:?]
at org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule.identifyToken(OAuthBearerLoginModule.java:316) [kafka-clients-2.8.0.jar:?]
at org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule.login(OAuthBearerLoginModule.java:301) [kafka-clients-2.8.0.jar:?]
at javax.security.auth.login.LoginContext.invoke(LoginContext.java:747) [?:?]
at javax.security.auth.login.LoginContext$4.run(LoginContext.java:672) [?:?]
at javax.security.auth.login.LoginContext$4.run(LoginContext.java:670) [?:?]
at java.security.AccessController.doPrivileged(Native Method) ~[?:?]
at javax.security.auth.login.LoginContext.invokePriv(LoginContext.java:670) [?:?]
at javax.security.auth.login.LoginContext.login(LoginContext.java:581) [?:?]
at org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin.reLogin(ExpiringCredentialRefreshingLogin.java:390) [kafka-clients-2.8.0.jar:?]
at org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin.access$600(ExpiringCredentialRefreshingLogin.java:40) [kafka-clients-2.8.0.jar:?]
at org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin$Refresher.run(ExpiringCredentialRefreshingLogin.java:106) [kafka-clients-2.8.0.jar:?]
at java.lang.Thread.run(Thread.java:829) [?:?]
Caused by: java.lang.InterruptedException
at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:385) ~[?:?]
at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1999) ~[?:?]
at org.asynchttpclient.netty.NettyResponseFuture.get(NettyResponseFuture.java:201) ~[async-http-client-2.12.1.jar:?]
at io.streamnative.pulsar.handlers.kop.security.oauth.ClientCredentialsFlow.authenticate(ClientCredentialsFlow.java:76) ~[oauth-client-2.10.0.8-SNAPSHOT.jar:?]
... 14 more

After that every creation of Producer becomes stuck, and I see this thread dump produced by the test listener:
"TestNG-method=basicRecoveryAbortedTransaction-1" #271 prio=5 os_prio=0 cpu=4.31ms elapsed=20.04s tid=0x00007fde9a04b800 nid=0x4097 waiting for monitor entry [0x00007fde33ad1000]
java.lang.Thread.State: BLOCKED (on object monitor)
at org.apache.kafka.common.security.authenticator.LoginManager.acquireLoginManager(LoginManager.java:100)
- waiting to lock <0x0000000088681498> (a java.lang.Class for org.apache.kafka.common.security.authenticator.LoginManager)
at org.apache.kafka.common.network.SaslChannelBuilder.configure(SaslChannelBuilder.java:170)
at org.apache.kafka.common.network.ChannelBuilders.create(ChannelBuilders.java:192)
at org.apache.kafka.common.network.ChannelBuilders.clientChannelBuilder(ChannelBuilders.java:81)
at org.apache.kafka.clients.ClientUtils.createChannelBuilder(ClientUtils.java:105)
at org.apache.kafka.clients.producer.KafkaProducer.newSender(KafkaProducer.java:448)
at org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:429)
at org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:291)
at org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:318)
at org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:303)
at io.streamnative.pulsar.handlers.kop.TransactionTest.buildTransactionProducer(TransactionTest.java:719)
at io.streamnative.pulsar.handlers.kop.TransactionTest.buildTransactionProducer(TransactionTest.java:701)
at io.streamnative.pulsar.handlers.kop.TransactionTest.basicRecoveryAbortedTransaction(TransactionTest.java:478)
at jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(java.base@11.0.17/Native Method)
at jdk.internal.reflect.NativeMethodAccessorImpl.invoke(java.base@11.0.17/NativeMethodAccessorImpl.java:62)
at jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(java.base@11.0.17/DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(java.base@11.0.17/Method.java:566)
at org.testng.internal.MethodInvocationHelper.invokeMethod(MethodInvocationHelper.java:124)
at org.testng.internal.InvokeMethodRunnable.runOne(InvokeMethodRunnable.java:54)
at org.testng.internal.InvokeMethodRunnable.run(InvokeMethodRunnable.java:44)
at java.util.concurrent.Executors$RunnableAdapter.call(java.base@11.0.17/Executors.java:515)
at java.util.concurrent.FutureTask.run(java.base@11.0.17/FutureTask.java:264)
at java.util.concurrent.ThreadPoolExecutor.runWorker(java.base@11.0.17/ThreadPoolExecutor.java:1128)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(java.base@11.0.17/ThreadPoolExecutor.java:628)
at java.lang.Thread.run(java.base@11.0.17/Thread.java:829)

"TestNG-method=basicRecoveryAbortedTransaction-1" #271 prio=5 os_prio=0 cpu=4.31ms elapsed=20.04s tid=0x00007fde9a04b800 nid=0x4097 waiting for monitor entry [0x00007fde33ad1000]
java.lang.Thread.State: BLOCKED (on object monitor)
at org.apache.kafka.common.security.authenticator.LoginManager.acquireLoginManager(LoginManager.java:100)
- waiting to lock <0x0000000088681498> (a java.lang.Class for org.apache.kafka.common.security.authenticator.LoginManager)
at org.apache.kafka.common.network.SaslChannelBuilder.configure(SaslChannelBuilder.java:170)
at org.apache.kafka.common.network.ChannelBuilders.create(ChannelBuilders.java:192)
at org.apache.kafka.common.network.ChannelBuilders.clientChannelBuilder(ChannelBuilders.java:81)
at org.apache.kafka.clients.ClientUtils.createChannelBuilder(ClientUtils.java:105)
at org.apache.kafka.clients.producer.KafkaProducer.newSender(KafkaProducer.java:448)
at org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:429)
at org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:291)
at org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:318)
at org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:303)
at io.streamnative.pulsar.handlers.kop.TransactionTest.buildTransactionProducer(TransactionTest.java:719)
at io.streamnative.pulsar.handlers.kop.TransactionTest.buildTransactionProducer(TransactionTest.java:701)
at io.streamnative.pulsar.handlers.kop.TransactionTest.basicRecoveryAbortedTransaction(TransactionTest.java:478)
at jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(java.base@11.0.17/Native Method)
at jdk.internal.reflect.NativeMethodAccessorImpl.invoke(java.base@11.0.17/NativeMethodAccessorImpl.java:62)
at jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(java.base@11.0.17/DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(java.base@11.0.17/Method.java:566)
at org.testng.internal.MethodInvocationHelper.invokeMethod(MethodInvocationHelper.java:124)
at org.testng.internal.InvokeMethodRunnable.runOne(InvokeMethodRunnable.java:54)
at org.testng.internal.InvokeMethodRunnable.run(InvokeMethodRunnable.java:44)
at java.util.concurrent.Executors$RunnableAdapter.call(java.base@11.0.17/Executors.java:515)
at java.util.concurrent.FutureTask.run(java.base@11.0.17/FutureTask.java:264)
at java.util.concurrent.ThreadPoolExecutor.runWorker(java.base@11.0.17/ThreadPoolExecutor.java:1128)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(java.base@11.0.17/ThreadPoolExecutor.java:628)
at java.lang.Thread.run(java.base@11.0.17/Thread.java:829)
.....
this happens for all the test methods

To Reproduce
This behaviour happens very often in my fork https://
I am investigating, maybe the 2.8.0 OAuth client implementation has some bugs

Expected behavior
No stuck CI