Hi all,

I'd like to pick up
[FLINK-32806](https://issues.apache.org/jira/browse/FLINK-32806)
("EmbeddedJobResultStore keeps the non-dirty job entries forever").
Before opening a PR
I'd like to align on the approach, since there is an open design
question around a new
config option.

For context: I'm a new contributor, with a few merged PRs in the
flink-agents repo
(#1135, #1142, #1144), and I'd like to start contributing to Flink core as well.

**Current state (checked against master today):**

`EmbeddedJobResultStore` keeps two plain `HashMap`s.
`markResultAsCleanInternal` moves an
entry from `dirtyJobResults` into `cleanJobResults`, and nothing ever
removes entries from
`cleanJobResults`. In a long-running session (non-HA) this map
therefore grows without
bound, one entry per finished job.

For comparison, `FileSystemJobResultStore` does not retain clean
entries at all: with
`job-result-store.delete-on-commit` (default `true`) it deletes the
underlying file when
the entry is marked clean. So the in-memory and the file-system
implementations currently
diverge in behaviour.

**Same pattern elsewhere:** `EmbeddedApplicationResultStore` keeps a
`cleanResults` map with
the identical never-evicted behaviour. It is documented as "for
testing purposes", so the
impact is much smaller — but should I include it in the same change,
or keep this PR
scoped to the job result store only?

**Prior work:**

PR #23531 (by Samrat002) addressed this by converting
`cleanJobResults` into a Guava
cache with a TTL and adding a new option `job-result-store.ttl-clean-job-result`
(default 10 minutes). It was closed by the stale bot after 180 days
without activity.
Samrat002 noted in that PR that adding the parameter might warrant a
discussion here —
which seems to be the missing step, so I'd like to have it now.

One problem with that implementation: it builds the cache with `new
Configuration()` —
an *empty* configuration — so the new option always resolves to its
default value and any
user setting would be silently ignored. Making the option actually
work means passing the
real `Configuration` into `EmbeddedJobResultStore`, which today is
created with a no-arg
constructor in `AbstractNonHaServices` (plus ~12 test call sites).
This is the part I'd
most like guidance on.

**Two options:**

*A — keep the previous approach (my preference):*
Convert `cleanJobResults` to a Guava cache bounded by a TTL,
controlled by a new config
option. Pros: bounded memory, behaviour is configurable and backward
compatible for users
who rely on `hasCleanJobResultEntry` within the TTL window. Cons: adds
a new config option.

*B — no new config, align with `delete-on-commit`:*
Make `EmbeddedJobResultStore` honour the existing
`job-result-store.delete-on-commit`
(default `true`) and simply drop the entry when it is marked clean, matching
`FileSystemJobResultStore`. Pros: no new option, fixes the growth
under the default
configuration. Cons: `hasCleanJobResultEntry` would always return
`false` under the
default config — a visible semantic change.

**One thing I'd like to confirm:**

`Dispatcher#createDirtyJobResultEntryIfMissingAsync` uses
`hasCleanJobResultEntryAsync`
as an idempotency guard (it skips creating a dirty entry if a clean
one exists). My
understanding is that this is low risk in the in-memory store, because
`EmbeddedJobResultStore` is used in non-HA setups and its contents are
lost on process
restart anyway — so an expiring entry is equivalent to a restart. Is
that reasoning
correct, or is there a case where the clean entry must outlive the TTL?

**Questions:**

1. Do we prefer A or B?
2. If A: is a new config option acceptable, and what should the default TTL be?
3. Any other concern about the Dispatcher idempotency guard?
4. Given the `new Configuration()` problem: is it acceptable to change
   `AbstractNonHaServices` to accept a `Configuration` (touching its
subclasses and ~12
   test call sites), or do we prefer to leave construction untouched
and use a fixed TTL?

I already have a minimal local implementation ready (Guava cache with
a fixed TTL, no
constructor changes) that I can adapt to whichever option we pick, and
I'll add coverage in
`EmbeddedJobResultStoreContractTest` / a dedicated unit test.

Thanks,
William

Reply via email to