Hi WIlliam, Thanks for picking it up. It looks like Samrat's discussion [1] wasn't picked up.
Regarding your questions: 1. I also lean toward option A. 2. The configuration parameter would default to theoretical infinity to maintain the current behavior. 3. The contract states that a dirty entry always precedes a clean entry after the job is globally terminated. There should not be a scenario where createDirtyJobResultEntryIfMissingAsync finds a clean entry already present. Therefore, there shouldn't be any concern regarding this scenario. 4. The configuration issue you brought up is an unaddressed mistake in the PR implementation. The Configuration should be passed in as a parameter. Regarding the ApplicationResultStore: It looks like the code was duplicated. The implementations look quite alike. However, the use case seems different enough to warrant its own configuration parameter, in my opinion. That configuration parameter could fall back to the one in the JobResultStore if it isn't set. Best, Matthias [1] https://lists.apache.org/thread/twkwkjcf6qds9yc0sxjvvbq0khhor8fw [2] https://github.com/apache/flink/pull/23531/changes#diff-afd3d07b8765e24db7fc1ee8288ca220a7097a3e0551c11f065dd90a724db335R49 On Wed, 23 Sept 2026, 23:38 L L, <[email protected]> wrote: > 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 >
