Native build fails with java.lang.NoSuchMethodException: org.apache.kafka.common.security.authenticator.SaslClientCallbackHandler

[weltonrodrigo]

In what version(s) of Spring for Apache Kafka are you seeing this issue?
spring-kafka:3.0.4
kafka-clients:3.3.2

Describe the bug

Tryied to create a native springboot with kafka using microsoft azure eventhub Kafka layer. That uses SASL PLAIN.

Run as docker image build with mvn -DskipTests -Pnative,nativeTest spring-boot:build-image

Failure to create KafkaConsumer Caused by: org.apache.kafka.common.KafkaException: org.apache.kafka.common.KafkaException: Could not find a public no-argument constructor for org.apache.kafka.common.security.authenticator.SaslClientCallbackHandler

To Reproduce

Using the following application.properties:

spring.kafka.bootstrap-servers=<redacted>.servicebus.windows.net:9093
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.security.protocol=SASL_SSL
spring.kafka.jaas.enabled=true
spring.kafka.properties.security.protocol=SASL_SSL
spring.kafka.properties.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="Endpoint=sb://<redacted>.servicebus.windows.net/;SharedAccessKeyName=<redacted>;SharedAccessKey=<redacted>";
spring.kafka.properties.sasl.mechanism=PLAIN
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer
spring.kafka.consumer.max-poll-records=1000

Expected behavior

With the normal (non-native), the app runs normally.

Maybe related:
#2545

Stacktrace

org.springframework.context.ApplicationContextException: Failed to start bean 'org.springframework.kafka.config.internalKafkaListenerEndpointRegistry'
        at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:181)
        at org.springframework.context.support.DefaultLifecycleProcessor$LifecycleGroup.start(DefaultLifecycleProcessor.java:356)
        at java.base@17.0.6/java.lang.Iterable.forEach(Iterable.java:75)
        at org.springframework.context.support.DefaultLifecycleProcessor.startBeans(DefaultLifecycleProcessor.java:155)
        at org.springframework.context.support.DefaultLifecycleProcessor.onRefresh(DefaultLifecycleProcessor.java:123)
        at org.springframework.context.support.AbstractApplicationContext.finishRefresh(AbstractApplicationContext.java:934)
        at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:587)
        at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.refresh(ServletWebServerApplicationContext.java:146)
        at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:732)
        at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:434)
        at org.springframework.boot.SpringApplication.run(SpringApplication.java:310)
        at org.springframework.boot.SpringApplication.run(SpringApplication.java:1304)
        at org.springframework.boot.SpringApplication.run(SpringApplication.java:1293)
        at br.gov.prf.websocket.WebsocketApplication.main(WebsocketApplication.java:12)
Caused by: org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
        at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:830)
        at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:666)
        at org.springframework.kafka.core.DefaultKafkaConsumerFactory.createRawConsumer(DefaultKafkaConsumerFactory.java:483)
        at org.springframework.kafka.core.DefaultKafkaConsumerFactory.createKafkaConsumer(DefaultKafkaConsumerFactory.java:451)
        at org.springframework.kafka.core.DefaultKafkaConsumerFactory.createConsumerWithAdjustedProperties(DefaultKafkaConsumerFactory.java:427)
        at org.springframework.kafka.core.DefaultKafkaConsumerFactory.createKafkaConsumer(DefaultKafkaConsumerFactory.java:394)
        at org.springframework.kafka.core.DefaultKafkaConsumerFactory.createConsumer(DefaultKafkaConsumerFactory.java:371)
        at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.<init>(KafkaMessageListenerContainer.java:852)
        at org.springframework.kafka.listener.KafkaMessageListenerContainer.doStart(KafkaMessageListenerContainer.java:381)
        at org.springframework.kafka.listener.AbstractMessageListenerContainer.start(AbstractMessageListenerContainer.java:531)
        at org.springframework.kafka.listener.ConcurrentMessageListenerContainer.doStart(ConcurrentMessageListenerContainer.java:226)
        at org.springframework.kafka.listener.AbstractMessageListenerContainer.start(AbstractMessageListenerContainer.java:531)
        at org.springframework.kafka.config.KafkaListenerEndpointRegistry.startIfNecessary(KafkaListenerEndpointRegistry.java:383)
        at org.springframework.kafka.config.KafkaListenerEndpointRegistry.start(KafkaListenerEndpointRegistry.java:328)
        at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:178)
        ... 13 common frames omitted
