https://flink.apache.org/ logo
Join Slack
Powered by
Hi, I’m trying to use the Flink 1.16.1 Kafka Conne...
# troubleshooting
a
Hi, I’m trying to use the Flink 1.16.1 Kafka Connector with 
OAUTHBEARER
authentication. I’m configuring the 
KafkaSource
like this:
Copy code
KafkaSource<String> source = KafkaSource.<String>builder()
				.setBootstrapServers(bootstrapServers)
				.setTopics(inputTopic)
				.setStartingOffsets(OffsetsInitializer.earliest())
				.setValueOnlyDeserializer(new SimpleStringSchema())
				.setProperty("security.protocol", "SASL_SSL")
				.setProperty("sasl.login.callback.handler.class", OauthLoginCallbackHandler.class.getName())
				.setProperty("sasl.mechanism", "OAUTHBEARER")
				.setProperty("sasl.jaas.config", "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required"
				+ " oauth.issuer.url=\"%s\""
				+ " oauth.credentials.url=\"%s\""
				+ " oauth.audience=\"%s\";";)
				.build();
When I submit my job I run into the following `NoClassDefFoundError`:
Copy code
2023-02-16 15:54:06,180 ERROR org.apache.flink.runtime.source.coordinator.SourceCoordinator [] - Uncaught exception in the SplitEnumerator for Source Source: Kafka source while starting the SplitEnumerator.. Triggering job failover.
org.apache.kafka.common.KafkaException: Failed to create new KafkaAdminClient
	at org.apache.kafka.clients.admin.KafkaAdminClient.createInternal(KafkaAdminClient.java:546) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.clients.admin.KafkaAdminClient.createInternal(KafkaAdminClient.java:478) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.clients.admin.Admin.create(Admin.java:133) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.clients.admin.AdminClient.create(AdminClient.java:39) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumerator.getKafkaAdminClient(KafkaSourceEnumerator.java:410) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumerator.start(KafkaSourceEnumerator.java:151) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.flink.runtime.source.coordinator.SourceCoordinator.lambda$start$1(SourceCoordinator.java:225) ~[flink-dist-1.16.1.jar:1.16.1]
	at org.apache.flink.runtime.source.coordinator.SourceCoordinator.lambda$runInEventLoop$10(SourceCoordinator.java:449) ~[flink-dist-1.16.1.jar:1.16.1]
	at org.apache.flink.util.ThrowableCatchingRunnable.run(ThrowableCatchingRunnable.java:40) [flink-dist-1.16.1.jar:1.16.1]
	at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source) [?:?]
	at java.util.concurrent.FutureTask.run(Unknown Source) [?:?]
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(Unknown Source) [?:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source) [?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source) [?:?]
	at java.lang.Thread.run(Unknown Source) [?:?]
Caused by: org.apache.kafka.common.KafkaException: java.lang.NoClassDefFoundError: com/fasterxml/jackson/databind/ObjectMapper
	at org.apache.kafka.common.network.SaslChannelBuilder.configure(SaslChannelBuilder.java:184) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.common.network.ChannelBuilders.create(ChannelBuilders.java:192) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.common.network.ChannelBuilders.clientChannelBuilder(ChannelBuilders.java:81) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.clients.ClientUtils.createChannelBuilder(ClientUtils.java:105) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.clients.admin.KafkaAdminClient.createInternal(KafkaAdminClient.java:517) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	... 14 more
Caused by: java.lang.NoClassDefFoundError: com/fasterxml/jackson/databind/ObjectMapper
	at org.apache.kafka.common.security.oauthbearer.internals.unsecured.OAuthBearerUnsecuredJws.toMap(OAuthBearerUnsecuredJws.java:299) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.common.security.oauthbearer.internals.unsecured.OAuthBearerUnsecuredJws.<init>(OAuthBearerUnsecuredJws.java:83) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.common.security.oauthbearer.internals.unsecured.OAuthBearerUnsecuredLoginCallbackHandler.handleTokenCallback(OAuthBearerUnsecuredLoginCallbackHandler.java:211) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.common.security.oauthbearer.internals.unsecured.OAuthBearerUnsecuredLoginCallbackHandler.handle(OAuthBearerUnsecuredLoginCallbackHandler.java:163) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule.identifyToken(OAuthBearerLoginModule.java:316) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule.login(OAuthBearerLoginModule.java:301) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at javax.security.auth.login.LoginContext.invoke(Unknown Source) ~[?:?]
	at javax.security.auth.login.LoginContext$4.run(Unknown Source) ~[?:?]
	at javax.security.auth.login.LoginContext$4.run(Unknown Source) ~[?:?]
	at java.security.AccessController.doPrivileged(Native Method) ~[?:?]
	at javax.security.auth.login.LoginContext.invokePriv(Unknown Source) ~[?:?]
	at javax.security.auth.login.LoginContext.login(Unknown Source) ~[?:?]
	at org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin.login(ExpiringCredentialRefreshingLogin.java:204) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.common.security.oauthbearer.internals.OAuthBearerRefreshingLogin.login(OAuthBearerRefreshingLogin.java:150) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.common.security.authenticator.LoginManager.<init>(LoginManager.java:62) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.common.security.authenticator.LoginManager.acquireLoginManager(LoginManager.java:105) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.common.network.SaslChannelBuilder.configure(SaslChannelBuilder.java:170) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.common.network.ChannelBuilders.create(ChannelBuilders.java:192) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.common.network.ChannelBuilders.clientChannelBuilder(ChannelBuilders.java:81) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.clients.ClientUtils.createChannelBuilder(ClientUtils.java:105) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.clients.admin.KafkaAdminClient.createInternal(KafkaAdminClient.java:517) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	... 14 more
