Hi everyone! We’re running a Flink 1.19.3 app on AWS EKS, using flink-connector-kafka:3.3.0-1.19 for KafkaSource and Sink. The Kafka cluster is hosted on Confluent Cloud. Issue: When the JobManager is replaced (e.g., during failover), the new JobManager starts fine, but then TaskManagers begin throwing exceptions related to

SaslClientAuthenticator

failing to configure. Relevant configuration: • Added

classloader.parent-first-patterns.additional

per Confluent Cloud support to avoid child-first classloading issues with OAuth:

org.apache.kafka.common.security.oauthbearer.;
org.apache.kafka.common.security.auth.;
org.apache.kafka.common.security.plain.;
org.apache.kafka.common.security.scram.

• Both

sasl.login.callback.handler.class

and

sasl.client.callback.handler.class

are set to:

org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler

Runtime classpath (Kafka-related):

+--- io.confluent:kafka-avro-serializer:7.4.12
|    +--- io.confluent:kafka-schema-serializer:7.4.12
|    |    +--- io.confluent:kafka-schema-registry-client:7.4.12
|    +--- io.confluent:kafka-schema-registry-client:7.4.12 (*)
+--- org.apache.flink:flink-connector-kafka:3.3.0-1.19
|    +--- org.apache.kafka:kafka-clients:3.4.0

Stacktrace excerpt:

Caused by: 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.OAuthBearerLoginCallbackHandler

Has anyone encountered this issue or have suggestions on how to resolve it? I'll add a thread and include a snippet with the full stacktrace.