Thanks again for providing the project to reproduce the error. Unfortunately, I
could not reproduce the error with your instructions. I've tried killing
TaskManagers with the restricted MetaSpace memory but I did not run out of
MetaSpace memory in the remaining TaskManager. I did not use YARN.
I saw your bug report in FLINK [1]. Some comments on that:
First of all your initial issue is unrelated to Beam but caused by the Metaspace
growing and Yarn killing your application [2]. The Metaspace grows because Flink
does not limit it by default and when a job is restarted it needs to temporarily
increase the MetaSpace size. Limiting it should force a GC instead. Read on to
find out why GC might not work.
You mentioned in the first comment of [1] that you copied the job jar into the
lib folder. This is problematic with "parent-first" class loading because now
libraries like Jackson in your job will be loaded from the parent classloader.
This can then install the Flink Job classloader into the Jackson cache. The
cache and Jackson will never be evicted as they are part of the parent
classloader. Note, that this would not be a problem if class loading was
"child-first" and Jackson is part of the user jar.
However,
1) The YARN "cluster" mode does seem to have a problem because it includes the
jars into the Flink class loader. So really "child-first" vs "parent-first" does
not matter because it will always fall back to the parent/Flink class loader.
2) You might want to restart the entire YARN cluster on failures in the YARN
"cluster" mode (see yarn.maximum-failed-containers), or use the "session" mode
with "child-first".
3) If the above is not feasible, you have to take care of cleaning the Jackson
cache or any other static cache, like here: https://github.com/apache/beam/pull/7552
I think the subtlety of this behavior warrants that we also merge the fix in
Beam, as we have seen that clearing the cache gets rid of the problem.
Cheers,
Max
[1] https://issues.apache.org/jira/browse/FLINK-11205
[2] https://issues.apache.org/jira/browse/FLINK-10317
On 21.01.19 03:46, Daniel Harper wrote:
"classloader.resolve-order” is set to ‘parent-first’
In the simple job that I wrote, the mvn dependency:tree yields the
following. It looks like beam-sdks-java-core includes jackson-databind,
and the flink runtime lib shades it
[INFO] Scanning for projects...
[WARNING]
[WARNING] Some problems were encountered while building the effective
model for example:streaming-job:jar:1.0-SNAPSHOT
[WARNING] 'build.plugins.plugin.version' for
org.apache.maven.plugins:maven-jar-plugin is missing. @ line 60, column 21
[WARNING]
[WARNING] It is highly recommended to fix these problems because they
threaten the stability of your build.
[WARNING]
[WARNING] For this reason, future Maven versions might no longer support
building such malformed projects.
[WARNING]
[INFO]
[INFO] -----------------------< example:streaming-job
------------------------
[INFO] Building streaming-job 1.0-SNAPSHOT
[INFO] --------------------------------[ jar
]---------------------------------
[INFO]
[INFO] --- maven-dependency-plugin:2.8:tree (default-cli) @ streaming-job
---
[INFO] example:streaming-job:jar:1.0-SNAPSHOT
[INFO] +- org.apache.beam:beam-sdks-java-core:jar:2.7.0:compile
[INFO] | +- com.fasterxml.jackson.core:jackson-core:jar:2.9.5:compile
[INFO] | +-
com.fasterxml.jackson.core:jackson-annotations:jar:2.9.5:compile
[INFO] | +- com.fasterxml.jackson.core:jackson-databind:jar:2.9.5:compile
[INFO] | +- org.slf4j:slf4j-api:jar:1.7.25:compile
[INFO] | +- org.apache.avro:avro:jar:1.8.2:compile
[INFO] | | +- org.codehaus.jackson:jackson-core-asl:jar:1.9.13:compile
[INFO] | | +- org.codehaus.jackson:jackson-mapper-asl:jar:1.9.13:compile
[INFO] | | \- com.thoughtworks.paranamer:paranamer:jar:2.7:compile
[INFO] | +- org.xerial.snappy:snappy-java:jar:1.1.4:compile
[INFO] | +- joda-time:joda-time:jar:2.4:compile
[INFO] | \- org.tukaani:xz:jar:1.8:compile
[INFO] \- org.apache.beam:beam-runners-flink_2.11:jar:2.7.0:compile
[INFO] +- org.apache.beam:beam-runners-core-java:jar:2.7.0:compile
[INFO] | +- org.apache.beam:beam-model-pipeline:jar:2.7.0:compile
[INFO] | | \-
com.google.errorprone:error_prone_annotations:jar:2.1.2:compile
[INFO] | \- org.apache.beam:beam-model-fn-execution:jar:2.7.0:compile
[INFO] +-
org.apache.beam:beam-runners-core-construction-java:jar:2.7.0:compile
[INFO] | \- org.apache.beam:beam-model-job-management:jar:2.7.0:compile
[INFO] +-
org.apache.beam:beam-runners-java-fn-execution:jar:2.7.0:compile
[INFO] | +-
org.apache.beam:beam-sdks-java-fn-execution:jar:2.7.0:compile
[INFO] | \-
org.apache.beam:beam-vendor-sdks-java-extensions-protobuf:jar:2.7.0:compile
[INFO] +- org.slf4j:slf4j-simple:jar:1.7.25:compile
[INFO] +- org.apache.commons:commons-compress:jar:1.16.1:compile
[INFO] | \- org.objenesis:objenesis:jar:2.6:compile
[INFO] +- args4j:args4j:jar:2.33:compile
[INFO] +- org.apache.flink:flink-clients_2.11:jar:1.5.2:compile
[INFO] | +- org.apache.flink:flink-optimizer_2.11:jar:1.5.2:compile
[INFO] | +- commons-cli:commons-cli:jar:1.3.1:compile
[INFO] | +- com.google.code.findbugs:jsr305:jar:1.3.9:compile
[INFO] | \- org.apache.flink:force-shading:jar:1.5.2:compile
[INFO] +- org.apache.flink:flink-core:jar:1.5.2:compile
[INFO] | +- org.apache.flink:flink-annotations:jar:1.5.2:compile
[INFO] | +- org.apache.flink:flink-shaded-asm:jar:5.0.4-2.0:compile
[INFO] | +- org.apache.commons:commons-lang3:jar:3.3.2:compile
[INFO] | +- com.esotericsoftware.kryo:kryo:jar:2.24.0:compile
[INFO] | | \- com.esotericsoftware.minlog:minlog:jar:1.2:compile
[INFO] | \- commons-collections:commons-collections:jar:3.2.2:compile
[INFO] +- org.apache.flink:flink-metrics-core:jar:1.5.2:compile
[INFO] +- org.apache.flink:flink-java:jar:1.5.2:compile
[INFO] | \- org.apache.commons:commons-math3:jar:3.5:compile
[INFO] +- org.apache.flink:flink-runtime_2.11:jar:1.5.2:compile
[INFO] | +-
org.apache.flink:flink-queryable-state-client-java_2.11:jar:1.5.2:compile
[INFO] | +- org.apache.flink:flink-hadoop-fs:jar:1.5.2:compile
[INFO] | +- commons-io:commons-io:jar:2.4:compile
[INFO] | +-
org.apache.flink:flink-shaded-netty:jar:4.0.27.Final-2.0:compile
[INFO] | +- org.apache.flink:flink-shaded-guava:jar:18.0-2.0:compile
[INFO] | +- org.apache.flink:flink-shaded-jackson:jar:2.7.9-3.0:compile
[INFO] | +- org.javassist:javassist:jar:3.18.2-GA:compile
[INFO] | +- org.scala-lang:scala-library:jar:2.11.12:compile
[INFO] | +- com.typesafe.akka:akka-actor_2.11:jar:2.4.20:compile
[INFO] | | +- com.typesafe:config:jar:1.3.0:compile
[INFO] | | \-
org.scala-lang.modules:scala-java8-compat_2.11:jar:0.7.0:compile
[INFO] | +- com.typesafe.akka:akka-stream_2.11:jar:2.4.20:compile
[INFO] | | +- org.reactivestreams:reactive-streams:jar:1.0.0:compile
[INFO] | | \- com.typesafe:ssl-config-core_2.11:jar:0.2.1:compile
[INFO] | | \-
org.scala-lang.modules:scala-parser-combinators_2.11:jar:1.0.4:compile
[INFO] | +- com.typesafe.akka:akka-protobuf_2.11:jar:2.4.20:compile
[INFO] | +- com.typesafe.akka:akka-slf4j_2.11:jar:2.4.20:compile
[INFO] | +- org.clapper:grizzled-slf4j_2.11:jar:1.0.2:compile
[INFO] | +- com.github.scopt:scopt_2.11:jar:3.5.0:compile
[INFO] | \- com.twitter:chill_2.11:jar:0.7.4:compile
[INFO] | \- com.twitter:chill-java:jar:0.7.4:compile
[INFO] \- org.apache.flink:flink-streaming-java_2.11:jar:1.5.2:compile
[INFO]
------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO]
------------------------------------------------------------------------
[INFO] Total time: 0.928 s
[INFO] Finished at: 2019-01-21T08:45:08Z
[INFO]
------------------------------------------------------------------------
On 18/01/2019, 17:01, "Maximilian Michels" <[email protected]> wrote:
Hi Daniel,
I did some more debugging. I think the fix we proposed only cures the
symptoms.
The cause is that your job uses Jackson which is also a dependency of
Flink.
So your job ends up using Flink's version of Jackson which then installs
classes
from your job in the Jackson cache. Now, this wouldn't be a problem, if
you
shaded your version of Jackson, i.e. renamed the Jackson package.
But even without shading, the default behavior of Flink is to load user
classes
first. Could you please check:
1) Is "classloader.resolve-order" set to "child-first"?
2) Do you include the Jackson library in your user jar?
Thanks,
Max
On 18.01.19 03:40, Daniel Harper wrote:
Thanks for raising this and the PR!
In our production streaming job we’re using Kinesis, so good shout on
the
UnboundedSupportWrapper.
On 17/01/2019, 21:08, "Maximilian Michels" <[email protected]> wrote:
I'm glad that solved your GC problem. I think dipose() is a good place,
it is
meant for cleanup.
In your case the DoFn is a NOOP, so the PipelineOptions are probably
loaded
through your UnboundedSource. If both happen to be scheduled in the
same
TaskManager that is fine. However, just for precaution we should also
include
the cache invalidation in UnboundedSourceWrapper.
This way we should be good for the streaming execution. Will try to get
this
into 2.10.0.
Thanks,
Max
Issue: https://jira.apache.org/jira/browse/BEAM-6460
On 17.01.19 12:50, Daniel Harper wrote:
Max, Juan,
Just tried patching this class
https://github.com/apache/beam/blob/v2.7.0/runners/flink/src/main/java/
or
g/
apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.j
av
a#
L389 and putting the clearCache call in the finally block.
Redoing the test causes the GC to kick in (see screenshot)
I¹m not sure if this is the best place to put this clean up code
though,
is this the final place where all BEAM related stuff get terminated?
Daniel.
On 17/01/2019, 16:18, "Maximilian Michels" <[email protected]> wrote:
Hi Daniel, hi Juan,
@Daniel Thanks a lot for investigating and reporting the issue.
Your analysis looks convincing, it may be that Jackson is holding on
to
the
Classloader. Beam uses Jackson to parse the FlinkPipelineOptions.
Have you already tried to call
TypeFactory.defaultInstance().clearCache()
in a
catch-all block within your synthetic Beam job, before actually
failing?
That
way we could see if the classloader is garbage-collected after a
restart.
Let me also investigate in the meantime. We are in the progress of
getting the
2.10.0 release ready with a few pending issues. So it would be a good
time to
fix this issue.
Thanks,
Max
On 17.01.19 09:50, Juan Carlos Garcia wrote:
Nice finding, we are also experiencing the same (Flink 1.5.4) where
few jobs
are dying of OOM for the metaspace as well after multiple restart,
in
our case
we have
a HA flink cluster and not using YARN for orchestration.
Good job with the diagnosing .
JC
On Thu, Jan 17, 2019 at 3:23 PM Daniel Harper
<[email protected]
<mailto:[email protected]>> wrote:
Environment:
BEAM 2.7.0
Flink 1.5.2
AWS EMR 5.17.0
Hadoop YARN for orchestration
We¹ve noticed the metaspace usage increasing when our Flink
job
restarts,
which in turn sometimes causes YARN to kill the container for
going
beyond
its physical memory limits. After setting the MaxMetaspaceSize
setting and
making the JVM dump its heap on OOM, we noticed quite a few
instances of the
FlinkUserClassLoader class hanging around, which corresponded
with
the
number of restarts that happened.
Originally I posted this issue on the FLINK mailing list here
http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/U
se
r-
ClassLoader-leak-on-job-restart-td25547.html
After investigation I think this is related to something in
the
BEAM code,
or the way BEAM interacts with the Flink class loading
mechanism,
because I
can see the following when selecting one of the Œold¹
classloaders
-> Path
to GC Roots using Eclipse MAT in one of the heap dumps
This looks to me like this issue
https://github.com/FasterXML/jackson-databind/issues/1363
It sounds like to resolve it, user code should call
TypeFactory.defaultInstance().clearCache()when threads are
shutdown. I¹m not
sure where in the FlinkRunner codebase this should be though
To try and narrow it down as much as possible/reduce the
number
of
dependencies I¹ve managed to reproduce this with a really
really
simple job
that just reads from a synthetic unbounded source (back-ported
from
the
master branch) and does nothing
https://github.com/djhworld/streaming-job,
this will run on a Flink environment.
To reproduce the OOM I just ran the job with
MaxMetaspaceSize=125M,
and then
killed a random task manager every 60 seconds, which yielded
the
following
As you can see the number of classes increases on each
restart,
which causes
the metaspace to increase and eventually cause an OOM.
Is there anything we could do to fix this? I¹ve not tested
this
on
2.7.0
because we are waiting for 2.10 to drop so we can run Flink
1.6/1.7
on EMR
With thanks,
Daniel
----------------------------
http://www.bbc.co.uk <http://www.bbc.co.uk>
This e-mail (and any attachments) is confidential and may
contain
personal
views which are not the views of the BBC unless specifically
stated.
If you have received it in error, please delete it from your
system.
Do not use, copy or disclose the information in any way nor
act
in
reliance
on it and notify the sender immediately.
Please note that the BBC monitors e-mails sent or received.
Further communication will signify your consent to this.
---------------------
--
JC
-----------------------------
http://www.bbc.co.uk
This e-mail (and any attachments) is confidential and
may contain personal views which are not the views of the BBC unless
specifically stated.
If you have received it in
error, please delete it from your system.
Do not use, copy or disclose the
information in any way nor act in reliance on it and notify the sender
immediately.
Please note that the BBC monitors e-mails
sent or received.
Further communication will signify your consent to
this.
-----------------------------
-----------------------------
http://www.bbc.co.uk
This e-mail (and any attachments) is confidential and
may contain personal views which are not the views of the BBC unless
specifically stated.
If you have received it in
error, please delete it from your system.
Do not use, copy or disclose the
information in any way nor act in reliance on it and notify the sender
immediately.
Please note that the BBC monitors e-mails
sent or received.
Further communication will signify your consent to
this.
-----------------------------
-----------------------------
http://www.bbc.co.uk
This e-mail (and any attachments) is confidential and
may contain personal views which are not the views of the BBC unless
specifically stated.
If you have received it in
error, please delete it from your system.
Do not use, copy or disclose the
information in any way nor act in reliance on it and notify the sender
immediately.
Please note that the BBC monitors e-mails
sent or received.
Further communication will signify your consent to
this.
-----------------------------