Caused by: java.lang.ClassNotFoundException: com.fasterxml.jackson.databind.ObjectMapper
	at java.net.URLClassLoader.findClass(Unknown Source) ~[?:?]
	at java.lang.ClassLoader.loadClass(Unknown Source) ~[?:?]
	at org.apache.flink.util.FlinkUserCodeClassLoader.loadClassWithoutExceptionHandling(FlinkUserCodeClassLoader.java:67) ~[flink-dist-1.16.1.jar:1.16.1]
	at org.apache.flink.util.ChildFirstClassLoader.loadClassWithoutExceptionHandling(ChildFirstClassLoader.java:74) ~[flink-dist-1.16.1.jar:1.16.1]
	at org.apache.flink.util.FlinkUserCodeClassLoader.loadClass(FlinkUserCodeClassLoader.java:51) ~[flink-dist-1.16.1.jar:1.16.1]
	at java.lang.ClassLoader.loadClass(Unknown Source) ~[?:?]
	at org.apache.kafka.common.security.oauthbearer.internals.unsecured.OAuthBearerUnsecuredJws.toMap(OAuthBearerUnsecuredJws.java:299) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.common.security.oauthbearer.internals.unsecured.OAuthBearerUnsecuredJws.<init>(OAuthBearerUnsecuredJws.java:83) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.common.security.oauthbearer.internals.unsecured.OAuthBearerUnsecuredLoginCallbackHandler.handleTokenCallback(OAuthBearerUnsecuredLoginCallbackHandler.java:211) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.common.security.oauthbearer.internals.unsecured.OAuthBearerUnsecuredLoginCallbackHandler.handle(OAuthBearerUnsecuredLoginCallbackHandler.java:163) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule.identifyToken(OAuthBearerLoginModule.java:316) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule.login(OAuthBearerLoginModule.java:301) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at javax.security.auth.login.LoginContext.invoke(Unknown Source) ~[?:?]
	at javax.security.auth.login.LoginContext$4.run(Unknown Source) ~[?:?]
	at javax.security.auth.login.LoginContext$4.run(Unknown Source) ~[?:?]
	at java.security.AccessController.doPrivileged(Native Method) ~[?:?]
	at javax.security.auth.login.LoginContext.invokePriv(Unknown Source) ~[?:?]
	at javax.security.auth.login.LoginContext.login(Unknown Source) ~[?:?]
	at org.apache.kafka.common.security.oauthbearer.internals.expiring.ExpiringCredentialRefreshingLogin.login(ExpiringCredentialRefreshingLogin.java:204) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.common.security.oauthbearer.internals.OAuthBearerRefreshingLogin.login(OAuthBearerRefreshingLogin.java:150) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.common.security.authenticator.LoginManager.<init>(LoginManager.java:62) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.common.security.authenticator.LoginManager.acquireLoginManager(LoginManager.java:105) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.common.network.SaslChannelBuilder.configure(SaslChannelBuilder.java:170) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.common.network.ChannelBuilders.create(ChannelBuilders.java:192) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.common.network.ChannelBuilders.clientChannelBuilder(ChannelBuilders.java:81) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.clients.ClientUtils.createChannelBuilder(ClientUtils.java:105) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	at org.apache.kafka.clients.admin.KafkaAdminClient.createInternal(KafkaAdminClient.java:517) ~[blob_p-e7cf6c9007e3786b6a409f1c61d4d65e35d634ce-99d0f3b0d7940f398cc7b5441915b1dd:?]
	... 14 more
Any idea why this happens?
s
Looks like it’s unrelated to Flink or Kafka, just an issue with dependencies. I suggest checking dependency graph, there are probably conflicting versions of jackson library.
a
I can’t see 
jackson-databind
in IntelliJs “Analyze Dependencies” view. I only added
Copy code
<dependency>
			<groupId>org.apache.flink</groupId>
			<artifactId>flink-connector-kafka</artifactId>
			<version>${flink.version}</version>
		</dependency>
To the 
flink-java-quickstart
project. Is there anything else I need to add to run the connector?
CC @Martijn Visser
m
Do you also have
Copy code
<dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>${kafka.version}</version>
        </dependency>
a
I just added it with 
compile
scope but still getting the same error
m
Just to be sure, have you reloaded the project after adding it?
a
Sure, I always run 
mvn clean package
then submit the job via the web dashboard. I also tried adding
Copy code
<dependency>			<groupId>com.fasterxml.jackson.core</groupId>
			<artifactId>jackson-databind</artifactId>
			<version>2.9.0</version>
		</dependency>
but it always fails with the same error
I just decompiled the jar file to make sure it’s included and I can see it in there oO
m
Sounds like something is broken 😉
a
Is there an older version of the connector I should try? I recreated my setup using the plain 
KafkaConsumer
locally and there it works
Sorry, I’m retarded 😂 I did not notice the logs got appended and was always looking at the same old log output from the first run. Looks like the classloading issue is solved now
Now I get 
java.lang.NoClassDefFoundError: org/apache/kafka/common/security/oauthbearer/internals/OAuthBearerSaslServer
but if I decompile the jar it is indeed in there
Okay, looks like it is related to this: https://issues.apache.org/jira/browse/FLINK-19132 When I copy the 
kafka-clients.jar
into the JobManager I don’t run into this issue
6 Views
Open in Slack
PreviousNext
Powered by