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!