What looks a bit strange to me is that with a running job, the
SystemProcessingTimeService should actually not be collected (since
it is
still in use)!
My guess is that something is indeed happening during that time frame
(maybe
job restarts?) and I would propose to check your logs for anything
suspicious
in there.
When I did experiments with Beam pipelines on our platform [1], I
also
noticed, that the standard fat jars that Beam creates include Flink
runtime
classes it shouldn't (at least if you are submitting to a separate
Flink
cluster). This can cause all sorts of problems and I would recommend
removing
those from the fat jar as documented in [1].
Nico
[1] https://ververica.zendesk.com/hc/en-us/articles/360014323099
On Thursday, 10 September 2020 13:44:32 CEST Piotr Nowojski wrote:
Hi Josson,
Thanks again for the detailed answer, and sorry that I can not help
you
with some immediate answer. I presume that jvm args for 1.8 are the
same?
Can you maybe post what exactly has crashed in your cases a) and b)?
Re c), in the previously attached word document, it looks like
Flink was
running without problems for a couple of hours/minutes, everything
was
stable, no signs of growing memory consumption, impending problem,
until
around 23:15, when the problem started, right? Has something else
happened
at that time, something that could explain the spike? A checkpoint?
Job
crash/restart? Load spike?
A couple of other random guesses:
- have you been monitoring other memory pools for Flink 1.4 and
1.8? Like
meta space? Growing meta space size can sometimes cause problems. It
shouldn't be the case here, as you configured XX:MaxMetaspaceSize,
but it
might be still worth checking...
- another random idea, have you tried upgrading JDK? Maybe that
would solve
the problem?
Best regards,
Piotrek
śr., 9 wrz 2020 o 19:53 Josson Paul <jossonp...@gmail.com>
napisał(a):
Hi Piotr,
*JVM start up for Flink 1.4*
*-------------------------------*
java-server-XX:HeapDumpPath=/opt/maglev/srv/diagnostics/pipelineruntime-ta
skmgr-assurance-1-77d44cf64-z8gd4.heapdump-
*Xmx6554m-Xms6554m*-*XX:MaxMetaspaceSize=512m*
-XX:+HeapDumpOnOutOfMemoryError-*XX:+UseG1GC*-XX:CICompilerCount=4
*-XX:MaxGCPauseMillis=1000*-XX:+DisableExplicitGC-*XX:ParallelGCThreads=4*
-Dsun.net.inetaddr.ttl=60-XX:OnOutOfMemoryError=kill -9
%p*-Dio.netty.eventLoopThreads=3*
-Dlog4j.configurationFile=/opt/maglev/sw/apps/pipelineruntime/resources/lo
g4j2.xml-Dorg.apache.flink.shaded.netty4.io.netty.eventLoopThreads=3-Dnetw
orkaddress.cache.ttl=120-Dnum.cores=3-
*XX:+UseStringDeduplication-Djava.util.concurrent.ForkJoinPool.common.par
allelism=3-XX:ConcGCThreads=4 *
-Djava.library.path=/usr/local/lib-Djava.net.preferIPv4Stack=true-Dapp.di
r=/opt/maglev/sw/apps/pipelineruntime-Dserver.name=pipelineruntime-Dlog.di
r=/opt/maglev/var/log/pipelineruntime-cp/opt/maglev/sw/apps/javacontainer/
resources:/opt/maglev/sw/apps/pipelineruntime/lib/*:/opt/maglev/sw/apps/pi
pelineruntime/resources:/opt/maglev/sw/apps/javacontainer/lib/*com.cisco.m
aglev.MaglevServerstartmaglev>
1. taskmanager.memory.fraction = 0.7f (This was coming to
4.5 GB. I
didn't know at that time that we could set memory fraction to
zero
because
ours is a streaming job. It was picking up the default )
2. Network buffer pool memory was 646MB on the Heap (I
think this
was the default based on some calculations in the Flink 1.4)
3. G1GC region size was 4MB (Default)
I tested this setup by reducing the JVM heap by *1GB.* It still
worked
perfectly with some lags here and there.
*JVM start up for Flink 1.8*
*------------------------------------*
a) I started with the same configuration as above. Kubenetis POD
went out
of memory. At this point I realized that in Flink 1.8 network
buffer
pools
are moved to native memory. Based on calculations it was coming
to 200MB
in
native memory. I increased the overall POD memory to accommodate
the
buffer pool change keeping the *heap the same*.
b) Even after I modified the overall POD memory, the POD still
crashed.
At this point I generated Flame graphs to identify the CPU/Malloc
calls
(Attached as part of the initial email). Realized that cpu usage
of G1GC
is
significantly different from Flink 1.4. Now I made 2 changes
1. taskmanager.memory.fraction = 0.01f (This will give more
heap for
user code)
2. Increased cpu from 3 to 4 cores.
Above changes helped to hold the cluster a little longer.
But it
still crashed after sometime.
c) Now I made the below changes.
1. I came across this ->
http://mail.openjdk.java.net/pipermail/hotspot-gc-use/2017-February/002
622.html . Now I changed the G1GC region space to *8MB
*instead of the
default 4MB*.*
2. -XX:MaxGCPauseMillis=2000 (I even tried higher in later
experiments)
3. Played around with G1RSetSparseRegionEntries
This helped to avoid the POD going out of memory. But the
Old Gen
heap issue was very evident now (Please see the attached word
document).
d) Allocated additional heap memory of *700 MB *along with the
above
changes. This also didn't help. It just prolonged the crash. Now
I need
help from others to which direction I want to take this to .
My worry is even if I upgrade to flink 1.11 this issue might still
persist.
I have attached a screenshot from Heap dump to show you the
difference
between Flink 1.4 and 1.8 the way HeapKeyedStateBackend is
created. Not
sure whether this change has something to do with this memory
issue that I
am facing.
Name Flink-1.4.jpg for the 1.4 and Flink-1.8.jpg for 1.8
Thanks,
Josson
On Wed, Sep 9, 2020 at 5:44 AM Piotr Nowojski <
pnowoj...@apache.org>
wrote:
Hi Josson,
Thanks for getting back.
What are the JVM settings and in particular GC settings that you
are
using (G1GC?)?
It could also be an issue that in 1.4 you were just slightly
below the
threshold of GC issues, while in 1.8, something is using a bit
more
memory,
causing the GC issues to appear? Have you tried just increasing
the heap
size?
Have you tried to compare on the job start up, what is the usage
and size
of JVM's memory pools with Flink 1.4 and 1.8? Maybe that can
point us in
the right direction.
My understanding on back pressure is that it is not based on
Heap
memory but based on how fast the Network buffers are filled. Is
this
correct?.
Does Flink use TCP connection to communicate between tasks if
the tasks
are in the same Task manager?.
No, local input channels are being used then, but memory for
network
buffers is assigned to tasks regardless of the fraction of local
input
channels in the task. However with just single taskmanager and
parallelism
of 4, the amount of the memory used by the network stack should
be
insignificant, at least as long as you have a reasonably sized
job graph
(32KB * (2 * parallelism + 7) * number of tasks).
What I noticed in Flink 1.4 is that it doesn't read data from
Kafka if
there is not sufficient heap memory to process data. Somehow
this is not
happening in Flink 1.8 and it fills the heap soon enough not to
get
GCed/Finalized. Any change around this between Flink 1.4 and
Flink 1.8.
No, there were no changes in this part as far as I remember.
Tasks when
producing records are serialising them and putting into the
network
buffers. If there are no available network buffers, the task is
back
pressuring and stops processing new records.
Best regards,
Piotrek
wt., 8 wrz 2020 o 21:51 Josson Paul <jossonp...@gmail.com>
napisał(a):
Hi Piotr,
2) SystemProcessingTimeService holds the
HeapKeyedStateBackend and
HeapKeyedStateBackend has lot of Objects and that is filling
the Heap
3) I am not using Flink Kafka Connector. But we are using
Apache Beam
kafka connector. There is a change in the Apache Beam version.
But the
kafka client we are using is the same as the one which was
working in
the
other cluster where Flink was 1.4.
*There is no change in Hardware/Java/Kafka/Kafka
Client/Application
between the cluster which is working and not working*
I am aware of the memory changes and network buffer changes
between 1.4
and 1.8.
Flink 1.4 had network buffers on Heap and 1.8 network buffers
are on the
native memory. I modified the Flink 1.8 code to put it back to
Heap
memory
but the issue didn't get resolved.
Mine is a streaming job so we set 'taskmanager.memory.fraction'
to very
minimal and that heap is fully available for user data.
Flink 1.4 was not using Credit based Flow control and Flink 1.8
uses
Credit based Flow control. *Our set up has only 1 task manager
and 4
parallelisms*. According to this video
https://www.youtube.com/watch?v=AbqatHF3tZI&ab_channel=FlinkForward (
*16:21*) if tasks are in same task manager, Flink doesn't use
Credit
Based Flow control. Essentially no change between Flink 1.4 and
1.8 in
*our
set up*. Still I tried to change the Credit Based Flow Control
to False
and test my setup. The problem persists.
What I noticed in Flink 1.4 is that it doesn't read data from
Kafka if
there is not sufficient heap memory to process data. Somehow
this is not
happening in Flink 1.8 and it fills the heap soon enough not to
get
GCed/Finalized. Any change around this between Flink 1.4 and
Flink 1.8.
My understanding on back pressure is that it is not based on
Heap memory
but based on how fast the Network buffers are filled. Is this
correct?.
Does Flink use TCP connection to communicate between tasks if
the tasks
are in the same Task manager?.
Thanks,
josson
On Thu, Sep 3, 2020 at 12:35 PM Piotr Nowojski <
pnowoj...@apache.org>
wrote:
Hi Josson,
2. Are you sure that all/vast majority of those objects are
pointing
towards SystemProcessingTimeService? And is this really the
problem of
those objects? Are they taking that much of the memory?
3. It still could be Kafka's problem, as it's likely that
between 1.4
and 1.8.x we bumped Kafka dependencies.
Frankly if that's not some other external dependency issue, I
would
expect that the problem might lie somewhere completely else.
Flink's
code
relaying on the finalisation hasn't changed since 2015/2016.
On the
other
hand there were quite a bit of changes between 1.4 and 1.8.x,
some of
them
were affecting memory usage. Have you read release notes for
versions
1.5,
1.6, 1.7 and 1.8? In particular both 1.5 [1] and 1.8 [2] have
memory
related notes that could be addressed via configuration
changes.
Thanks,
Piotrek
[1]
https://ci.apache.org/projects/flink/flink-docs-release-1.5/release-not
es/flink-1.5.html [2]
https://ci.apache.org/projects/flink/flink-docs-release-1.8/release-not
es/flink-1.8.html>>>>
czw., 3 wrz 2020 o 18:50 Josson Paul <jossonp...@gmail.com>
napisał(a):
1) We are in the process of migrating to Flink 1.11. But it
is going
to take a while before we can make everything work with the
latest
version.
Meanwhile since this is happening in production I am trying
to solve
this.
2) Finalizae class is pointing
to
org.apache.flink.streaming.runtime.tasks.SystemProcessingTimeService
.
This class has a finalize method. I have attached spreadsheet
(
*Object-explorer.csv*) to give you a high level view
3) The difference between working cluster and NON working
cluster is
only on Beam and Flink. Hardware, Input message rate,
Application
jars,
Kafka are all the same between those 2 clusters. Working
cluster was
with
Flink 1.4 and Beam 2.4.0
Any insights into this will help me to debug further
Thanks,
Josson
On Thu, Sep 3, 2020 at 3:34 AM Piotr Nowojski <
pnowoj...@apache.org>
wrote:
Hi,
Have you tried using a more recent Flink version? 1.8.x is
no longer
supported, and latest versions might not have this issue
anymore.
Secondly, have you tried backtracking those references to the
Finalizers? Assuming that Finalizer is indeed the class
causing
problems.
Also it may well be a non Flink issue [1].
Best regards,
Piotrek
[1] https://issues.apache.org/jira/browse/KAFKA-8546
czw., 3 wrz 2020 o 04:47 Josson Paul <jossonp...@gmail.com>
napisał(a):
Hi All,
*ISSUE*
------
Flink application runs for sometime and suddenly the CPU
shoots up
and touches the peak, POD memory reaches to the peak, GC
count
increases,
Old-gen spaces reach close to 100%. Full GC doesn't clean
up heap
space. At
this point I stopped sending the data and cancelled the
Flink Jobs.
Still
the Old-Gen space doesn't come down. I took a heap dump and
can see
that
lot of Objects in the java.lang.Finalizer class. I have
attached the
details in a word document. I do have the heap dump but it
is close
to 2GB
of compressed size. Is it safe to upload somewhere and
share it
here?.
This issue doesn't happen in Flink: 1.4.0 and Beam:
release-2.4.0
*WORKING CLUSTER INFO* (Flink: 1.4.0 and Beam:
release-2.4.0)
----------------------------------------------------
Application reads from Kafka and does aggregations and
writes into
Kafka. Application has 5 minutes windows. Application uses
Beam
constructs
to build the pipeline. To read and write we use Beam
connectors.
Flink version: 1.4.0
Beam version: release-2.4.0
Backend State: State backend is in the Heap and check
pointing
happening to the distributed File System.
No of task Managers: 1
Heap: 6.4 GB
CPU: 4 Cores
Standalone cluster deployment on a Kubernetes pod
*NOT WORKING CLUSTER INFO* (Flink version: 1.8.3 and Beam
version:
release-2.15.0)
----------
Application details are same as above
*No change in application and the rate at which data is
injected.
But change in Flink and Beam versions*
Flink version: 1.8.3
Beam version: release-2.15.0
Backend State: State backend is in the Heap and check
pointing
happening to the distributed File System.
No of task Managers: 1
Heap: 6.5 GB
CPU: 4 Cores
Deployment: Standalone cluster deployment on a Kubernetes
pod
My Observations
-------------
1) CPU flame graph shows that in the working version, the
cpu time
on GC is lesser compared to non-working version (Please see
the
attached
Flame Graph. *CPU-flame-WORKING.svg* for working cluster and
*CPU-flame-NOT-working.svg*)
2) I have attached the flame graph for native memory MALLOC
calls
when the issue was happening. Please find the attached SVG
image (
*malloc-NOT-working.svg*). The POD memory peaks when this
issue
happens. For me, it looks like the GC process is requesting
a lot of
native
memory.
3) When the issue is happening the GC cpu usage is very
high. Please
see the flame graph (*CPU-graph-at-issuetime.svg*)
Note: SVG file can be opened using any browser and it is
clickable
while opened.
--
Thanks
Josson
--
Thanks
Josson
--
Thanks
Josson
--
Thanks
Josson
