GitHub user revprez opened a pull request:
https://github.com/apache/flink/pull/1511
Using AccumulatorSnapshot's class loader deserializing accumulators â¦
AccumulatorSnapshot is currently calling ClassLoader.getSystemClassLoader()
when deserializing accumulators. When running a SocketTextStreamWordCount
example instrumented by Spring Boot, the exception below is thrown.
This issue is apparently related to #1507, in which constructing certain
fat jars--specifically Spring Boot fat jars--somehow resulted in a loss of
context for the task runners. I don't know enough about Flink or class loading
to really know what's going on, and again I'm not sure if this is a general
enough fix for the issue or if the matter is best handled somehow by the user.
In any case, for your review. Exception print out is below and available
here as a [gist](https://gist.github.com/revprez/42b0058827728aa9c572)
```
java.lang.IllegalStateException: Failed to execute CommandLineRunner
at
org.springframework.boot.SpringApplication.callRunner(SpringApplication.java:807)
[spring-boot-1.3.1.RELEASE.jar:1.3.1.RELEASE]
at
org.springframework.boot.SpringApplication.callRunners(SpringApplication.java:788)
[spring-boot-1.3.1.RELEASE.jar:1.3.1.RELEASE]
at
org.springframework.boot.SpringApplication.afterRefresh(SpringApplication.java:775)
[spring-boot-1.3.1.RELEASE.jar:1.3.1.RELEASE]
at
org.springframework.boot.SpringApplication.doRun(SpringApplication.java:366)
[spring-boot-1.3.1.RELEASE.jar:1.3.1.RELEASE]
at
org.springframework.boot.SpringApplication.run(SpringApplication.java:305)
[spring-boot-1.3.1.RELEASE.jar:1.3.1.RELEASE]
at
org.springframework.boot.SpringApplication.run(SpringApplication.java:1124)
[spring-boot-1.3.1.RELEASE.jar:1.3.1.RELEASE]
at
org.springframework.boot.SpringApplication.run(SpringApplication.java:1113)
[spring-boot-1.3.1.RELEASE.jar:1.3.1.RELEASE]
at
org.opencorrelate.exercise.jvm.scala.SocketTextStreamWordCount$.main(SocketTextStreamWordCount.scala:88)
[classes/:na]
at
org.opencorrelate.exercise.jvm.scala.SocketTextStreamWordCount.main(SocketTextStreamWordCount.scala)
[classes/:na]
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
~[na:1.8.0_65]
at
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
~[na:1.8.0_65]
at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
~[na:1.8.0_65]
at java.lang.reflect.Method.invoke(Method.java:497) ~[na:1.8.0_65]
at
org.springframework.boot.maven.AbstractRunMojo$LaunchRunner.run(AbstractRunMojo.java:467)
[spring-boot-maven-plugin-1.3.1.RELEASE.jar:1.3.1.RELEASE]
at java.lang.Thread.run(Thread.java:745) [na:1.8.0_65]
Caused by: org.apache.flink.runtime.client.JobExecutionException: Job
execution failed.
at
org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$7.apply$mcV$sp(JobManager.scala:657)
~[flink-runtime_2.11-1.0-SNAPSHOT.jar:1.0-SNAPSHOT]
at
org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$7.apply(JobManager.scala:603)
~[flink-runtime_2.11-1.0-SNAPSHOT.jar:1.0-SNAPSHOT]
at
org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$7.apply(JobManager.scala:603)
~[flink-runtime_2.11-1.0-SNAPSHOT.jar:1.0-SNAPSHOT]
at
scala.concurrent.impl.Future$PromiseCompletingRunnable.liftedTree1$1(Future.scala:24)
~[scala-library-2.11.7.jar:na]
at
scala.concurrent.impl.Future$PromiseCompletingRunnable.run(Future.scala:24)
~[scala-library-2.11.7.jar:na]
at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:41)
~[akka-actor_2.11-2.3.7.jar:na]
at
akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:401)
~[akka-actor_2.11-2.3.7.jar:na]
at
scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
~[scala-library-2.11.7.jar:na]
at
scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.pollAndExecAll(ForkJoinPool.java:1253)
~[scala-library-2.11.7.jar:na]
at
scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1346)
~[scala-library-2.11.7.jar:na]
at
scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
~[scala-library-2.11.7.jar:na]
at
scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
~[scala-library-2.11.7.jar:na]
Caused by: java.lang.ClassNotFoundException:
org.apache.flink.runtime.accumulators.AccumulatorRegistry$Metric
at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
~[na:1.8.0_65]
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
~[na:1.8.0_65]
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)
~[na:1.8.0_65]
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
~[na:1.8.0_65]
at java.lang.Class.forName0(Native Method) ~[na:1.8.0_65]
at java.lang.Class.forName(Class.java:348) ~[na:1.8.0_65]
at
org.apache.flink.util.InstantiationUtil$ClassLoaderObjectInputStream.resolveClass(InstantiationUtil.java:61)
~[flink-core_2.11-1.0-SNAPSHOT.jar:1.0-SNAPSHOT]
at
java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1613)
~[na:1.8.0_65]
at
java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1518)
~[na:1.8.0_65]
at java.io.ObjectInputStream.readEnum(ObjectInputStream.java:1726)
~[na:1.8.0_65]
at
java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1348)
~[na:1.8.0_65]
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:371)
~[na:1.8.0_65]
at java.util.HashMap.readObject(HashMap.java:1394) ~[na:1.8.0_65]
at sun.reflect.GeneratedMethodAccessor34.invoke(Unknown Source)
~[na:na]
at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
~[na:1.8.0_65]
at java.lang.reflect.Method.invoke(Method.java:497) ~[na:1.8.0_65]
at
java.io.ObjectStreamClass.invokeReadObject(ObjectStreamClass.java:1058)
~[na:1.8.0_65]
at
java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1900)
~[na:1.8.0_65]
at
java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801)
~[na:1.8.0_65]
at
java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351)
~[na:1.8.0_65]
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:371)
~[na:1.8.0_65]
at
org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:287)
~[flink-core_2.11-1.0-SNAPSHOT.jar:1.0-SNAPSHOT]
at
org.apache.flink.util.SerializedValue.deserializeValue(SerializedValue.java:55)
~[flink-core_2.11-1.0-SNAPSHOT.jar:1.0-SNAPSHOT]
at
org.apache.flink.runtime.accumulators.AccumulatorSnapshot.deserializeFlinkAccumulators(AccumulatorSnapshot.java:74)
~[flink-runtime_2.11-1.0-SNAPSHOT.jar:1.0-SNAPSHOT]
at
org.apache.flink.runtime.executiongraph.ExecutionGraph.updateState(ExecutionGraph.java:1119)
~[flink-runtime_2.11-1.0-SNAPSHOT.jar:1.0-SNAPSHOT]
at
org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$4.apply$mcV$sp(JobManager.scala:458)
~[flink-runtime_2.11-1.0-SNAPSHOT.jar:1.0-SNAPSHOT]
at
org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$4.apply(JobManager.scala:457)
~[flink-runtime_2.11-1.0-SNAPSHOT.jar:1.0-SNAPSHOT]
at
org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$4.apply(JobManager.scala:457)
~[flink-runtime_2.11-1.0-SNAPSHOT.jar:1.0-SNAPSHOT]
... 9 common frames omitted
```
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/OCForks/flink spring-classloading-issues-0.10
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/1511.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #1511
----
commit 382345a624a5095a3fb6e75f6b479e6259ef4bcc
Author: Prez Cannady <[email protected]>
Date: 2016-01-14T23:58:41Z
Using AccumulatorSnapshot's class loader deserializing accumulators with
the system class loader results in a ClassNotFoundException for
AccumulatorRegistry and probably other stuff. Exposed by Spring.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
