Submission Problem

[Matthias J. Sax-2]

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

[Till Rohrmann]

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
​

[Matthias J. Sax-2]

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
> ​
>

[Till Rohrmann]

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 <[hidden email]> 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
> > ​
> >
>

[Matthias J. Sax-2]

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 <[hidden email]> 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
>>> ​
>>>
>>
>

[Stephan Ewen]

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 <[hidden email]> 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 <[hidden email]>
> 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
> >>> ​
> >>>
> >>
> >
>
>

[Ufuk Celebi-2]

@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 <[hidden email]> 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 <[hidden email]> 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 <[hidden email]>
>> 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
>> >>>
>> >>>
>> >>
>> >
>>
>>

[Matthias J. Sax-2]

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 <[hidden email]> 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 <[hidden email]>
>> 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
>>>>> ​
>>>>>
>>>>
>>>
>>
>>
>

[Stephan Ewen]

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 <[hidden email]> 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 <[hidden email]>
> 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 <[hidden email]>
> >> 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
> >>>>> ​
> >>>>>
> >>>>
> >>>
> >>
> >>
> >
>
>

[till.rohrmann]

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" <[hidden email]> 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 <[hidden email]> 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 <[hidden email]>
> > 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 <[hidden email]>
> > >> 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
> > >>>>> ​
> > >>>>>
> > >>>>
> > >>>
> > >>
> > >>
> > >
> >
> >
>