CODEWITHSUNDEEP
javaapache-flink
yokin
yokin

Reputation: 31

Flink Application throw Class Not Found Exception in Java

I have a Flink Cluster with Yarn, use the flink-quickstart-java Archetype to build a demo project. After building a fat-jar with 'mvn clean package -Pbuild-jar' command, and submit the program with 'flink run -m yarn-cluster -yn 2 ./flink-SNAPSHOT-1.0.jar', the program throw the following exception:

java.lang.NoClassDefFoundError: java.lang.NoClassDefFoundError: org/apache/kafka/common/serialization/ByteArrayDeserializer at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09.setDeserializer(FlinkKafkaConsumer09.java:290) at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09.(FlinkKafkaConsumer09.java:216) at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09.(FlinkKafkaConsumer09.java:154) at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010.(FlinkKafkaConsumer010.java:128) at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010.(FlinkKafkaConsumer010.java:112) at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010.(FlinkKafkaConsumer010.java:79) at stream.TransferKafka.main(TransferKafka.java:19) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:498) at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:525) at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:417) at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:395) at org.apache.flink.client.CliFrontend.executeProgram(CliFrontend.java:828) at org.apache.flink.client.CliFrontend.run(CliFrontend.java:283) at org.apache.flink.client.CliFrontend.parseParameters(CliFrontend.java:1080) at org.apache.flink.client.CliFrontend$1.call(CliFrontend.java:1127) at org.apache.flink.client.CliFrontend$1.call(CliFrontend.java:1124) at java.security.AccessController.doPrivileged(Native Method) at javax.security.auth.Subject.doAs(Subject.java:422) at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1781) at org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41) at org.apache.flink.client.CliFrontend.main(CliFrontend.java:1124) Caused by: java.lang.ClassNotFoundException: org.apache.kafka.common.serialization.ByteArrayDeserializer at java.net.URLClassLoader.findClass(URLClassLoader.java:381) at java.lang.ClassLoader.loadClass(ClassLoader.java:424) at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:338) at java.lang.ClassLoader.loadClass(ClassLoader.java:357) ... 24 more

And Here is my demo:

public static void main(String[] args) {
 final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 Properties props = new Properties();
 props.setProperty("bootstrap.servers", "ip:port");
 props.setProperty("group.id", "NewFlinkTest");
 DataStreamSource < String > stream = env.addSource(new FlinkKafkaConsumer010 < > ("kafka_test", new SimpleStringSchema(), props));
 stream.addSink(new FlinkKafkaProducer010 < > ("kafka_test_out", new SimpleStringSchema(), props));
 try {
  env.execute("Flink Jar Test");
 } catch (Exception e) {
  e.printStackTrace();
 }
}

And some version information:
FLink Version: 1.4.0

Hadoop Version: 2.7.2

Kafka Version: 0.10.2.1

JDK Version: 1.8

Pom dependencies

Edit1:

<?xml version="1.0" encoding="UTF-8"?>
<dependencies>
   <!-- Apache Flink dependencies -->
   <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-core</artifactId>
      <version>${flink.version}</version>
   </dependency>
   <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-java</artifactId>
      <version>${flink.version}</version>
   </dependency>
   <dependency>
      <!-- This dependency is required to actually execute jobs. It is currently pulled in by           flink-streaming-java, but we explicitly depend on it to safeguard against future changes. -->
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-clients_${scala.binary.version}</artifactId>
      <version>${flink.version}</version>
   </dependency>
   <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
      <version>${flink.version}</version>
   </dependency>
   <!-- explicitly add a standard logging framework, as Flink does not have     a hard dependency on one specific framework by default -->
   <dependency>
      <groupId>org.slf4j</groupId>
      <artifactId>slf4j-log4j12</artifactId>
      <version>${slf4j.version}</version>
   </dependency>
   <dependency>
      <groupId>log4j</groupId>
      <artifactId>log4j</artifactId>
      <version>${log4j.version}</version>
   </dependency>
   <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-connector-rabbitmq_2.11</artifactId>
      <version>1.4.0</version>
   </dependency>
   <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-connector-kafka-0.10_${scala.binary.version}</artifactId>
      <version>1.4.0</version>
   </dependency>
</dependencies>

After some attempts, I find the code throws exception is not the same jar that I packed into my uber-jar. I think the main reason is the client server has older version of the flink-connector-kafka jar, but no matter how I set the config yaml property 'yarn.per-job-cluster.include-user-jar', the program always throws the same exception.

Edit2:

After add kafka-clients:0.10.2.1 to flink_home/lib/, it works. But still don't know the reason why it doesn't read class file in uber jar.

Upvotes: 3

Views: 3165

Answers (2)

yiksanchan
yiksanchan

Reputation: 1940

You probably want to add <scope>provided</scope> to flink-streaming-scala, flink-clients, link-table-api-scala-bridge and flink-table-planner-blink - that solves my problem

Upvotes: 0

Victor Wong
Victor Wong

Reputation: 161

First, you may verify if the missing class is in your jar file by grep 'ByteArrayDeserializer' ./flink-SNAPSHOT-1.0.jar.

Upvotes: 0

Related Questions