title: Flink with Kafka connection
body: <p>Environment:</p>
  
  <pre><code>Ubuntu 16.04.1 LTS
  Flink 1.1.3
  Kakfa 0.10.1.1
  </code></pre>
  
  <p>I'm trying to connect flink with kafka (Flink 1.1.3 Kakfa 0.10.1.1)
  I already try all the fixes that i could find, but none of them work.</p>
  
  <p><strong>pom.xml :</strong></p>
  
  <pre><code>&lt;project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"&gt;
  &lt;modelVersion&gt;4.0.0&lt;/modelVersion&gt;
  &lt;groupId&gt;ux&lt;/groupId&gt;
  &lt;artifactId&gt;logs&lt;/artifactId&gt;
  &lt;version&gt;1.3-SNAPSHOT&lt;/version&gt;
  &lt;packaging&gt;jar&lt;/packaging&gt;
  &lt;name&gt;Flink Quickstart Job&lt;/name&gt;
  &lt;url&gt;http://www.myorganization.org&lt;/url&gt;
  &lt;properties&gt;
      &lt;project.build.sourceEncoding&gt;UTF-8&lt;/project.build.sourceEncoding&gt;
      &lt;flink.version&gt;1.3-SNAPSHOT&lt;/flink.version&gt;
      &lt;slf4j.version&gt;1.7.7&lt;/slf4j.version&gt;
      &lt;log4j.version&gt;1.2.17&lt;/log4j.version&gt;
  &lt;/properties&gt;
  &lt;repositories&gt;
      &lt;repository&gt;
          &lt;id&gt;apache.snapshots&lt;/id&gt;
          &lt;name&gt;Apache Development Snapshot Repository&lt;/name&gt;
          &lt;url&gt;https://repository.apache.org/content/repositories/snapshots/&lt;/url&gt;
          &lt;releases&gt;
              &lt;enabled&gt;false&lt;/enabled&gt;
          &lt;/releases&gt;
          &lt;snapshots&gt;
              &lt;enabled&gt;true&lt;/enabled&gt;
          &lt;/snapshots&gt;
      &lt;/repository&gt;
  &lt;/repositories&gt;
  &lt;dependencies&gt;
      &lt;dependency&gt;
          &lt;groupId&gt;org.apache.flink&lt;/groupId&gt;
          &lt;artifactId&gt;flink-java&lt;/artifactId&gt;
          &lt;version&gt;${flink.version}&lt;/version&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
          &lt;groupId&gt;org.apache.flink&lt;/groupId&gt;
          &lt;artifactId&gt;flink-streaming-java_2.11&lt;/artifactId&gt;
          &lt;version&gt;${flink.version}&lt;/version&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
          &lt;groupId&gt;org.apache.flink&lt;/groupId&gt;
          &lt;artifactId&gt;flink-clients_2.11&lt;/artifactId&gt;
          &lt;version&gt;${flink.version}&lt;/version&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
          &lt;groupId&gt;org.apache.flink&lt;/groupId&gt;
          &lt;artifactId&gt;flink-connector-kafka-0.10_2.10&lt;/artifactId&gt;
          &lt;version&gt;1.3-SNAPSHOT&lt;/version&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
          &lt;groupId&gt;org.slf4j&lt;/groupId&gt;
          &lt;artifactId&gt;slf4j-log4j12&lt;/artifactId&gt;
          &lt;version&gt;${slf4j.version}&lt;/version&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
          &lt;groupId&gt;log4j&lt;/groupId&gt;
          &lt;artifactId&gt;log4j&lt;/artifactId&gt;
          &lt;version&gt;${log4j.version}&lt;/version&gt;
      &lt;/dependency&gt;
  &lt;/dependencies&gt;
  &lt;profiles&gt;
      &lt;profile&gt;
          &lt;id&gt;build-jar&lt;/id&gt;
          &lt;activation&gt;
              &lt;activeByDefault&gt;false&lt;/activeByDefault&gt;
          &lt;/activation&gt;
          &lt;dependencies&gt;
              &lt;dependency&gt;
                  &lt;groupId&gt;org.apache.flink&lt;/groupId&gt;
                  &lt;artifactId&gt;flink-java&lt;/artifactId&gt;
                  &lt;version&gt;${flink.version}&lt;/version&gt;
                  &lt;scope&gt;provided&lt;/scope&gt;
              &lt;/dependency&gt;
              &lt;dependency&gt;
                  &lt;groupId&gt;org.apache.flink&lt;/groupId&gt;
                  &lt;artifactId&gt;flink-streaming-java_2.11&lt;/artifactId&gt;
                  &lt;version&gt;${flink.version}&lt;/version&gt;
                  &lt;scope&gt;provided&lt;/scope&gt;
              &lt;/dependency&gt;
              &lt;dependency&gt;
                  &lt;groupId&gt;org.apache.flink&lt;/groupId&gt;
                  &lt;artifactId&gt;flink-clients_2.11&lt;/artifactId&gt;
                  &lt;version&gt;${flink.version}&lt;/version&gt;
                  &lt;scope&gt;provided&lt;/scope&gt;
              &lt;/dependency&gt;
              &lt;dependency&gt;
                  &lt;groupId&gt;org.slf4j&lt;/groupId&gt;
                  &lt;artifactId&gt;slf4j-log4j12&lt;/artifactId&gt;
                  &lt;version&gt;${slf4j.version}&lt;/version&gt;
                  &lt;scope&gt;provided&lt;/scope&gt;
              &lt;/dependency&gt;
              &lt;dependency&gt;
                  &lt;groupId&gt;log4j&lt;/groupId&gt;
                  &lt;artifactId&gt;log4j&lt;/artifactId&gt;
                  &lt;version&gt;${log4j.version}&lt;/version&gt;
                  &lt;scope&gt;provided&lt;/scope&gt;
              &lt;/dependency&gt;
          &lt;/dependencies&gt;
          &lt;build&gt;
              &lt;plugins&gt;
                  &lt;plugin&gt;
                      &lt;groupId&gt;org.apache.maven.plugins&lt;/groupId&gt;
                      &lt;artifactId&gt;maven-shade-plugin&lt;/artifactId&gt;
                      &lt;version&gt;2.4.1&lt;/version&gt;
                      &lt;executions&gt;
                          &lt;execution&gt;
                              &lt;phase&gt;package&lt;/phase&gt;
                              &lt;goals&gt;
                                  &lt;goal&gt;shade&lt;/goal&gt;
                              &lt;/goals&gt;
                              &lt;configuration&gt;
                                  &lt;artifactSet&gt;
                                      &lt;excludes combine.self="override"&gt;&lt;/excludes&gt;
                                  &lt;/artifactSet&gt;
                              &lt;/configuration&gt;
                          &lt;/execution&gt;
                      &lt;/executions&gt;
                  &lt;/plugin&gt;
              &lt;/plugins&gt;
          &lt;/build&gt;
      &lt;/profile&gt;
  &lt;/profiles&gt;
  &lt;build&gt;
      &lt;plugins&gt;
          &lt;plugin&gt;
              &lt;groupId&gt;org.apache.maven.plugins&lt;/groupId&gt;
              &lt;artifactId&gt;maven-shade-plugin&lt;/artifactId&gt;
              &lt;version&gt;2.4.1&lt;/version&gt;
              &lt;executions&gt;
                  &lt;execution&gt;
                      &lt;phase&gt;package&lt;/phase&gt;
                      &lt;goals&gt;
                          &lt;goal&gt;shade&lt;/goal&gt;
                      &lt;/goals&gt;
                      &lt;configuration&gt;
                          &lt;artifactSet&gt;
                              &lt;excludes&gt;
                                  &lt;exclude&gt;org.apache.flink:flink-annotations&lt;/exclude&gt;
                                  &lt;exclude&gt;org.apache.flink:flink-shaded-hadoop2&lt;/exclude&gt;
                                  &lt;exclude&gt;org.apache.flink:flink-shaded-curator-recipes&lt;/exclude&gt;
                                  &lt;exclude&gt;org.apache.flink:flink-core&lt;/exclude&gt;
                                  &lt;exclude&gt;org.apache.flink:flink-scala_2.11&lt;/exclude&gt;
                                  &lt;exclude&gt;org.apache.flink:flink-runtime_2.11&lt;/exclude&gt;
                                  &lt;exclude&gt;org.apache.flink:flink-optimizer_2.11&lt;/exclude&gt;
                                  &lt;exclude&gt;org.apache.flink:flink-clients_2.11&lt;/exclude&gt;
                                  &lt;exclude&gt;org.apache.flink:flink-avro_2.11&lt;/exclude&gt;
                                  &lt;exclude&gt;org.apache.flink:flink-examples-batch_2.11&lt;/exclude&gt;
                                  &lt;exclude&gt;org.apache.flink:flink-examples-streaming_2.11&lt;/exclude&gt;
                                  &lt;exclude&gt;org.apache.flink:flink-streaming-scala_2.11&lt;/exclude&gt;
                                  &lt;exclude&gt;org.apache.flink:flink-scala-shell_2.11&lt;/exclude&gt;
                                  &lt;exclude&gt;org.apache.flink:flink-python&lt;/exclude&gt;
                                  &lt;exclude&gt;org.apache.flink:flink-metrics-core&lt;/exclude&gt;
                                  &lt;exclude&gt;org.apache.flink:flink-metrics-jmx&lt;/exclude&gt;
                                  &lt;exclude&gt;org.apache.flink:flink-statebackend-rocksdb_2.11&lt;/exclude&gt;
                                  &lt;exclude&gt;log4j:log4j&lt;/exclude&gt;
                                  &lt;exclude&gt;org.scala-lang:scala-library&lt;/exclude&gt;
                                  &lt;exclude&gt;org.scala-lang:scala-compiler&lt;/exclude&gt;
                                  &lt;exclude&gt;org.scala-lang:scala-reflect&lt;/exclude&gt;
                                  &lt;exclude&gt;com.data-artisans:flakka-actor_*&lt;/exclude&gt;
                                  &lt;exclude&gt;com.data-artisans:flakka-remote_*&lt;/exclude&gt;
                                  &lt;exclude&gt;com.data-artisans:flakka-slf4j_*&lt;/exclude&gt;
                                  &lt;exclude&gt;io.netty:netty-all&lt;/exclude&gt;
                                  &lt;exclude&gt;io.netty:netty&lt;/exclude&gt;
                                  &lt;exclude&gt;commons-fileupload:commons-fileupload&lt;/exclude&gt;
                                  &lt;exclude&gt;org.apache.avro:avro&lt;/exclude&gt;
                                  &lt;exclude&gt;commons-collections:commons-collections&lt;/exclude&gt;
                                  &lt;exclude&gt;org.codehaus.jackson:jackson-core-asl&lt;/exclude&gt;
                                  &lt;exclude&gt;org.codehaus.jackson:jackson-mapper-asl&lt;/exclude&gt;
                                  &lt;exclude&gt;com.thoughtworks.paranamer:paranamer&lt;/exclude&gt;
                                  &lt;exclude&gt;org.xerial.snappy:snappy-java&lt;/exclude&gt;
                                  &lt;exclude&gt;org.apache.commons:commons-compress&lt;/exclude&gt;
                                  &lt;exclude&gt;org.tukaani:xz&lt;/exclude&gt;
                                  &lt;exclude&gt;com.esotericsoftware.kryo:kryo&lt;/exclude&gt;
                                  &lt;exclude&gt;com.esotericsoftware.minlog:minlog&lt;/exclude&gt;
                                  &lt;exclude&gt;org.objenesis:objenesis&lt;/exclude&gt;
                                  &lt;exclude&gt;com.twitter:chill_*&lt;/exclude&gt;
                                  &lt;exclude&gt;com.twitter:chill-java&lt;/exclude&gt;
                                  &lt;exclude&gt;commons-lang:commons-lang&lt;/exclude&gt;
                                  &lt;exclude&gt;junit:junit&lt;/exclude&gt;
                                  &lt;exclude&gt;org.apache.commons:commons-lang3&lt;/exclude&gt;
                                  &lt;exclude&gt;org.slf4j:slf4j-api&lt;/exclude&gt;
                                  &lt;exclude&gt;org.slf4j:slf4j-log4j12&lt;/exclude&gt;
                                  &lt;exclude&gt;log4j:log4j&lt;/exclude&gt;
                                  &lt;exclude&gt;org.apache.commons:commons-math&lt;/exclude&gt;
                                  &lt;exclude&gt;org.apache.sling:org.apache.sling.commons.json&lt;/exclude&gt;
                                  &lt;exclude&gt;commons-logging:commons-logging&lt;/exclude&gt;
                                  &lt;exclude&gt;commons-codec:commons-codec&lt;/exclude&gt;
                                  &lt;exclude&gt;com.fasterxml.jackson.core:jackson-core&lt;/exclude&gt;
                                  &lt;exclude&gt;com.fasterxml.jackson.core:jackson-databind&lt;/exclude&gt;
                                  &lt;exclude&gt;com.fasterxml.jackson.core:jackson-annotations&lt;/exclude&gt;
                                  &lt;exclude&gt;stax:stax-api&lt;/exclude&gt;
                                  &lt;exclude&gt;com.typesafe:config&lt;/exclude&gt;
                                  &lt;exclude&gt;org.uncommons.maths:uncommons-maths&lt;/exclude&gt;
                                  &lt;exclude&gt;com.github.scopt:scopt_*&lt;/exclude&gt;
                                  &lt;exclude&gt;commons-io:commons-io&lt;/exclude&gt;
                                  &lt;exclude&gt;commons-cli:commons-cli&lt;/exclude&gt;
                              &lt;/excludes&gt;
                          &lt;/artifactSet&gt;
                          &lt;filters&gt;
                              &lt;filter&gt;
                                  &lt;artifact&gt;org.apache.flink:*&lt;/artifact&gt;
                                  &lt;excludes&gt;
                                      &lt;exclude&gt;org/apache/flink/shaded/com/**&lt;/exclude&gt;
                                      &lt;exclude&gt;web-docs/**&lt;/exclude&gt;
                                  &lt;/excludes&gt;
                              &lt;/filter&gt;
                              &lt;filter&gt;
                                  &lt;artifact&gt;*:*&lt;/artifact&gt;
                                  &lt;excludes&gt;
                                      &lt;exclude&gt;META-INF/*.SF&lt;/exclude&gt;
                                      &lt;exclude&gt;META-INF/*.DSA&lt;/exclude&gt;
                                      &lt;exclude&gt;META-INF/*.RSA&lt;/exclude&gt;
                                  &lt;/excludes&gt;
                              &lt;/filter&gt;
                          &lt;/filters&gt;
                          &lt;createDependencyReducedPom&gt;false&lt;/createDependencyReducedPom&gt;
                      &lt;/configuration&gt;
                  &lt;/execution&gt;
              &lt;/executions&gt;
          &lt;/plugin&gt;
          &lt;plugin&gt;
              &lt;groupId&gt;org.apache.maven.plugins&lt;/groupId&gt;
              &lt;artifactId&gt;maven-compiler-plugin&lt;/artifactId&gt;
              &lt;version&gt;3.1&lt;/version&gt;
              &lt;configuration&gt;
                  &lt;source&gt;1.8&lt;/source&gt;
                  &lt;target&gt;1.8&lt;/target&gt;
              &lt;/configuration&gt;
          &lt;/plugin&gt;
      &lt;/plugins&gt;
  &lt;/build&gt;
  </code></pre>
  
  <p></p>
  
  <p>my java code : </p>
  
  <pre><code>import java.util.Properties;
  import org.apache.flink.api.common.functions.MapFunction;
  import org.apache.flink.streaming.api.datastream.DataStream;
  import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
  import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
  
  public class App 
  {
      public static void main(String[] args) throws Exception {
  
          StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  
          Properties properties = new Properties();
          properties.setProperty("bootstrap.servers", "localhost:9092");
          properties.setProperty("zookeeper.connect", "localhost:2181");
          properties.setProperty("group.id", "flink_consumer");
  
          DataStream&lt;String&gt; messageStream = env.addSource(new FlinkKafkaConsumer010&lt;&gt;
              ("ux_logs", new SimpleStringSchema(), properties));
  
          messageStream.rebalance().map(new MapFunction&lt;String, String&gt;() {
  
              private static final long serialVersionUID = -6867736771747690202L;
  
              public String map(String value) throws Exception {
                  return "Kafka and Flink says: " + value;
              }
          }).print();
  
          env.execute();
      }
  }
  </code></pre>
  
  <p>But I get this <strong>error:</strong></p>
  
  <pre><code>    java.lang.NoClassDefFoundError: org/apache/flink/streaming/api/checkpoint/CheckpointedFunction
      at java.lang.ClassLoader.defineClass1(Native Method)
      at java.lang.ClassLoader.defineClass(ClassLoader.java:763)
      at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
      at java.net.URLClassLoader.defineClass(URLClassLoader.java:467)
      at java.net.URLClassLoader.access$100(URLClassLoader.java:73)
      at java.net.URLClassLoader$1.run(URLClassLoader.java:368)
      at java.net.URLClassLoader$1.run(URLClassLoader.java:362)
      at java.security.AccessController.doPrivileged(Native Method)
      at java.net.URLClassLoader.findClass(URLClassLoader.java:361)
      at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
      at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
      at java.lang.ClassLoader.defineClass1(Native Method)
      at java.lang.ClassLoader.defineClass(ClassLoader.java:763)
      at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
      at java.net.URLClassLoader.defineClass(URLClassLoader.java:467)
      at java.net.URLClassLoader.access$100(URLClassLoader.java:73)
      at java.net.URLClassLoader$1.run(URLClassLoader.java:368)
      at java.net.URLClassLoader$1.run(URLClassLoader.java:362)
      at java.security.AccessController.doPrivileged(Native Method)
      at java.net.URLClassLoader.findClass(URLClassLoader.java:361)
      at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
      at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
      at java.lang.ClassLoader.defineClass1(Native Method)
      at java.lang.ClassLoader.defineClass(ClassLoader.java:763)
      at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
      at java.net.URLClassLoader.defineClass(URLClassLoader.java:467)
      at java.net.URLClassLoader.access$100(URLClassLoader.java:73)
      at java.net.URLClassLoader$1.run(URLClassLoader.java:368)
      at java.net.URLClassLoader$1.run(URLClassLoader.java:362)
      at java.security.AccessController.doPrivileged(Native Method)
      at java.net.URLClassLoader.findClass(URLClassLoader.java:361)
      at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
      at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
      at ux.App.main(App.java:27)
      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:509)
      at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:403)
      at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:320)
      at org.apache.flink.client.CliFrontend.executeProgram(CliFrontend.java:777)
      at org.apache.flink.client.CliFrontend.run(CliFrontend.java:253)
      at org.apache.flink.client.CliFrontend.parseParameters(CliFrontend.java:1005)
      at org.apache.flink.client.CliFrontend.main(CliFrontend.java:1048)
      Caused by: java.lang.ClassNotFoundException: org.apache.flink.streaming.api.checkpoint.CheckpointedFunction
      at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
      at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
      at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
  </code></pre>
  
  <p>Do i need to remove my kafka, and run an older version?
  Is my flink kafka connector wrong?
  i tried to use this plugin but it didn't work. (<a href="https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/linking.html" rel="nofollow noreferrer">https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/linking.html</a>)</p>
  
  <p>Thanks.</p>