seungjoo choi created FLINK-40645:
-------------------------------------
Summary: ForSt file cache: closed CachedDataInputStreams stay
registered in FileCacheEntry.openedStreams until the file is evicted,
accumulating on the heap
Key: FLINK-40645
URL: https://issues.apache.org/jira/browse/FLINK-40645
Project: Flink
Issue Type: Bug
Affects Versions: 2.1.3, 2.2.1, 2.3.0
Reporter: seungjoo choi
h3. Symptom
TaskManagers using the ForSt file cache die of heap exhaustion: old-generation
GC count climbs, heap reaches ~99 % within tens of minutes, then heartbeat
timeouts or {{The TaskExecutor is shutting down}}. {{jmap -histo}} shows tens
of thousands of {{S3AInputStream}} / {{CachedDataInputStream}} instances — in
our case 83,403 {{S3AInputStream}} after 24 minutes, ~650 MB including the
Hadoop IOStatistics objects each one carries.
h3. Root cause
{{FileCacheEntry.open()}} adds every {{CachedDataInputStream}} it creates to
{{FileCacheEntry.openedStreams}} (a {{LinkedBlockingQueue}}), but the only
place that removes entries is {{doRemoveFile()}}, i.e. when the file is evicted
from or deleted in the cache. While a file stays cached, every open/close cycle
of a stream on it leaves a closed stream object in the queue. Streams are
opened and closed repeatedly when the RocksDB table cache reopens SST files
({{state.backend.forst.files.open}} small relative to the number of live
files), when the per-file stream pool overflows, or when a read fails.
h3. Reproduction
Flink 2.1.3, ForSt, {{state.backend.forst.files.open=64}} with a few hundred
live SST files: ~58 reopens/s per TaskManager, heap full in ~25 minutes. With
the default {{files.open=-1}} the growth is slow (closes then only happen for
compaction input files, which are removed soon after), which is probably why
this has gone unnoticed.
h3. Fix
Remove the stream from {{cacheEntry.openedStreams}} in
{{CachedDataInputStream.close()}}, before closing the underlying streams so
that a failing close cannot skip the removal. {{LinkedBlockingQueue.remove}} is
safe against the concurrent iteration in {{doRemoveFile()}}.
Related: FLINK-<버그 1 번호> (same {{close()}} method — the two fixes are
independent but touch the same lines, so I plan one PR with one commit per
ticket). Different from FLINK-40327, which is a native ({{ReadOptions}}) leak.
I'd like to work on this — could someone assign the ticket to me?
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
