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
