Hi Raghu,I switched to FlinkPipelineOptions instead of Options interface. The
FlinkPipelineRunner invokes a method in
org/apache/beam/runners/dataflow/DataflowPipelineRunner.
Which Beam jar file contains this DataflowPipelineRunner class?
aused by: java.lang.NoClassDefFoundError:
org/apache/beam/runners/dataflow/DataflowPipelineRunner at
org.apache.beam.runners.flink.FlinkPipelineRunner.fromOptions(FlinkPipelineRunner.java:8
Thanks so much for your time.AB From: amir bahmanyari <[email protected]>
To: "[email protected]" <[email protected]>; Raghu
Angadi <[email protected]>
Sent: Sunday, May 8, 2016 11:37 PM
Subject: Re: How to set groupId in KafkaIO.Read()
Hi Raghu,I passed this exception and many more behind by placing required jars
in Flink Lib folder.I MUST have the folliwng jars in Flink Lib folder to avoid
such runtime exceptions:
I now get a different issue. I use interface Options in my code following a
Beam kafka example.Exact same as how it is in the published example :
public static interface Options extends PipelineOptions, FlinkPipelineOptions {
@Description("Path of the file to write to") @Validation.Required
String getOutput(); void setOutput(String value); }
When deploying my Beam app fat jar to Flink cluster, it complains that Options
is not visible to the class loader.Is there a property I need to set to make
Options available to the class loader?I tried both public & public static,
different name etc....same issue.Following is the stacktrace.I appreciate your
help.Cheers
The program finished with the following
exception:org.apache.flink.client.program.ProgramInvocationException: The main
method caused an error. at
org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:520)
at
org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:403)
at org.apache.flink.client.program.Client.runBlocking(Client.java:248)
at
org.apache.flink.client.CliFrontend.executeProgramBlocking(CliFrontend.java:860)
at org.apache.flink.client.CliFrontend.run(CliFrontend.java:327)
at org.apache.flink.client.CliFrontend.parseParameters(CliFrontend.java:1184)
at org.apache.flink.client.CliFrontend.main(CliFrontend.java:1234)Caused
by: java.lang.IllegalArgumentException: interface
com.myco.tech.arc.ReadFromKafka2$Options is not visible from class loader
at java.lang.reflect.Proxy$ProxyClassFactory.apply(Proxy.java:581) at
java.lang.reflect.Proxy$ProxyClassFactory.apply(Proxy.java:557) at
java.lang.reflect.WeakCache$Factory.get(WeakCache.java:230) at
java.lang.reflect.WeakCache.get(WeakCache.java:127) at
java.lang.reflect.Proxy.getProxyClass0(Proxy.java:419) at
java.lang.reflect.Proxy.getProxyClass(Proxy.java:371) at
org.apache.beam.sdk.options.PipelineOptionsFactory.validateWellFormed(PipelineOptionsFactory.java:620)
at
org.apache.beam.sdk.options.ProxyInvocationHandler.as(ProxyInvocationHandler.java:208)
at
org.apache.beam.sdk.options.PipelineOptionsFactory$Builder.as(PipelineOptionsFactory.java:297)
at
org.apache.beam.sdk.options.PipelineOptionsFactory.as(PipelineOptionsFactory.java:130)
at com.walmart.tech.arc.ReadFromKafka2.main(ReadFromKafka2.java:125)
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:505)
... 6 more
From: amir bahmanyari <[email protected]>
To: Raghu Angadi <[email protected]>
Cc: "[email protected]" <[email protected]>
Sent: Sunday, May 8, 2016 7:48 PM
Subject: Re: How to set groupId in KafkaIO.Read()
Yes , I am wondering too. It happens when executing p.run().What checkpoints
can i do?Following is the complete stacktrace, FYI.Thanks very much for your
attention Raghu.
java.lang.RuntimeException: Pipeline execution failed at
org.apache.beam.runners.flink.FlinkPipelineRunner.run(FlinkPipelineRunner.java:119)
at
org.apache.beam.runners.flink.FlinkPipelineRunner.run(FlinkPipelineRunner.java:51)
at org.apache.beam.sdk.Pipeline.run(Pipeline.java:182) at
com.walmart.tech.arc.ReadFromKafka2.main(ReadFromKafka2.java:321) 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:505)
at
org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:403)
at org.apache.flink.client.program.Client.runBlocking(Client.java:248)
at
org.apache.flink.client.CliFrontend.executeProgramBlocking(CliFrontend.java:860)
at org.apache.flink.client.CliFrontend.run(CliFrontend.java:327)
at org.apache.flink.client.CliFrontend.parseParameters(CliFrontend.java:1184)
at org.apache.flink.client.CliFrontend.main(CliFrontend.java:1234)Caused
by: org.apache.flink.client.program.ProgramInvocationException: The program
execution failed: Failed to submit job 2acc4c7e104ea9695a09eda8bc7a6d36
(readfromkafka2-abahman-0509024419) at
org.apache.flink.client.program.Client.runBlocking(Client.java:381) at
org.apache.flink.client.program.Client.runBlocking(Client.java:355) at
org.apache.flink.streaming.api.environment.StreamContextEnvironment.execute(StreamContextEnvironment.java:65)
at
org.apache.beam.runners.flink.FlinkPipelineExecutionEnvironment.executePipeline(FlinkPipelineExecutionEnvironment.java:146)
at
org.apache.beam.runners.flink.FlinkPipelineRunner.run(FlinkPipelineRunner.java:116)
... 14 moreCaused by:
org.apache.flink.runtime.client.JobExecutionException: Failed to submit job
2acc4c7e104ea9695a09eda8bc7a6d36 (readfromkafka2-abahman-0509024419) at
org.apache.flink.runtime.jobmanager.JobManager.org$apache$flink$runtime$jobmanager$JobManager$$submitJob(JobManager.scala:1223)
at
org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1.applyOrElse(JobManager.scala:469)
at
scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
at
scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
at
scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
at
org.apache.flink.runtime.LeaderSessionMessageFilter$$anonfun$receive$1.applyOrElse(LeaderSessionMessageFilter.scala:36)
at
scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
at
scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
at
scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
at
org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:33)
at org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:28)
at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:118)
at
org.apache.flink.runtime.LogMessages$$anon$1.applyOrElse(LogMessages.scala:28)
at akka.actor.Actor$class.aroundReceive(Actor.scala:465) at
org.apache.flink.runtime.jobmanager.JobManager.aroundReceive(JobManager.scala:113)
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516) at
akka.actor.ActorCell.invoke(ActorCell.scala:487) at
akka.dispatch.Mailbox.processMailbox(Mailbox.scala:254) at
akka.dispatch.Mailbox.run(Mailbox.scala:221) at
akka.dispatch.Mailbox.exec(Mailbox.scala:231) at
scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) at
scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
at
scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
at
scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)Caused
by: org.apache.flink.runtime.JobException: Creating the input splits caused an
error: Could not create input splits from Source. at
org.apache.flink.runtime.executiongraph.ExecutionJobVertex.<init>(ExecutionJobVertex.java:172)
at
org.apache.flink.runtime.executiongraph.ExecutionGraph.attachJobGraph(ExecutionGraph.java:701)
at
org.apache.flink.runtime.jobmanager.JobManager.org$apache$flink$runtime$jobmanager$JobManager$$submitJob(JobManager.scala:1147)
... 23 moreCaused by: java.io.IOException: Could not create input
splits from Source. at
org.apache.beam.runners.flink.translation.wrappers.SourceInputFormat.createInputSplits(SourceInputFormat.java:109)
at
org.apache.beam.runners.flink.translation.wrappers.SourceInputFormat.createInputSplits(SourceInputFormat.java:41)
at
org.apache.flink.runtime.executiongraph.ExecutionJobVertex.<init>(ExecutionJobVertex.java:156)
... 25 moreCaused by: org.apache.kafka.common.config.ConfigException:
Missing required configuration "group.id" which has no default value. at
org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:124) at
org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:48)
at
org.apache.kafka.clients.consumer.ConsumerConfig.<init>(ConsumerConfig.java:194)
at
org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:380)
at
org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:363)
at
org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:350)
at org.apache.beam.sdk.io.kafka.KafkaIO$Read$1.apply(KafkaIO.java:339)
at org.apache.beam.sdk.io.kafka.KafkaIO$Read$1.apply(KafkaIO.java:337)
at
org.apache.beam.sdk.io.kafka.KafkaIO$UnboundedKafkaSource.generateInitialSplits(KafkaIO.java:572)
at
org.apache.beam.sdk.io.BoundedReadFromUnboundedSource$UnboundedToBoundedSourceAdapter.splitIntoBundles(BoundedReadFromUnboundedSource.java:165)
at
org.apache.beam.runners.flink.translation.wrappers.SourceInputFormat.createInputSplits(SourceInputFormat.java:101)
... 27 more
From: Raghu Angadi <[email protected]>
To: amir bahmanyari <[email protected]>
Cc: "[email protected]" <[email protected]>
Sent: Sunday, May 8, 2016 6:35 PM
Subject: Re: How to set groupId in KafkaIO.Read()
The main issue is why are seeing this exception in the first place. You don't
have to provide a group id.
On Sunday, May 8, 2016, Raghu Angadi <[email protected]> wrote:
Please see javadic for the method. You can create a map anyway you like. Do you
need help creating a map in Java?
On Sunday, May 8, 2016, amir bahmanyari <[email protected]> wrote:
Hi Raghu,Am using Kafka 0.9.0. Added
updateConsumerProperites(ImmutableMap.of("group.id", "myGroup").I get
ImmutableMap not resolved
tried many ways to put the right jar to resolve it. Didnt cut it.
Thanks.
What version of Kafka are you using? KafkaIO does not need you to set a
group.id. Not sure why you are seeing this exception.
Though KafkaIO does not need it, you can set it:
KafkaIO.read().updateConsumerProperites(ImmutableMap.of("group.id", "temp"));
On Sat, May 7, 2016 at 9:34 PM, amir bahmanyari <[email protected]> wrote:
Sorry missed the subject! :)
From: amir bahmanyari <[email protected]>
To: "[email protected]" <[email protected]>
Sent: Saturday, May 7, 2016 9:31 PM
Subject:
Hi colleagues,Hope you are having a great weekend.I get a Kafka configuration
exception, GroupId not being set, when trying to run my Beam app (Flink
runner) in a Flink cluster.
I couldn't find a reference to a method that seta a GroupId similar to what we
do for withBootstrapServers("host:9092").withTopics(topics).withMaxNumRecords
etc.The bottom of the stack trace is provide below.How can I set a GroupId
property for KafkaIO.read()?Thanks for your help.
... 25 moreCaused by: org.apache.kafka.common.config.ConfigException:
Missing required configuration "group.id" which has no default value. at
org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:124) at
org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:48)
at
org.apache.kafka.clients.consumer.ConsumerConfig.<init>(ConsumerConfig.java:194)
at
org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:380)
at
org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:363)
at
org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:350)
at org.apache.beam.sdk.io.kafka.KafkaIO$Read$1.apply(KafkaIO.java:339)
at org.apache.beam.sdk.io.kafka.KafkaIO$Read$1.apply(KafkaIO.java:337)
at
org.apache.beam.sdk.io.kafka.KafkaIO$UnboundedKafkaSource.generateInitialSplits(KafkaIO.java:572)
at
org.apache.beam.sdk.io.BoundedReadFromUnboundedSource$UnboundedToBoundedSourceAdapter.splitIntoBundles(BoundedReadFromUnboundedSource.java:165)
at
org.apache.beam.runners.flink.translation.wrappers.SourceInputFormat.createInputSplits(SourceInputFormat.java:101)
... 27 more
