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/>*