NoClassDefFoundError when trying to integrate Kafka with Flink

Question

I'm trying to interegate kafka and flink. The idea is to consume a kafka queue and transform data using flink. I'm following below mentioned example

https://github.com/abhishek-ch/evolveML/blob/master/flink/kafka-flink-example/pom.xml

These are my dependencies.

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>1.3.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-core</artifactId>
            <version>0.9.1</version>
           </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_2.11</artifactId>
            <version>1.3.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka-0.8_2.11</artifactId>
            <version>1.3.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka_2.11</artifactId>
            <version>1.0.0</version>
        </dependency>

Also i'm including kafka and flink classes in project as follows.

        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-dependency-plugin</artifactId>
            <version>3.0.2</version>
            <executions>
                <execution>
                    <id>unpack</id>
                    <!-- executed just before the package phase -->
                    <phase>prepare-package</phase>
                    <goals>
                        <goal>unpack</goal>
                    </goals>
                    <configuration>
                        <artifactItems>
                            <!-- For Flink connector classes -->
                            <artifactItem>
                                <groupId>org.apache.flink</groupId>
                                <artifactId>flink-connector-kafka-0.8_2.11</artifactId>
                                <version>1.3.2</version>
                                <type>jar</type>
                                <overWrite>false</overWrite>
                                <outputDirectory>${project.build.directory}/classes</outputDirectory>
                                <includes>org/apache/flink/**</includes>
                            </artifactItem>
                            <!-- For Kafka API classes -->
                            <artifactItem>
                                <groupId>org.apache.kafka</groupId>
                                <artifactId>kafka_2.11</artifactId>
                                <version>1.0.0</version>
                                <type>jar</type>
                                <overWrite>false</overWrite>
                                <outputDirectory>${project.build.directory}/classes</outputDirectory>
                                <includes>kafka/**</includes>
                            </artifactItem>
                        </artifactItems>
                    </configuration>
                </execution>
            </executions>
       </plugin>

My Java code for Consuming kafka queues is

  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    Map<String, String> map = new HashMap<>();
    map.put("bootstrap.servers", kafka_server);
    map.put("zookeeper.connect", "localhost:40862");
    map.put("group.id", "test");
    map.put("topic", "data");

    // parse user parameters
    ParameterTool parameterTool = ParameterTool.fromMap(map);

    DataStream<String> messageStream = null;
    try {
        messageStream = env.addSource(new org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer082<>(
                parameterTool.getRequired("topic"),
                new SimpleStringSchema(),
                parameterTool.getProperties()));
    } catch (Exception e) {
        LOGGER.error("Error", e);
    }

    // print() will write the contents of the stream to the TaskManager's standard out stream
    // the rebelance call is causing a repartitioning of the data so that all machines
    // see the messages (for example in cases when "num kafka partitions" < "num flink operators"
    messageStream.rebalance().map(new MapFunction<String, String>() {
        private static final long serialVersionUID = -6867736771747690202L;

        @Override
        public String map(String value) throws Exception {
            LOGGER.info("============================" + value);
            return "Kafka and Flink says: " + value;
        }
    }).print();

    try {
        env.execute();
    } catch (Exception e) {
        e.printStackTrace();
    }

This code sample is from github project i mentioned earlier. This code runs inside war file deployed in tomcat.

When running this code I get following error.

      Unrecoverable error java.util.concurrent.CompletionException: java.lang.NoClassDefFoundError: org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer082 

I have mentioned classes in war extracts. I'm trying to figure out how to resolve this. Any help or advice highly appreciated.

Answer 1

You should modify the flink-streaming-core dependency to instead be a dependency on flink-streaming-java_2.11, version 1.3.2. (flink-streaming-core was renamed to flink-streaming-java and flink-streaming-scala a couple of years ago.)

Also, flink-connector-kafka-0.8_2.11 is for Kafka version 0.8.x, whereas you are combining it with Kafka version 1.0.0. I suggest you delete the kafka_2.11 dependency and rely on maven to transitively include the correct version of the kafka jar.

Answer 2

A NoClassDefFoundError often hints to version / dependency issues and in fact your dependencies are a bit messed up.

You are importing Flink dependencies from 1.3.2 (current release) and 0.9.1 (a rather ancient version). The Flink Kafka connector is for Kafka 0.8, but you pull in a Kafka 1.0.0 dependency.