Daebeom Lee created FLINK-14012:
-----------------------------------
Summary: Failed to start job for consuming Secure Kafka after the
job cancel
Key: FLINK-14012
URL: https://issues.apache.org/jira/browse/FLINK-14012
Project: Flink
Issue Type: Bug
Components: Connectors / Kafka
Affects Versions: 1.9.0
Environment: * Kubernetes 1.13.2
* Flink 1.9.0
* Kafka client libary 2.2.0
Reporter: Daebeom Lee
Hello, this is Daebeom Lee.
h2. Background
I installed Flink 1.9.0 at this our Kubernetes cluster.
We use Flink session cluster. - build fatJar file and upload it at the UI, run
serval jobs.
At first, our jobs are good to start.
But, when we cancel some jobs, the job failed
This is the error code.
{code:java}
// code placeholder
java.lang.NoClassDefFoundError:
org/apache/kafka/common/security/scram/internals/ScramSaslClient
at
org.apache.kafka.common.security.scram.internals.ScramSaslClient$ScramSaslClientFactory.createSaslClient(ScramSaslClient.java:235)
at javax.security.sasl.Sasl.createSaslClient(Sasl.java:384)
at
org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.lambda$createSaslClient$0(SaslClientAuthenticator.java:180)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:422)
at
org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.createSaslClient(SaslClientAuthenticator.java:176)
at
org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.<init>(SaslClientAuthenticator.java:168)
at
org.apache.kafka.common.network.SaslChannelBuilder.buildClientAuthenticator(SaslChannelBuilder.java:254)
at
org.apache.kafka.common.network.SaslChannelBuilder.lambda$buildChannel$1(SaslChannelBuilder.java:202)
at
org.apache.kafka.common.network.KafkaChannel.<init>(KafkaChannel.java:140)
at
org.apache.kafka.common.network.SaslChannelBuilder.buildChannel(SaslChannelBuilder.java:210)
at
org.apache.kafka.common.network.Selector.buildAndAttachKafkaChannel(Selector.java:334)
at
org.apache.kafka.common.network.Selector.registerChannel(Selector.java:325)
at org.apache.kafka.common.network.Selector.connect(Selector.java:257)
at
org.apache.kafka.clients.NetworkClient.initiateConnect(NetworkClient.java:920)
at org.apache.kafka.clients.NetworkClient.ready(NetworkClient.java:287)
at
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.trySend(ConsumerNetworkClient.java:474)
at
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:255)
at
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:236)
at
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:215)
at
org.apache.kafka.clients.consumer.internals.Fetcher.getTopicMetadata(Fetcher.java:292)
at
org.apache.kafka.clients.consumer.KafkaConsumer.partitionsFor(KafkaConsumer.java:1803)
at
org.apache.kafka.clients.consumer.KafkaConsumer.partitionsFor(KafkaConsumer.java:1771)
at
org.apache.flink.streaming.connectors.kafka.internal.KafkaPartitionDiscoverer.getAllPartitionsForTopics(KafkaPartitionDiscoverer.java:77)
at
org.apache.flink.streaming.connectors.kafka.internals.AbstractPartitionDiscoverer.discoverPartitions(AbstractPartitionDiscoverer.java:131)
at
org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.open(FlinkKafkaConsumerBase.java:508)
at
org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:36)
at
org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:102)
at
org.apache.flink.streaming.runtime.tasks.StreamTask.openAllOperators(StreamTask.java:529)
at
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:393)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:705)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:530)
at java.lang.Thread.run(Thread.java:748)
{code}
h2. Our workaround
* I think that this is Flink JVM classloader issue.
* Classloader unloads when job cancels by the way kafka client library is
included fatJar.
* So, I located Kafka client library to /opt/flink/lib
** /opt/flink/lib/kafka-clients-2.2.0.jar
* And then all issue solved.
* But there are weird points
** When Flink 1.8.1 has no problem before 2 weeks
** Before 1 week I rollback from 1.9.0 to 1.8.1, same errors occurred.
** Maybe docker image is changed at docker repository (
[https://github.com/docker-flink/docker-flink )
|https://github.com/docker-flink/docker-flink]
h2. Suggestion
* I'd like to know why this error occurred exactly reason after upgrade 1.9.0.
* Does anybody know a better solution in this case?
Thank you in advance.
--
This message was sent by Atlassian Jira
(v8.3.2#803003)
