[FLINK-19132] Failed to start jobs for consuming Secure Kafka after cluster restart

• Type: Bug
• Status: Open
• Priority: Not a Priority
• Resolution: Unresolved
• Affects Version/s: 1.9.1, 1.10.1
• Component/s: Connectors / Kafka
• Labels: auto-deprioritized-major auto-deprioritized-minor
• Assignee: Unassigned
• Reporter: Olivier Zembri
• Created: 03/Sep/20 11:47
• Updated: 11/Nov/21 22:38

Description

We deploy Flink jobs packaged as fat jar files compiled with Java 1.8 on a Flink session cluster in Kubernetes.

After restarting the Kubernetes cluster, the jobs fail to start and we get several NoClassDefFoundError in the Task Manager log.

Stack trace

java.lang.NoClassDefFoundError: org.apache.kafka.common.security.scram.ScramSaslClient
{
 "class": "org.apache.kafka.common.security.scram.ScramSaslClient",
 "method": "evaluateChallenge",
 "file": "ScramSaslClient.java",
 "line": 128,
 },
 {
 "class": "org.apache.kafka.common.security.authenticator.SaslClientAuthenticator$2",
 "method": "run",
 "file": "SaslClientAuthenticator.java",
 "line": 280,
 },
 {
 "class": "org.apache.kafka.common.security.authenticator.SaslClientAuthenticator$2",
 "method": "run",
 "file": "SaslClientAuthenticator.java",
 "line": 278,
 },
 {
 "class": "java.security.AccessController",
 "method": "doPrivileged",
 "file": "AccessController.java",
 "line": -2,
 },
 {
 "class": "javax.security.auth.Subject",
 "method": "doAs",
 "file": "Subject.java",
 "line": 422,
 },
 {
 "class": "org.apache.kafka.common.security.authenticator.SaslClientAuthenticator",
 "method": "createSaslToken",
 "file": "SaslClientAuthenticator.java",
 "line": 278,
 },
 {
 "class": "org.apache.kafka.common.security.authenticator.SaslClientAuthenticator",
 "method": "sendSaslToken",
 "file": "SaslClientAuthenticator.java",
 "line": 215,
 },
 {
 "class": "org.apache.kafka.common.security.authenticator.SaslClientAuthenticator",
 "method": "authenticate",
 "file": "SaslClientAuthenticator.java",
 "line": 189,
 },
 {
 "class": "org.apache.kafka.common.network.KafkaChannel",
 "method": "prepare",
 "file": "KafkaChannel.java",
 "line": 76,
 },
 {
 "class": "org.apache.kafka.common.network.Selector",
 "method": "pollSelectionKeys",
 "file": "Selector.java",
 "line": 376,
 },
 {
 "class": "org.apache.kafka.common.network.Selector",
 "method": "poll",
 "file": "Selector.java",
 "line": 326,
 },
 {
 "class": "org.apache.kafka.clients.NetworkClient",
 "method": "poll",
 "file": "NetworkClient.java",
 "line": 433,
 },
 {
 "class": "org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient",
 "method": "poll",
 "file": "ConsumerNetworkClient.java",
 "line": 232,
 },
 {
 "class": "org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient",
 "method": "poll",
 "file": "ConsumerNetworkClient.java",
 "line": 208,
 },
 {
 "class": "org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient",
 "method": "poll",
 "file": "ConsumerNetworkClient.java",
 "line": 184,
 },
 {
 "class": "org.apache.kafka.clients.consumer.internals.Fetcher",
 "method": "getTopicMetadata",
 "file": "Fetcher.java",
 "line": 314,
 },
 {
 "class": "org.apache.kafka.clients.consumer.KafkaConsumer",
 "method": "partitionsFor",
 "file": "KafkaConsumer.java",
 "line": 1386,
 },
 {
 "class": "org.apache.flink.streaming.connectors.kafka.internal.Kafka09PartitionDiscoverer",
 "method": "getAllPartitionsForTopics",
 "file": "Kafka09PartitionDiscoverer.java",
 },
 {
 "class": "org.apache.flink.streaming.connectors.kafka.internals.AbstractPartitionDiscoverer",
 "method": "discoverPartitions",
 "file": "AbstractPartitionDiscoverer.java",
 "line": 131,
 },
 {
 "class": "org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase",
 "method": "open",
 "file": "FlinkKafkaConsumerBase.java",
 "line": 508,
 },
 {
 "class": "org.apache.flink.api.common.functions.util.FunctionUtils",
 "method": "openFunction",
 "file": "FunctionUtils.java",
 "line": 36,
 },
 {
 "class": "org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator",
 "method": "open",
 "file": "AbstractUdfStreamOperator.java",
 "line": 102,
 },
 {
 "class": "org.apache.flink.streaming.runtime.tasks.StreamTask",
 "method": "openAllOperators",
 "file": "StreamTask.java",
 "line": 532,
 },
 {
 "class": "org.apache.flink.streaming.runtime.tasks.StreamTask",
 "method": "invoke",
 "file": "StreamTask.java",
 "line": 396,
 },
 {
 "class": "org.apache.flink.runtime.taskmanager.Task",
 "method": "doRun",
 "file": "Task.java",
 "line": 705,
 },
 {
 "class": "org.apache.flink.runtime.taskmanager.Task",
 "method": "run",
 "file": "Task.java",
 "line": 530,
 },
 {
 "class": "java.lang.Thread",
 "method": "run",
 "file": "Thread.java",
 "line": 748,
 }

 

Workaround:

- Copy the jar file containing the missing classes in the /lib folder
      /opt/flink/lib/kafka-clients-0.11.0.jar

• Update the flink-conf.yaml with
  classloader.parent-first-patterns.additional: org.apache.kafka

Note: This issue is very similar to https://issues.apache.org/jira/browse/FLINK-14012.