Hi, I have a small question about oauth in flink. Our kafka cluster uses OAUTHBEARER through a custom implementation, when I try to deploy it I get the following error in the flink job

Caused by: <http://org.apache.flink.kafka.shaded.org|org.apache.flink.kafka.shaded.org>.apache.kafka.common.KafkaException: java.lang.ClassCastException: com.***.datahub.oauth.oauthbearer.OAuthAuthenticateLoginCallbackHandler cannot be cast to <http://org.apache.flink.kafka.shaded.org|org.apache.flink.kafka.shaded.org>.apache.kafka.common.security.auth.AuthenticateCallbackHandler

I found online that you need to supply the kafka client libs as part of the docker img (we are using the operator) and they are present, yet this still happens. My create table stmt

CREATE TABLE engagement_v1(

client_id ROW<id INT>,

count_of_positions INT

) WITH (

'connector' = 'kafka',

'topic' = 'engagement-notification_triggers-crypto_position_notification-log-v1',

'properties.bootstrap.servers' = '***:9092',

'properties.security.protocol' = 'SASL_SSL',

'properties.sasl.mechanism' = 'OAUTHBEARER',

'properties.sasl.jaas.config' = '***',

'properties.sasl.login.callback.handler.class' = 'com.***.datahub.oauth.oauthbearer.OAuthAuthenticateLoginCallbackHandler',

'properties.group.id' = 'python-demo-stream-flink',

'scan.startup.mode' = 'earliest-offset',

'format' = 'protobuf',

'protobuf.message-class-name' = 'iit.datahub.engagement.notification_triggers.v1.OpenCryptoPositionNotificationOuterClass$OpenCryptoPositionNotification',

'protobuf.ignore-parse-errors' = 'true'

)

Docker img is based on flink:1.17.1-scala_2.12-java8 and the following libs are copied into the docker img

RUN wget -P /opt/flink/lib/ <https://artifacts.cf>.***/artifactory/maven/org/apache/flink/flink-json/${FLINK_VERSION}/flink-json-${FLINK_VERSION}.jar; \

wget -P /opt/flink/lib/ <https://artifacts.cf>.***/artifactory/maven/org/apache/flink/flink-sql-connector-kafka/${FLINK_VERSION}/flink-sql-connector-kafka-${FLINK_VERSION}.jar; \

wget -P /opt/flink/lib/ <https://artifacts.cf>.***/artifactory/maven/org/apache/flink/flink-sql-connector-elasticsearch7/${FLINK_VERSION}/flink-sql-connector-elasticsearch7-${FLINK_VERSION}.jar; \

wget -P /opt/flink/lib/ <https://artifacts.cf>.***/artifactory/maven/org/apache/flink/flink-protobuf/${FLINK_VERSION}/flink-protobuf-${FLINK_VERSION}.jar; \

wget -P /opt/flink/lib/ <https://artifacts.cf>.***/artifactory/maven/org/apache/kafka/kafka-clients/${KAFKA_CLIENT_VERSION}/kafka-clients-${KAFKA_CLIENT_VERSION}.jar; \

It seems

com.***.datahub.oauth.oauthbearer.OAuthAuthenticateLoginCallbackHandler

is not implementing the expected interfaces

The same class works just fine with Kafka (been running for the past 3 years)

Flink is shading the interface. Carefully read which class it is trying to cast to

<http://org.apache.flink.kafka.shaded.org|org.apache.flink.kafka.shaded.org>.apache.kafka.common.security.auth.AuthenticateCallbackHandler