Skip to content

[FLINK-40327][state/forst] Close ReadOptions in ForStGeneralMultiGetOperation to fix native memory leak - #28985

Open
YordanPavlov wants to merge 1 commit into
apache:masterfrom
YordanPavlov:FLINK-40327
Open

[FLINK-40327][state/forst] Close ReadOptions in ForStGeneralMultiGetOperation to fix native memory leak#28985
YordanPavlov wants to merge 1 commit into
apache:masterfrom
YordanPavlov:FLINK-40327

Conversation

@YordanPavlov

Copy link
Copy Markdown

What is the purpose of the change

ForStGeneralMultiGetOperation.process() allocates a new ReadOptions() (a JNI object owning native memory) for every executed async read batch, and no code path closes it. On a ForSt + async State V2 job the resulting native memory leak is proportional to read volume. On a production workload we measured TaskManager RSS growth of 2.2–3.9 GiB/h, leading to container OOM kills every ~10–14 hours. jemalloc allocation profiling showed Java_org_forstdb_ReadOptions_newReadOptions dominating the live-allocation diff.

This PR closes the per-batch ReadOptions with try-with-resources. This is safe because:

  • db.multiGetAsList(readOptions, ...) is synchronous within the executor lambda and returns byte[] copies — nothing referencing the ReadOptions outlives the batch;
  • all early returns and exception paths are covered by try-with-resources;
  • the sync ForSt backend already treats ReadOptions as a managed, closeable resource (ForStResourceContainer#getReadOptions, registered in handlesToClose).

Brief change log

  • Wrap the per-batch ReadOptions in ForStGeneralMultiGetOperation#process in try-with-resources so it is closed after the batch completes.

Verifying this change

This change is already covered by existing tests: functional behavior of process() for value/list/map states is exercised by ForStGeneralMultiGetOperationTest (full flink-statebackend-forst suite green locally: 979 tests, 0 failures). No new test asserts the closure itself — the ReadOptions is local to the executor lambda, and we preferred not to add a test-only seam to the production code; happy to add one if preferred.

Additionally, verified in production (Flink 2.3.0, ForSt async state backend, with this class patched): TaskManager RSS growth dropped from 2.2–3.9 GiB/h to ~20 MiB/h (flat over 19+ hours) with unchanged job output.

Does this pull request potentially affect one of the following parts

  • Dependencies (does it add or upgrade a dependency): no
  • The public API, i.e., is any changed class annotated with @Public(Evolving): no
  • The serializers: no
  • The runtime per-record code paths (performance sensitive): yes — async state read path; the change adds one native object close per read batch, negligible relative to the multiGet itself
  • Anything that affects deployment or recovery: no
  • The S3 file system connector: no

Documentation

  • Does this pull request introduce a new feature? no
  • If yes, how is the feature documented? not applicable

…peration to fix native memory leak

ForStGeneralMultiGetOperation.process() creates a native ReadOptions per
async read batch and never closes it, leaking native memory proportional
to async read volume. Wrap it in try-with-resources: multiGetAsList is
synchronous within the executor lambda and returns byte[] copies, so
nothing referencing the ReadOptions outlives the batch.
@YordanPavlov

Copy link
Copy Markdown
Author

@flinkbot run azure

@flinkbot

flinkbot commented Aug 17, 2026

Copy link
Copy Markdown
Collaborator

CI report:

Bot commands The @flinkbot bot supports the following commands:
  • @flinkbot run azure re-run the last Azure build

@YordanPavlov

Copy link
Copy Markdown
Author

Full Flink CI passed on my fork. The two E2E job failures in the first attempt were runner-infra issues — minikube startup and download rate limits — green on rerun.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants