You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by shotte <ho...@gmail.com> on 2016/02/10 00:40:05 UTC
Simple Flink - Kafka Test
Hi,
I am new to Flink and Kafka
I am trying to read from Flink a Kafka topic and sent it back to another
Kafka topic
Here my setup:
Flink 0.10.1
Kafka 0.9
All that on a single node
I successfully wrote a Java program that send message to Kafka (topic =
demo), and I have a consumer (in a shell) that read it. so that working.
When I execute the flink program I got this error
See code and Error Below...it is something between step C & D
What I am doing wrong ?
Thanks
Sylvain
<Code start>
package com.sylvain;
import java.util.Properties;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import
org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer082;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
/**
* Skeleton for a Flink Job.
*
* For a full example of a Flink Job, see the WordCountJob.java file in the
* same package/directory or have a look at the website.
*
* You can also generate a .jar file that you can submit on your Flink
* cluster.
* Just type
* mvn clean package
* in the projects root directory.
* You will find the jar in
* target/flink-quickstart-0.1-SNAPSHOT-Sample.jar
*
*/
public class Job {
public static void main(String[] args) throws Exception {
// set up the execution environment
System.out.println("Step A");
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
System.out.println("Step B");
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "localhost:9092");
properties.setProperty("zookeeper.connect", "localhost:2181");
properties.setProperty("group.id", "test");
System.out.println("Step C");
DataStream<String> messageStream = env.addSource(new
FlinkKafkaConsumer082<>("demo", new SimpleStringSchema(), properties));
System.out.println("Step D");
messageStream.map(new MapFunction<String, String>(){
@Override
public String map(String value) throws Exception {
// TODO Auto-generated method stub
return "Blablabla " + value;
}
}).addSink(new FlinkKafkaProducer("localhost:9092", "demo2", new
SimpleStringSchema()));
System.out.println("Step E");
env.execute();
System.out.println("Step F");
}
}
<Code end>
<Error Start>
[shotte@localhost flink-kafka]$ flink run ./target/flink-kafka-0.1.jar
Step A
Step B
Step C
java.lang.NoClassDefFoundError: scala/collection/GenTraversableOnce$class
at kafka.utils.Pool.<init>(Pool.scala:28)
at
kafka.consumer.FetchRequestAndResponseStatsRegistry$.<init>(FetchRequestAndResponseStats.scala:60)
at
kafka.consumer.FetchRequestAndResponseStatsRegistry$.<clinit>(FetchRequestAndResponseStats.scala)
at kafka.consumer.SimpleConsumer.<init>(SimpleConsumer.scala:39)
at
kafka.javaapi.consumer.SimpleConsumer.<init>(SimpleConsumer.scala:34)
at
org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer.getPartitionsForTopic(FlinkKafkaConsumer.java:691)
at
org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer.<init>(FlinkKafkaConsumer.java:281)
at
org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer082.<init>(FlinkKafkaConsumer082.java:49)
at com.sylvain.Job.main(Job.java:64)
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:483)
at
org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:497)
at
org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:395)
at
org.apache.flink.client.program.Client.runBlocking(Client.java:252)
at
org.apache.flink.client.CliFrontend.executeProgramBlocking(CliFrontend.java:676)
at org.apache.flink.client.CliFrontend.run(CliFrontend.java:326)
at
org.apache.flink.client.CliFrontend.parseParameters(CliFrontend.java:978)
at org.apache.flink.client.CliFrontend.main(CliFrontend.java:1028)
Caused by: java.lang.ClassNotFoundException:
scala.collection.GenTraversableOnce$class
at java.net.URLClassLoader$1.run(URLClassLoader.java:372)
at java.net.URLClassLoader$1.run(URLClassLoader.java:361)
at java.security.AccessController.doPrivileged(Native Method)
at java.net.URLClassLoader.findClass(URLClassLoader.java:360)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
... 20 more
<Error End>
--
View this message in context: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Simple-Flink-Kafka-Test-tp4828.html
Sent from the Apache Flink User Mailing List archive. mailing list archive at Nabble.com.
Re: Simple Flink - Kafka Test
Posted by Robert Metzger <rm...@apache.org>.
Quick clarification on Stephan's comment: In Flink 0.10, no suffix means
scala 2.10, for Scala 2.11 you have to add the _2.11 suffix to ALL
dependencies (including flink-java_2.11, flink-core_2.11 and so on).
In Flink 1.0, all artifacts depending on scala have a version suffix. For
example flink-clients_2.10 or flink-runtime_2.10 depend on scala and have
suffixes. However flink-java or flink-core do not depend on Scala at all,
hence you can not put a suffix.
I think both approaches are confusing and inconsistent in some sense.
On Wed, Feb 10, 2016 at 3:49 PM, Stephan Ewen <se...@apache.org> wrote:
> Yes, 0.10.x does not always have Scala version suffixes.
>
> 1.0 is doing this consistently, should cause less confusion...
>
> On Wed, Feb 10, 2016 at 2:38 PM, shotte <ho...@gmail.com> wrote:
>
>> Ok It is working now
>>
>> I had to change a few dependency with the _2.11 suffix
>>
>> Thanks
>>
>> Sylvain
>>
>>
>> <dependencies>
>> <dependency>
>> <groupId>org.apache.flink</groupId>
>> <artifactId>flink-java</artifactId>
>> <version>${flink.version}</version>
>> </dependency>
>> <dependency>
>> <groupId>org.apache.flink</groupId>
>> <artifactId>flink-clients_2.11</artifactId>
>> <version>${flink.version}</version>
>> </dependency>
>> <dependency>
>> <groupId>org.apache.flink</groupId>
>>
>> <artifactId>flink-connector-kafka_2.11</artifactId>
>> <version>0.10.1</version>
>> </dependency>
>> <dependency>
>> <groupId>org.apache.flink</groupId>
>> <artifactId>flink-streaming-java_2.11</artifactId>
>> <version>0.10.1</version>
>> </dependency>
>> </dependencies>
>>
>>
>>
>>
>> --
>> View this message in context:
>> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Simple-Flink-Kafka-Test-tp4828p4852.html
>> Sent from the Apache Flink User Mailing List archive. mailing list
>> archive at Nabble.com.
>>
>
>
Re: Simple Flink - Kafka Test
Posted by Stephan Ewen <se...@apache.org>.
Yes, 0.10.x does not always have Scala version suffixes.
1.0 is doing this consistently, should cause less confusion...
On Wed, Feb 10, 2016 at 2:38 PM, shotte <ho...@gmail.com> wrote:
> Ok It is working now
>
> I had to change a few dependency with the _2.11 suffix
>
> Thanks
>
> Sylvain
>
>
> <dependencies>
> <dependency>
> <groupId>org.apache.flink</groupId>
> <artifactId>flink-java</artifactId>
> <version>${flink.version}</version>
> </dependency>
> <dependency>
> <groupId>org.apache.flink</groupId>
> <artifactId>flink-clients_2.11</artifactId>
> <version>${flink.version}</version>
> </dependency>
> <dependency>
> <groupId>org.apache.flink</groupId>
> <artifactId>flink-connector-kafka_2.11</artifactId>
> <version>0.10.1</version>
> </dependency>
> <dependency>
> <groupId>org.apache.flink</groupId>
> <artifactId>flink-streaming-java_2.11</artifactId>
> <version>0.10.1</version>
> </dependency>
> </dependencies>
>
>
>
>
> --
> View this message in context:
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Simple-Flink-Kafka-Test-tp4828p4852.html
> Sent from the Apache Flink User Mailing List archive. mailing list archive
> at Nabble.com.
>
Re: Simple Flink - Kafka Test
Posted by shotte <ho...@gmail.com>.
Ok It is working now
I had to change a few dependency with the _2.11 suffix
Thanks
Sylvain
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_2.11</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka_2.11</artifactId>
<version>0.10.1</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_2.11</artifactId>
<version>0.10.1</version>
</dependency>
</dependencies>
--
View this message in context: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Simple-Flink-Kafka-Test-tp4828p4852.html
Sent from the Apache Flink User Mailing List archive. mailing list archive at Nabble.com.
Re: Simple Flink - Kafka Test
Posted by shotte <ho...@gmail.com>.
Hi,
Thanks for your reply, but I am still a bit confuse.
I have downloaded flink-0.10.1-bin-hadoop27-scala_2.11.tgz
and kafka_2.11-0.9.0.0.tgz
I did not install myself Scala
Now tell me if I understand correctly.
Depending on the version of Flink I have (in my case the scala 2.11) I must
specify a dependency in the pom file
But still have the same error
Here's an extract of my file
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka</artifactId>
<version>0.10.1</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_2.11</artifactId>
<version>0.10.1</version>
</dependency>
</dependencies>
--
View this message in context: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Simple-Flink-Kafka-Test-tp4828p4850.html
Sent from the Apache Flink User Mailing List archive. mailing list archive at Nabble.com.
Re: Simple Flink - Kafka Test
Posted by Chiwan Park <ch...@apache.org>.
The documentation I sent is for Flink 1.0.
In Flink 0.10.x, there is no suffix of dependencies for Scala 2.10 (e.g. flink-streaming-java). But there is a suffix of dependencies for Scala 2.11 (e.g. flink-streaming-java_2.11).
Regards,
Chiwan Park
> On Feb 10, 2016, at 1:46 PM, Chiwan Park <ch...@apache.org> wrote:
>
> Hi shotte,
>
> The exception is caused by Scala version mismatch. If you want to use Scala 2.11, you have to set Flink dependencies compiled for Scala 2.11. We have a documentation about this in wiki [1].
>
> I hope this helps.
>
> [1]: https://cwiki.apache.org/confluence/display/FLINK/Maven+artifact+names+suffixed+with+Scala+version
>
> Regards,
> Chiwan Park
>
>> On Feb 10, 2016, at 9:39 AM, shotte <ho...@gmail.com> wrote:
>>
>> Do I need to go to Flink 1.0 or the downgrade to Kafka 0.8 ?
>>
>>
>>
>> --
>> View this message in context: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Simple-Flink-Kafka-Test-tp4828p4829.html
>> Sent from the Apache Flink User Mailing List archive. mailing list archive at Nabble.com.
>
Re: Simple Flink - Kafka Test
Posted by Chiwan Park <ch...@apache.org>.
Hi shotte,
The exception is caused by Scala version mismatch. If you want to use Scala 2.11, you have to set Flink dependencies compiled for Scala 2.11. We have a documentation about this in wiki [1].
I hope this helps.
[1]: https://cwiki.apache.org/confluence/display/FLINK/Maven+artifact+names+suffixed+with+Scala+version
Regards,
Chiwan Park
> On Feb 10, 2016, at 9:39 AM, shotte <ho...@gmail.com> wrote:
>
> Do I need to go to Flink 1.0 or the downgrade to Kafka 0.8 ?
>
>
>
> --
> View this message in context: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Simple-Flink-Kafka-Test-tp4828p4829.html
> Sent from the Apache Flink User Mailing List archive. mailing list archive at Nabble.com.
Re: Simple Flink - Kafka Test
Posted by shotte <ho...@gmail.com>.
Do I need to go to Flink 1.0 or the downgrade to Kafka 0.8 ?
--
View this message in context: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Simple-Flink-Kafka-Test-tp4828p4829.html
Sent from the Apache Flink User Mailing List archive. mailing list archive at Nabble.com.