Hi Alex, How recent is your Flink 1.0-SNAPSHOT build? Maybe the code on the (local) cluster (in /git/flink/flink-dist/target/flink-1.0-SNAPSHOT-bin/flink-1.0-SNAPSHOT/) is not up to date?
I just tried it locally, and the job seems to execute: ./bin/flink run /home/robert/Downloads/flink-poc/target/flink-poc-1.0-SNAPSHOT.jar org.apache.flink.streaming.api.scala.DataStream@436bc3601/05/2016 21:44:09 Job execution switched to status RUNNING. 01/05/2016 21:44:09 Source: Custom Source -> Map(1/1) switched to SCHEDULED 01/05/2016 21:44:09 Source: Custom Source -> Map(1/1) switched to DEPLOYING 01/05/2016 21:44:09 Source: Custom Source -> Map(1/1) switched to RUNNING By the way, in order to print the stream, you have to call counts.print() instead of print(counts). On Tue, Jan 5, 2016 at 9:35 PM, Alex Rovner <alex.rov...@magnetic.com> wrote: > I believe I have set the version uniformly, unless I am overlooking > something in the pom. Attaching my project. > > I have tried building with both "mvn clean package" and "mvn clean package > -Pbuild-jar" > and I get the same exception. > > I am running my app with the following command: > > ~/git/flink/flink-dist/target/flink-1.0-SNAPSHOT-bin/flink-1.0-SNAPSHOT/bin/flink > run -c com.magnetic.KafkaWordCount > ~/git/flink-poc/target/flink-poc-1.0-SNAPSHOT.jar > > On Tue, Jan 5, 2016 at 12:47 PM Robert Metzger <rmetz...@apache.org> > wrote: > >> I think the problem is that you only set the version of the Kafka >> connector to 1.0-SNAPSHOT, not for the rest of the Flink dependencies. >> >> On Tue, Jan 5, 2016 at 6:18 PM, Alex Rovner <alex.rov...@magnetic.com> >> wrote: >> >>> Thanks Till for the info. I tried switching to 1.0-SNAPSHOT and now >>> facing another error: >>> >>> Caused by: java.lang.NoSuchMethodError: >>> org.apache.flink.streaming.api.operators.StreamingRuntimeContext.isCheckpointingEnabled()Z >>> at >>> org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer.run(FlinkKafkaConsumer.java:413) >>> at >>> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:58) >>> at >>> org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:55) >>> at >>> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:218) >>> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:584) >>> at java.lang.Thread.run(Thread.java:745) >>> >>> >>> On Tue, Jan 5, 2016 at 3:54 AM Till Rohrmann <trohrm...@apache.org> >>> wrote: >>> >>>> Hi Alex, >>>> >>>> this is a bug in the `0.10` release. Is it possible for you to switch >>>> to version `1.0-SNAPSHOT`. With this version, the error should no longer >>>> occur. >>>> >>>> Cheers, >>>> Till >>>> >>>> On Tue, Jan 5, 2016 at 1:31 AM, Alex Rovner <alex.rov...@magnetic.com> >>>> wrote: >>>> >>>>> Hello Flinkers! >>>>> >>>>> The below program produces the following error when running locally. I >>>>> am building the program using maven, using 0.10.0 and running in streaming >>>>> only local mode "start-local-streaming.sh". I have verified that kafka >>>>> and >>>>> the topic is working properly by using kafka-console-*.sh scripts. What am >>>>> I doing wrong? Any help would be appreciated it. >>>>> >>>>> Caused by: java.lang.NumberFormatException: For input string: "" >>>>> >>>>> at >>>>> java.lang.NumberFormatException.forInputString(NumberFormatException.java:65) >>>>> >>>>> at java.lang.Long.parseLong(Long.java:601) >>>>> >>>>> at java.lang.Long.valueOf(Long.java:803) >>>>> >>>>> at >>>>> org.apache.flink.streaming.connectors.kafka.internals.ZookeeperOffsetHandler.getOffsetFromZooKeeper(ZookeeperOffsetHandler.java:125) >>>>> >>>>> at >>>>> org.apache.flink.streaming.connectors.kafka.internals.ZookeeperOffsetHandler.seekFetcherToInitialOffsets(ZookeeperOffsetHandler.java:88) >>>>> >>>>> >>>>> def main(args: Array[String]) { >>>>> val env = StreamExecutionEnvironment.getExecutionEnvironment >>>>> >>>>> val properties = new Properties(); >>>>> properties.setProperty("bootstrap.servers", "localhost:9092"); >>>>> properties.setProperty("zookeeper.connect", "localhost:2181"); >>>>> properties.setProperty("group.id", "test"); >>>>> >>>>> val stream = env >>>>> .addSource(new FlinkKafkaConsumer082[String]("topic", new >>>>> SimpleStringSchema(), properties)) >>>>> >>>>> val counts = stream.map(f=>f.split(",")) >>>>> >>>>> print(counts) >>>>> >>>>> env.execute() >>>>> } >>>>> >>>>> -- >>>>> *Alex Rovner* >>>>> *Director, Data Engineering * >>>>> *o:* 646.759.0052 >>>>> >>>>> * <http://www.magnetic.com/>* >>>>> >>>>> >>>> -- >>> *Alex Rovner* >>> *Director, Data Engineering * >>> *o:* 646.759.0052 >>> >>> * <http://www.magnetic.com/>* >>> >>> >> -- > *Alex Rovner* > *Director, Data Engineering * > *o:* 646.759.0052 > > * <http://www.magnetic.com/>* > >