[
https://issues.apache.org/jira/browse/KAFKA-14102?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18112197#comment-18112197
]
Seokmo Yoo commented on KAFKA-14102:
------------------------------------
This still reproduces on trunk.
Below is the mechanism and the fix I would propose.
h3. Cause
{{Sasl.createSaslClient}} resolves a {{SaslClientFactory}} through the JVM-wide
security provider registry, and {{Security.addProvider}} keeps only the first
provider registered under a given name. {{OAuthBearerSaslClientProvider}} is
registered when {{OAuthBearerLoginModule}} initializes, so with Kafka loaded in
more than one class loader the registered provider belongs to whichever class
loader got there first. That factory cannot use a callback handler from any
other class loader:
{code:java}
Callback handler must be castable to
org.apache.kafka.common.security.auth.AuthenticateCallbackHandler:
org.apache.kafka.common.security.oauthbearer.internals.OAuthBearerSaslClientCallbackHandler
{code}
The two class names in that message are identical, which is what makes the
error confusing. They are the same name loaded by two different class loaders,
so the cast fails. Only the first application in the process can authenticate.
Class loader isolation cannot fix this on its own. The registry is
{{{}java.security.Security{}}}, and there is exactly one per JVM no matter how
the
application is isolated. Making the Kafka classes parent-first avoids the
problem by leaving only one copy of them, but that is a deployment constraint
rather than a fix.
h3. Proposed fix
# {{OAuthBearerSaslClientFactory}} returns null instead of throwing when the
callback handler is not usable. {{SaslClientFactory}} is specified to return
null when it cannot create a client, and throwing stops {{Sasl}} from trying the
remaining providers.
# {{SaslClientAuthenticator}} falls back to the factory of its own class loader
when the registry yields no client, for the mechanisms Kafka provides. The
registry is consulted first, so a provider configured through
{{security.providers}} still takes precedence.
The fallback only covers a path that fails today with
{{{}SaslAuthenticationException("Failed to create SaslClient with mechanism
..."){}}},
so it cannot break a configuration that works now. SCRAM is unaffected: its
factory passes the callback handler through without a cast.
h3. Testing
* {{{}OAuthBearerSaslClientTest{}}}: 4 tests, including
{{createSaslClientReturnsNullForUnusableCallbackHandler}}
* {{{}SaslAuthenticatorTest{}}}: 79 tests, including
{{testSaslClientFromLocalFactory}}
No failures, and checkstyle passes. Reverting the factory change makes the first
new test fail.
h3. One open question
Consulting the registry first and falling back locally preserves
{{security.providers}} overrides, at the cost of reaching the local factory only
after the registry has already failed. Reversing that order would be simpler to
reason about but would drop the override. I can switch if reviewers prefer it.
I will attach a pull request. Happy to take the assignment if that is useful.
> (SASL/OAUTHBEARER) multiple applications in one JVM process, only the first
> started app can consume messages
> ------------------------------------------------------------------------------------------------------------
>
> Key: KAFKA-14102
> URL: https://issues.apache.org/jira/browse/KAFKA-14102
> Project: Kafka
> Issue Type: Bug
> Components: clients
> Affects Versions: 3.0.1
> Reporter: Shuo Chen
> Priority: Blocker
> Attachments: image-2022-10-12-08-39-01-004.png,
> image-2022-10-12-08-43-57-597.png
>
>
> We have 2 web applications (A and B) will consume messages from the same
> Kafka Server, so they have the same configurations:
> {code:java}
> security.protocol=SASL_SSL
> sasl.mechanism=OAUTHBEARER
> sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule
> required;
> sasl.login.callback.handler.class=MyOauth2AuthenticateCallbackHandler
> jaas.enabled=true{code}
>
> A and B deployed together in one Tomcat server (means they are in JVM
> process), startup sequential is A -> B, then we find B cannot consume the
> message with following exception:
> {code:java}
> [2022-07-22 02:52:45,184] [ INFO] 6
> [org.springframework.kafka.KafkaListenerEndpointContainer#5-0-C-1]
> o.a.k.c.n.SaslChannelBuilder - - [Consumer
> clientId=consumer-XXX-7d8650290c70c1fc3da6305099bde64c-1,
> groupId=XXX-7d8650290c70c1fc3da6305099bde64c] Failed to create channel due to
> org.apache.kafka.common.errors.SaslAuthenticationException: Failed to
> configure SaslClientAuthenticator
> Caused by: java.lang.IllegalArgumentException: Callback handler must be
> castable to
> org.apache.kafka.common.security.auth.AuthenticateCallbackHandler:
> org.apache.kafka.common.security.oauthbearer.internals.OAuthBearerSaslClientCallbackHandler
> at
> org.apache.kafka.common.security.oauthbearer.internals.OAuthBearerSaslClient$OAuthBearerSaslClientFactory.createSaslClient(OAuthBearerSaslClient.java:182)
> ~[kafka-clients-3.0.1.jar:?]
> at javax.security.sasl.Sasl.createSaslClient(Sasl.java:420) ~[?:1.8.0_332]
> at
> org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.lambda$createSaslClient$0(SaslClientAuthenticator.java:219)
> ~[kafka-clients-3.0.1.jar:?]
> ... suppressed 2 lines
> at
> org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.createSaslClient(SaslClientAuthenticator.java:215)
> ~[kafka-clients-3.0.1.jar:?]
> at
> org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.<init>(SaslClientAuthenticator.java:206)
> ~[kafka-clients-3.0.1.jar:?]
> at
> org.apache.kafka.common.network.SaslChannelBuilder.buildClientAuthenticator(SaslChannelBuilder.java:286)
> ~[kafka-clients-3.0.1.jar:?]
> at
> org.apache.kafka.common.network.SaslChannelBuilder.lambda$buildChannel$1(SaslChannelBuilder.java:228)
> ~[kafka-clients-3.0.1.jar:?]
> at org.apache.kafka.common.network.KafkaChannel.<init>(KafkaChannel.java:143)
> ~[kafka-clients-3.0.1.jar:?]
> at
> org.apache.kafka.common.network.SaslChannelBuilder.buildChannel(SaslChannelBuilder.java:236)
> ~[kafka-clients-3.0.1.jar:?]
> at
> org.apache.kafka.common.network.Selector.buildAndAttachKafkaChannel(Selector.java:338)
> ~[kafka-clients-3.0.1.jar:?]
> at
> org.apache.kafka.common.network.Selector.registerChannel(Selector.java:329)
> ~[kafka-clients-3.0.1.jar:?]
> at org.apache.kafka.common.network.Selector.connect(Selector.java:256)
> ~[kafka-clients-3.0.1.jar:?]
> at
> org.apache.kafka.clients.NetworkClient.initiateConnect(NetworkClient.java:981)
> ~[kafka-clients-3.0.1.jar:?]
> at org.apache.kafka.clients.NetworkClient.access$600(NetworkClient.java:73)
> ~[kafka-clients-3.0.1.jar:?]
> at
> org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.maybeUpdate(NetworkClient.java:1152)
> ~[kafka-clients-3.0.1.jar:?]
> at
> org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.maybeUpdate(NetworkClient.java:1040)
> ~[kafka-clients-3.0.1.jar:?]
> at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:549)
> ~[kafka-clients-3.0.1.jar:?]
> at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:265)
> ~[kafka-clients-3.0.1.jar:?]
> at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:236)
> ~[kafka-clients-3.0.1.jar:?]
> at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:227)
> ~[kafka-clients-3.0.1.jar:?]
> at
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.awaitMetadataUpdate(ConsumerNetworkClient.java:164)
> ~[kafka-clients-3.0.1.jar:?]
> at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:258)
> ~[kafka-clients-3.0.1.jar:?]
> at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:483)
> ~[kafka-clients-3.0.1.jar:?]
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1262)
> ~[kafka-clients-3.0.1.jar:?]
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1231)
> ~[kafka-clients-3.0.1.jar:?]
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1211)
> ~[kafka-clients-3.0.1.jar:?]
> at
> org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollConsumer(KafkaMessageListenerContainer.java:1522)
> ~[spring-kafka-2.8.6.jar:2.8.6]
> at
> org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doPoll(KafkaMessageListenerContainer.java:1512)
> ~[spring-kafka-2.8.6.jar:2.8.6]
> at
> org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1340)
> ~[spring-kafka-2.8.6.jar:2.8.6]
> at
> org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1252)
> ~[spring-kafka-2.8.6.jar:2.8.6]{code}
>
> I have also debugged into the issue and found the key error is:
> {code:java}
> Caused by: java.lang.IllegalArgumentException: Callback handler must be
> castable to
> org.apache.kafka.common.security.auth.AuthenticateCallbackHandler:
> org.apache.kafka.common.security.oauthbearer.internals.OAuthBearerSaslClientCallbackHandler
> {code}
> Application A start up first, it register OAuthBearerSaslClientProvider in
> the static block of OAuthBearerLoginModule class, So the
> OAuthBearerSaslClientProvider is loaded and initialized by
> ParallelWebappClassLoader for Application A.
> {code:java}
> static {
> OAuthBearerSaslClientProvider.initialize(); // not part of public API
> OAuthBearerSaslServerProvider.initialize(); // not part of public API
> } {code}
> Application B then start up, it try to register OAuthBearerSaslClientProvider
> again with same code, but it won't have any effect because the provider was
> already registered by Application A.
> When application B try to create the SaslClient, AuthenticateCallbackHandler
> class was loaded from the one for Application A , while
> OAuthBearerSaslClientCallbackHandler class was loaded from the one for
> Application B.
>
>
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
