You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@flink.apache.org by "Matthias J. Sax" <mj...@apache.org> on 2016/03/31 14:24:26 UTC
Submission Problem
Hi,
I just tried to submit Flink's Storm-Topology example via command line:
bin/flink run
~/workspace_flink/flink/flink-contrib/flink-storm-examples/target/WordCount-StormTopology.jar
However, I get a timeout and the program is not submitted. I tracked the
problem down to the following statement:
JobClient -> line 211:
> Future<Object> future = jobManagerGateway.ask(
> new JobManagerMessages.SubmitJob(
> jobGraph,
> ListeningBehaviour.DETACHED // only receive the Acknowledge for the job submission message
> ),
> timeout);
>
> result = Await.result(future, timeout);
The jobManagerGateway has the "value":
AkkaActorGateway(akka.tcp://flink@127.0.0.1:6123/user/jobmanager, null)
Not sure why Await.result does not return, as the value of
jobManagerGateway seems to be correct. Any idea?
-Matthias
Re: Submission Problem
Posted by Till Rohrmann <ti...@gmail.com>.
The PR Ufuk referenced should fix the problem. It is a known issue. It's
not merged yet, because I wanted to add an improved test case.
In the documentation for version 1.1-Snapshot, I've already addressed the
problem with the configuration parameter description.
Cheers,
Till
On Mar 31, 2016 6:09 PM, "Stephan Ewen" <se...@apache.org> wrote:
> I guess the fix is to put the entire ExecutionConfig into a
> "SerializedValue<ExecutionConfig>"?
>
> On Thu, Mar 31, 2016 at 5:34 PM, Matthias J. Sax <mj...@apache.org> wrote:
>
> > StormConfig is set a a global job parameter
> >
> > FlinkClient.java line 337ff
> >
> > > ExecutionConfig flinkConfig =
> > topology.getExecutionEnvironment().getConfig();
> > > flinkConfig.setGlobalJobParameters(new StormConfig(conf));
> >
> >
> >
> > On 03/31/2016 05:05 PM, Stephan Ewen wrote:
> > > Hmm, it is wrong that the JobManager tries to load that class directly
> > from
> > > the actor message.
> > > All user code should be deserialized lazily.
> > >
> > > How is that class passed? Implicitly through some config?
> > >
> > > On Thu, Mar 31, 2016 at 4:51 PM, Matthias J. Sax <mj...@apache.org>
> > wrote:
> > >
> > >> Here we go...
> > >>
> > >> StormConfig.class is contained in the user jar file. I guess I need to
> > >> "register" it somehow? Or is it a class loading issue?
> > >>
> > >>
> > >>> 2016-03-31 16:47:33,095 ERROR akka.remote.EndpointWriter
> > >> - AssociationError [akka.tcp://
> flink@127.0.0.1:6123
> > ]
> > >> <- [akka.tcp://flink@127.0.0.1:32775]: Error
> > >> [org.apache.flink.storm.util.StormConfig] [
> > >>> java.lang.ClassNotFoundException:
> > org.apache.flink.storm.util.StormConfig
> > >>> at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
> > >>> at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
> > >>> at java.security.AccessController.doPrivileged(Native Method)
> > >>> at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
> > >>> at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
> > >>> at
> sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308)
> > >>> at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
> > >>> at java.lang.Class.forName0(Native Method)
> > >>> at java.lang.Class.forName(Class.java:278)
> > >>> at
> > >> java.io.ObjectInputStream.resolveClass(ObjectInputStream.java:625)
> > >>> at
> > >>
> >
> akka.util.ClassLoaderObjectInputStream.resolveClass(ClassLoaderObjectInputStream.scala:19)
> > >>> at
> > >>
> java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1612)
> > >>> at
> > >> java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1517)
> > >>> at
> > >>
> > java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1771)
> > >>> at
> > >> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
> > >>> at
> > >>
> java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1997)
> > >>> at
> > >> java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1921)
> > >>> at
> > >>
> > java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
> > >>> at
> > >> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
> > >>> at
> > >>
> java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1997)
> > >>> at
> > >> java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1921)
> > >>> at
> > >>
> > java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
> > >>> at
> > >> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
> > >>> at
> > >>
> java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1997)
> > >>> at
> > >> java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1921)
> > >>> at
> > >>
> > java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
> > >>> at
> > >> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
> > >>> at
> > >>
> java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1997)
> > >>> at
> > >> java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1921)
> > >>> at
> > >>
> > java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
> > >>> at
> > >> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
> > >>> at
> > java.io.ObjectInputStream.readObject(ObjectInputStream.java:370)
> > >>> at
> > >>
> akka.serialization.JavaSerializer$$anonfun$1.apply(Serializer.scala:136)
> > >>> at
> scala.util.DynamicVariable.withValue(DynamicVariable.scala:57)
> > >>> at
> > >> akka.serialization.JavaSerializer.fromBinary(Serializer.scala:136)
> > >>> at
> > >>
> >
> akka.serialization.Serialization$$anonfun$deserialize$1.apply(Serialization.scala:104)
> > >>> at scala.util.Try$.apply(Try.scala:161)
> > >>> at
> > >> akka.serialization.Serialization.deserialize(Serialization.scala:98)
> > >>> at
> > >> akka.remote.MessageSerializer$.deserialize(MessageSerializer.scala:23)
> > >>> at
> > >>
> >
> akka.remote.DefaultMessageDispatcher.payload$lzycompute$1(Endpoint.scala:58)
> > >>> at
> > >> akka.remote.DefaultMessageDispatcher.payload$1(Endpoint.scala:58)
> > >>> at
> > akka.remote.DefaultMessageDispatcher.dispatch(Endpoint.scala:76)
> > >>> at
> > >>
> >
> akka.remote.EndpointReader$$anonfun$receive$2.applyOrElse(Endpoint.scala:937)
> > >>> at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
> > >>> at akka.remote.EndpointActor.aroundReceive(Endpoint.scala:415)
> > >>> 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)
> > >>> ]
> > >>
> > >>
> > >>
> > >> On 03/31/2016 04:31 PM, Till Rohrmann wrote:
> > >>> Could you please rerun the whole job with debug log level and logging
> > of
> > >>> Akka's lifecycle events turned on?
> > >>>
> > >>> Cheers,
> > >>> Till
> > >>>
> > >>> On Thu, Mar 31, 2016 at 4:21 PM, Matthias J. Sax <mj...@apache.org>
> > >> wrote:
> > >>>
> > >>>> enclosed the logs.. maybe you can make some sense out if them.
> > >>>>
> > >>>> On 03/31/2016 02:52 PM, Till Rohrmann wrote:
> > >>>>> I would assume that something went wrong on the JobManager side.
> > Could
> > >>>> you
> > >>>>> check the logs if they contain something suspicious? Additionally
> you
> > >>>> could
> > >>>>> turn on lifecycle event logging for Akka.
> > >>>>>
> > >>>>> Cheers,
> > >>>>> Till
> > >>>>>
> > >>>>>
> > >>>>
> > >>>
> > >>
> > >>
> > >
> >
> >
>
Re: Submission Problem
Posted by Stephan Ewen <se...@apache.org>.
I guess the fix is to put the entire ExecutionConfig into a
"SerializedValue<ExecutionConfig>"?
On Thu, Mar 31, 2016 at 5:34 PM, Matthias J. Sax <mj...@apache.org> wrote:
> StormConfig is set a a global job parameter
>
> FlinkClient.java line 337ff
>
> > ExecutionConfig flinkConfig =
> topology.getExecutionEnvironment().getConfig();
> > flinkConfig.setGlobalJobParameters(new StormConfig(conf));
>
>
>
> On 03/31/2016 05:05 PM, Stephan Ewen wrote:
> > Hmm, it is wrong that the JobManager tries to load that class directly
> from
> > the actor message.
> > All user code should be deserialized lazily.
> >
> > How is that class passed? Implicitly through some config?
> >
> > On Thu, Mar 31, 2016 at 4:51 PM, Matthias J. Sax <mj...@apache.org>
> wrote:
> >
> >> Here we go...
> >>
> >> StormConfig.class is contained in the user jar file. I guess I need to
> >> "register" it somehow? Or is it a class loading issue?
> >>
> >>
> >>> 2016-03-31 16:47:33,095 ERROR akka.remote.EndpointWriter
> >> - AssociationError [akka.tcp://flink@127.0.0.1:6123
> ]
> >> <- [akka.tcp://flink@127.0.0.1:32775]: Error
> >> [org.apache.flink.storm.util.StormConfig] [
> >>> java.lang.ClassNotFoundException:
> org.apache.flink.storm.util.StormConfig
> >>> at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
> >>> at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
> >>> at java.security.AccessController.doPrivileged(Native Method)
> >>> at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
> >>> at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
> >>> at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308)
> >>> at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
> >>> at java.lang.Class.forName0(Native Method)
> >>> at java.lang.Class.forName(Class.java:278)
> >>> at
> >> java.io.ObjectInputStream.resolveClass(ObjectInputStream.java:625)
> >>> at
> >>
> akka.util.ClassLoaderObjectInputStream.resolveClass(ClassLoaderObjectInputStream.scala:19)
> >>> at
> >> java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1612)
> >>> at
> >> java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1517)
> >>> at
> >>
> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1771)
> >>> at
> >> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
> >>> at
> >> java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1997)
> >>> at
> >> java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1921)
> >>> at
> >>
> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
> >>> at
> >> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
> >>> at
> >> java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1997)
> >>> at
> >> java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1921)
> >>> at
> >>
> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
> >>> at
> >> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
> >>> at
> >> java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1997)
> >>> at
> >> java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1921)
> >>> at
> >>
> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
> >>> at
> >> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
> >>> at
> >> java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1997)
> >>> at
> >> java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1921)
> >>> at
> >>
> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
> >>> at
> >> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
> >>> at
> java.io.ObjectInputStream.readObject(ObjectInputStream.java:370)
> >>> at
> >> akka.serialization.JavaSerializer$$anonfun$1.apply(Serializer.scala:136)
> >>> at scala.util.DynamicVariable.withValue(DynamicVariable.scala:57)
> >>> at
> >> akka.serialization.JavaSerializer.fromBinary(Serializer.scala:136)
> >>> at
> >>
> akka.serialization.Serialization$$anonfun$deserialize$1.apply(Serialization.scala:104)
> >>> at scala.util.Try$.apply(Try.scala:161)
> >>> at
> >> akka.serialization.Serialization.deserialize(Serialization.scala:98)
> >>> at
> >> akka.remote.MessageSerializer$.deserialize(MessageSerializer.scala:23)
> >>> at
> >>
> akka.remote.DefaultMessageDispatcher.payload$lzycompute$1(Endpoint.scala:58)
> >>> at
> >> akka.remote.DefaultMessageDispatcher.payload$1(Endpoint.scala:58)
> >>> at
> akka.remote.DefaultMessageDispatcher.dispatch(Endpoint.scala:76)
> >>> at
> >>
> akka.remote.EndpointReader$$anonfun$receive$2.applyOrElse(Endpoint.scala:937)
> >>> at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
> >>> at akka.remote.EndpointActor.aroundReceive(Endpoint.scala:415)
> >>> 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)
> >>> ]
> >>
> >>
> >>
> >> On 03/31/2016 04:31 PM, Till Rohrmann wrote:
> >>> Could you please rerun the whole job with debug log level and logging
> of
> >>> Akka's lifecycle events turned on?
> >>>
> >>> Cheers,
> >>> Till
> >>>
> >>> On Thu, Mar 31, 2016 at 4:21 PM, Matthias J. Sax <mj...@apache.org>
> >> wrote:
> >>>
> >>>> enclosed the logs.. maybe you can make some sense out if them.
> >>>>
> >>>> On 03/31/2016 02:52 PM, Till Rohrmann wrote:
> >>>>> I would assume that something went wrong on the JobManager side.
> Could
> >>>> you
> >>>>> check the logs if they contain something suspicious? Additionally you
> >>>> could
> >>>>> turn on lifecycle event logging for Akka.
> >>>>>
> >>>>> Cheers,
> >>>>> Till
> >>>>>
> >>>>>
> >>>>
> >>>
> >>
> >>
> >
>
>
Re: Submission Problem
Posted by "Matthias J. Sax" <mj...@apache.org>.
StormConfig is set a a global job parameter
FlinkClient.java line 337ff
> ExecutionConfig flinkConfig = topology.getExecutionEnvironment().getConfig();
> flinkConfig.setGlobalJobParameters(new StormConfig(conf));
On 03/31/2016 05:05 PM, Stephan Ewen wrote:
> Hmm, it is wrong that the JobManager tries to load that class directly from
> the actor message.
> All user code should be deserialized lazily.
>
> How is that class passed? Implicitly through some config?
>
> On Thu, Mar 31, 2016 at 4:51 PM, Matthias J. Sax <mj...@apache.org> wrote:
>
>> Here we go...
>>
>> StormConfig.class is contained in the user jar file. I guess I need to
>> "register" it somehow? Or is it a class loading issue?
>>
>>
>>> 2016-03-31 16:47:33,095 ERROR akka.remote.EndpointWriter
>> - AssociationError [akka.tcp://flink@127.0.0.1:6123]
>> <- [akka.tcp://flink@127.0.0.1:32775]: Error
>> [org.apache.flink.storm.util.StormConfig] [
>>> java.lang.ClassNotFoundException: org.apache.flink.storm.util.StormConfig
>>> at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
>>> at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
>>> at java.security.AccessController.doPrivileged(Native Method)
>>> at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
>>> at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
>>> at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308)
>>> at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
>>> at java.lang.Class.forName0(Native Method)
>>> at java.lang.Class.forName(Class.java:278)
>>> at
>> java.io.ObjectInputStream.resolveClass(ObjectInputStream.java:625)
>>> at
>> akka.util.ClassLoaderObjectInputStream.resolveClass(ClassLoaderObjectInputStream.scala:19)
>>> at
>> java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1612)
>>> at
>> java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1517)
>>> at
>> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1771)
>>> at
>> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
>>> at
>> java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1997)
>>> at
>> java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1921)
>>> at
>> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
>>> at
>> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
>>> at
>> java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1997)
>>> at
>> java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1921)
>>> at
>> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
>>> at
>> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
>>> at
>> java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1997)
>>> at
>> java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1921)
>>> at
>> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
>>> at
>> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
>>> at
>> java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1997)
>>> at
>> java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1921)
>>> at
>> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
>>> at
>> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
>>> at java.io.ObjectInputStream.readObject(ObjectInputStream.java:370)
>>> at
>> akka.serialization.JavaSerializer$$anonfun$1.apply(Serializer.scala:136)
>>> at scala.util.DynamicVariable.withValue(DynamicVariable.scala:57)
>>> at
>> akka.serialization.JavaSerializer.fromBinary(Serializer.scala:136)
>>> at
>> akka.serialization.Serialization$$anonfun$deserialize$1.apply(Serialization.scala:104)
>>> at scala.util.Try$.apply(Try.scala:161)
>>> at
>> akka.serialization.Serialization.deserialize(Serialization.scala:98)
>>> at
>> akka.remote.MessageSerializer$.deserialize(MessageSerializer.scala:23)
>>> at
>> akka.remote.DefaultMessageDispatcher.payload$lzycompute$1(Endpoint.scala:58)
>>> at
>> akka.remote.DefaultMessageDispatcher.payload$1(Endpoint.scala:58)
>>> at akka.remote.DefaultMessageDispatcher.dispatch(Endpoint.scala:76)
>>> at
>> akka.remote.EndpointReader$$anonfun$receive$2.applyOrElse(Endpoint.scala:937)
>>> at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
>>> at akka.remote.EndpointActor.aroundReceive(Endpoint.scala:415)
>>> 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)
>>> ]
>>
>>
>>
>> On 03/31/2016 04:31 PM, Till Rohrmann wrote:
>>> Could you please rerun the whole job with debug log level and logging of
>>> Akka's lifecycle events turned on?
>>>
>>> Cheers,
>>> Till
>>>
>>> On Thu, Mar 31, 2016 at 4:21 PM, Matthias J. Sax <mj...@apache.org>
>> wrote:
>>>
>>>> enclosed the logs.. maybe you can make some sense out if them.
>>>>
>>>> On 03/31/2016 02:52 PM, Till Rohrmann wrote:
>>>>> I would assume that something went wrong on the JobManager side. Could
>>>> you
>>>>> check the logs if they contain something suspicious? Additionally you
>>>> could
>>>>> turn on lifecycle event logging for Akka.
>>>>>
>>>>> Cheers,
>>>>> Till
>>>>>
>>>>>
>>>>
>>>
>>
>>
>
Re: Submission Problem
Posted by Ufuk Celebi <uc...@apache.org>.
@Till: Isn't this an instance of the bug fixed in
https://github.com/apache/flink/pull/1818?
On Thu, Mar 31, 2016 at 5:05 PM, Stephan Ewen <se...@apache.org> wrote:
> Hmm, it is wrong that the JobManager tries to load that class directly from
> the actor message.
> All user code should be deserialized lazily.
>
> How is that class passed? Implicitly through some config?
>
> On Thu, Mar 31, 2016 at 4:51 PM, Matthias J. Sax <mj...@apache.org> wrote:
>
>> Here we go...
>>
>> StormConfig.class is contained in the user jar file. I guess I need to
>> "register" it somehow? Or is it a class loading issue?
>>
>>
>> > 2016-03-31 16:47:33,095 ERROR akka.remote.EndpointWriter
>> - AssociationError [akka.tcp://flink@127.0.0.1:6123]
>> <- [akka.tcp://flink@127.0.0.1:32775]: Error
>> [org.apache.flink.storm.util.StormConfig] [
>> > java.lang.ClassNotFoundException: org.apache.flink.storm.util.StormConfig
>> > at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
>> > at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
>> > at java.security.AccessController.doPrivileged(Native Method)
>> > at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
>> > at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
>> > at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308)
>> > at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
>> > at java.lang.Class.forName0(Native Method)
>> > at java.lang.Class.forName(Class.java:278)
>> > at
>> java.io.ObjectInputStream.resolveClass(ObjectInputStream.java:625)
>> > at
>> akka.util.ClassLoaderObjectInputStream.resolveClass(ClassLoaderObjectInputStream.scala:19)
>> > at
>> java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1612)
>> > at
>> java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1517)
>> > at
>> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1771)
>> > at
>> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
>> > at
>> java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1997)
>> > at
>> java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1921)
>> > at
>> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
>> > at
>> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
>> > at
>> java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1997)
>> > at
>> java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1921)
>> > at
>> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
>> > at
>> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
>> > at
>> java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1997)
>> > at
>> java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1921)
>> > at
>> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
>> > at
>> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
>> > at
>> java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1997)
>> > at
>> java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1921)
>> > at
>> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
>> > at
>> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
>> > at java.io.ObjectInputStream.readObject(ObjectInputStream.java:370)
>> > at
>> akka.serialization.JavaSerializer$$anonfun$1.apply(Serializer.scala:136)
>> > at scala.util.DynamicVariable.withValue(DynamicVariable.scala:57)
>> > at
>> akka.serialization.JavaSerializer.fromBinary(Serializer.scala:136)
>> > at
>> akka.serialization.Serialization$$anonfun$deserialize$1.apply(Serialization.scala:104)
>> > at scala.util.Try$.apply(Try.scala:161)
>> > at
>> akka.serialization.Serialization.deserialize(Serialization.scala:98)
>> > at
>> akka.remote.MessageSerializer$.deserialize(MessageSerializer.scala:23)
>> > at
>> akka.remote.DefaultMessageDispatcher.payload$lzycompute$1(Endpoint.scala:58)
>> > at
>> akka.remote.DefaultMessageDispatcher.payload$1(Endpoint.scala:58)
>> > at akka.remote.DefaultMessageDispatcher.dispatch(Endpoint.scala:76)
>> > at
>> akka.remote.EndpointReader$$anonfun$receive$2.applyOrElse(Endpoint.scala:937)
>> > at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
>> > at akka.remote.EndpointActor.aroundReceive(Endpoint.scala:415)
>> > 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)
>> > ]
>>
>>
>>
>> On 03/31/2016 04:31 PM, Till Rohrmann wrote:
>> > Could you please rerun the whole job with debug log level and logging of
>> > Akka's lifecycle events turned on?
>> >
>> > Cheers,
>> > Till
>> >
>> > On Thu, Mar 31, 2016 at 4:21 PM, Matthias J. Sax <mj...@apache.org>
>> wrote:
>> >
>> >> enclosed the logs.. maybe you can make some sense out if them.
>> >>
>> >> On 03/31/2016 02:52 PM, Till Rohrmann wrote:
>> >>> I would assume that something went wrong on the JobManager side. Could
>> >> you
>> >>> check the logs if they contain something suspicious? Additionally you
>> >> could
>> >>> turn on lifecycle event logging for Akka.
>> >>>
>> >>> Cheers,
>> >>> Till
>> >>>
>> >>>
>> >>
>> >
>>
>>
Re: Submission Problem
Posted by Stephan Ewen <se...@apache.org>.
Hmm, it is wrong that the JobManager tries to load that class directly from
the actor message.
All user code should be deserialized lazily.
How is that class passed? Implicitly through some config?
On Thu, Mar 31, 2016 at 4:51 PM, Matthias J. Sax <mj...@apache.org> wrote:
> Here we go...
>
> StormConfig.class is contained in the user jar file. I guess I need to
> "register" it somehow? Or is it a class loading issue?
>
>
> > 2016-03-31 16:47:33,095 ERROR akka.remote.EndpointWriter
> - AssociationError [akka.tcp://flink@127.0.0.1:6123]
> <- [akka.tcp://flink@127.0.0.1:32775]: Error
> [org.apache.flink.storm.util.StormConfig] [
> > java.lang.ClassNotFoundException: org.apache.flink.storm.util.StormConfig
> > at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
> > at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
> > at java.security.AccessController.doPrivileged(Native Method)
> > at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
> > at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
> > at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308)
> > at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
> > at java.lang.Class.forName0(Native Method)
> > at java.lang.Class.forName(Class.java:278)
> > at
> java.io.ObjectInputStream.resolveClass(ObjectInputStream.java:625)
> > at
> akka.util.ClassLoaderObjectInputStream.resolveClass(ClassLoaderObjectInputStream.scala:19)
> > at
> java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1612)
> > at
> java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1517)
> > at
> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1771)
> > at
> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
> > at
> java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1997)
> > at
> java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1921)
> > at
> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
> > at
> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
> > at
> java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1997)
> > at
> java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1921)
> > at
> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
> > at
> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
> > at
> java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1997)
> > at
> java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1921)
> > at
> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
> > at
> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
> > at
> java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1997)
> > at
> java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1921)
> > at
> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
> > at
> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
> > at java.io.ObjectInputStream.readObject(ObjectInputStream.java:370)
> > at
> akka.serialization.JavaSerializer$$anonfun$1.apply(Serializer.scala:136)
> > at scala.util.DynamicVariable.withValue(DynamicVariable.scala:57)
> > at
> akka.serialization.JavaSerializer.fromBinary(Serializer.scala:136)
> > at
> akka.serialization.Serialization$$anonfun$deserialize$1.apply(Serialization.scala:104)
> > at scala.util.Try$.apply(Try.scala:161)
> > at
> akka.serialization.Serialization.deserialize(Serialization.scala:98)
> > at
> akka.remote.MessageSerializer$.deserialize(MessageSerializer.scala:23)
> > at
> akka.remote.DefaultMessageDispatcher.payload$lzycompute$1(Endpoint.scala:58)
> > at
> akka.remote.DefaultMessageDispatcher.payload$1(Endpoint.scala:58)
> > at akka.remote.DefaultMessageDispatcher.dispatch(Endpoint.scala:76)
> > at
> akka.remote.EndpointReader$$anonfun$receive$2.applyOrElse(Endpoint.scala:937)
> > at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
> > at akka.remote.EndpointActor.aroundReceive(Endpoint.scala:415)
> > 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)
> > ]
>
>
>
> On 03/31/2016 04:31 PM, Till Rohrmann wrote:
> > Could you please rerun the whole job with debug log level and logging of
> > Akka's lifecycle events turned on?
> >
> > Cheers,
> > Till
> >
> > On Thu, Mar 31, 2016 at 4:21 PM, Matthias J. Sax <mj...@apache.org>
> wrote:
> >
> >> enclosed the logs.. maybe you can make some sense out if them.
> >>
> >> On 03/31/2016 02:52 PM, Till Rohrmann wrote:
> >>> I would assume that something went wrong on the JobManager side. Could
> >> you
> >>> check the logs if they contain something suspicious? Additionally you
> >> could
> >>> turn on lifecycle event logging for Akka.
> >>>
> >>> Cheers,
> >>> Till
> >>>
> >>>
> >>
> >
>
>
Re: Submission Problem
Posted by "Matthias J. Sax" <mj...@apache.org>.
Here we go...
StormConfig.class is contained in the user jar file. I guess I need to
"register" it somehow? Or is it a class loading issue?
> 2016-03-31 16:47:33,095 ERROR akka.remote.EndpointWriter - AssociationError [akka.tcp://flink@127.0.0.1:6123] <- [akka.tcp://flink@127.0.0.1:32775]: Error [org.apache.flink.storm.util.StormConfig] [
> java.lang.ClassNotFoundException: org.apache.flink.storm.util.StormConfig
> at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
> at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
> at java.security.AccessController.doPrivileged(Native Method)
> at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
> at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
> at java.lang.Class.forName0(Native Method)
> at java.lang.Class.forName(Class.java:278)
> at java.io.ObjectInputStream.resolveClass(ObjectInputStream.java:625)
> at akka.util.ClassLoaderObjectInputStream.resolveClass(ClassLoaderObjectInputStream.scala:19)
> at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1612)
> at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1517)
> at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1771)
> at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
> at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1997)
> at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1921)
> at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
> at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
> at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1997)
> at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1921)
> at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
> at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
> at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1997)
> at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1921)
> at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
> at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
> at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1997)
> at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1921)
> at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1798)
> at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
> at java.io.ObjectInputStream.readObject(ObjectInputStream.java:370)
> at akka.serialization.JavaSerializer$$anonfun$1.apply(Serializer.scala:136)
> at scala.util.DynamicVariable.withValue(DynamicVariable.scala:57)
> at akka.serialization.JavaSerializer.fromBinary(Serializer.scala:136)
> at akka.serialization.Serialization$$anonfun$deserialize$1.apply(Serialization.scala:104)
> at scala.util.Try$.apply(Try.scala:161)
> at akka.serialization.Serialization.deserialize(Serialization.scala:98)
> at akka.remote.MessageSerializer$.deserialize(MessageSerializer.scala:23)
> at akka.remote.DefaultMessageDispatcher.payload$lzycompute$1(Endpoint.scala:58)
> at akka.remote.DefaultMessageDispatcher.payload$1(Endpoint.scala:58)
> at akka.remote.DefaultMessageDispatcher.dispatch(Endpoint.scala:76)
> at akka.remote.EndpointReader$$anonfun$receive$2.applyOrElse(Endpoint.scala:937)
> at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
> at akka.remote.EndpointActor.aroundReceive(Endpoint.scala:415)
> 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)
> ]
On 03/31/2016 04:31 PM, Till Rohrmann wrote:
> Could you please rerun the whole job with debug log level and logging of
> Akka's lifecycle events turned on?
>
> Cheers,
> Till
>
> On Thu, Mar 31, 2016 at 4:21 PM, Matthias J. Sax <mj...@apache.org> wrote:
>
>> enclosed the logs.. maybe you can make some sense out if them.
>>
>> On 03/31/2016 02:52 PM, Till Rohrmann wrote:
>>> I would assume that something went wrong on the JobManager side. Could
>> you
>>> check the logs if they contain something suspicious? Additionally you
>> could
>>> turn on lifecycle event logging for Akka.
>>>
>>> Cheers,
>>> Till
>>>
>>>
>>
>
Re: Submission Problem
Posted by Till Rohrmann <tr...@apache.org>.
Could you please rerun the whole job with debug log level and logging of
Akka's lifecycle events turned on?
Cheers,
Till
On Thu, Mar 31, 2016 at 4:21 PM, Matthias J. Sax <mj...@apache.org> wrote:
> enclosed the logs.. maybe you can make some sense out if them.
>
> On 03/31/2016 02:52 PM, Till Rohrmann wrote:
> > I would assume that something went wrong on the JobManager side. Could
> you
> > check the logs if they contain something suspicious? Additionally you
> could
> > turn on lifecycle event logging for Akka.
> >
> > Cheers,
> > Till
> >
> >
>
Re: Submission Problem
Posted by "Matthias J. Sax" <mj...@apache.org>.
enclosed the logs.. maybe you can make some sense out if them.
On 03/31/2016 02:52 PM, Till Rohrmann wrote:
> I would assume that something went wrong on the JobManager side. Could you
> check the logs if they contain something suspicious? Additionally you could
> turn on lifecycle event logging for Akka.
>
> Cheers,
> Till
>
>
Re: Submission Problem
Posted by Till Rohrmann <tr...@apache.org>.
I would assume that something went wrong on the JobManager side. Could you
check the logs if they contain something suspicious? Additionally you could
turn on lifecycle event logging for Akka.
Cheers,
Till