Caused by: org.apache.kafka.common.KafkaException: org.apache.kafka.common.KafkaException: Could not find a public no-argument constructor for org.apache.kafka.common.security.authenticator.SaslClientCallbackHandler
        at org.apache.kafka.common.network.SaslChannelBuilder.configure(SaslChannelBuilder.java:184)
        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.consumer.KafkaConsumer.<init>(KafkaConsumer.java:738)
        ... 27 common frames omitted
Caused by: org.apache.kafka.common.KafkaException: Could not find a public no-argument constructor for org.apache.kafka.common.security.authenticator.SaslClientCallbackHandler
        at org.apache.kafka.common.utils.Utils.newInstance(Utils.java:394)
        at org.apache.kafka.common.network.SaslChannelBuilder.createClientCallbackHandler(SaslChannelBuilder.java:305)
        at org.apache.kafka.common.network.SaslChannelBuilder.configure(SaslChannelBuilder.java:148)
        ... 31 common frames omitted
Caused by: java.lang.NoSuchMethodException: org.apache.kafka.common.security.authenticator.SaslClientCallbackHandler.<init>()
        at java.base@17.0.6/java.lang.Class.getConstructor0(DynamicHub.java:3585)
        at java.base@17.0.6/java.lang.Class.getDeclaredConstructor(DynamicHub.java:2754)
        at org.apache.kafka.common.utils.Utils.newInstance(Utils.java:392)
        ... 33 common frames omitted

[garyrussell]

While we provided some hints to get a basic Spring for Apache Kafka application running as a native image, we (Spring) can't take on the responsibility of making all aspects of the kafka-clients native compatible.

You can add your own hints or reproduce the issue without Spring and raise an issue over there.

If you get it working with your own hints, we'd consider accepting a PR, if you want to contribute it.

[artembilan]

I'd prefer to have Apache Kafka client-specific native hints exactly in Apache Kafka.
Spring does nothing with that reflection around org.apache.kafka.common.security.authenticator.SaslClientCallbackHandler, so it is unfair to ask us to fix something what really is out of this project scope.
Sure! It is easy to have a RuntimeHintsRegistrar implementation for Spring AOT, but it is already a target project feature if you can't convince Apache Kafka to add native images support.

[matthenry87]

I've got my Spring Kafka consumers compiling as native image. There is a core Java security hint needed in my case which I am narrowing down from the dump I got from the tracing agent.

The kicker for me right now is the ErrorHandlingDeserializer, since it serializes exceptions into the header. I plan to implement a version that just stores the stack trace instead.

[garyrussell]

that just stores the stack trace instead.

The DeadLetterPublishingRecoverer (or any recoverer that wants to analyze the problem) needs access to the raw data (byte[]) that couldn't be deserialized (currently a field on the DeserializationException).

I suppose you could add that as a separate header and add code to the DLPR to get it from there, if present.

Feel free to submit a PR with these changes.

[matthenry87]

that just stores the stack trace instead.

The DeadLetterPublishingRecoverer needs access to the raw data (byte[]) that couldn't be deserialized (currently a field on the DeserializationException.

I suppose you could add that as a separate header and add code to the DLPR to get it from there, if present.

Feel free to submit a PR with these changes.

Ah interesting I see now. I'll take a look!

[matthenry87]

I've got this working and we've been using it in production all year. I want to submit a PR but when I went to write the tests to submit the PR for spring-kafka, I realized it's not going to be simple.

Can't try and fake-out the static variable that indicates whether we're running in native or not because of the way tests start up. I think I'll have to actually make the tests run in native.

[sobychacko]

@matthenry87 What changes are we talking about here? Is this still about the original report (SaslClientCallbackHandler) or something else? Did you add any native hints that are specific to Spring Kafka? If it is some native-specific features that you are trying to test, we have a dedicated project for that: https://github.com/spring-projects/spring-aot-smoke-tests/tree/main/integration/spring-kafka. Maybe you can test your native features via the smoke tests? But we would like to see where the underlying changes are happening. Thanks!

[matthenry87]

@matthenry87 What changes are we talking about here? Is this still about the original report (SaslClientCallbackHandler) or something else? Did you add any native hints that are specific to Spring Kafka? If it is some native-specific features that you are trying to test, we have a dedicated project for that: https://github.com/spring-projects/spring-aot-smoke-tests/tree/main/integration/spring-kafka. Maybe you can test your native features via the smoke tests? But we would like to see where the underlying changes are happening. Thanks!

My changes are the ones mentioned above that enable a consumer running native. They do include a hint for the original report, and we're even using IAM authentication with MSK.

I'll have a look at those tests.