Unload a key only after its state was persisted

[tobiajo]

persistPeriodicallyAndUnloadOrphaned unloads a key whenever the unload threshold is crossed, even when the persist right before it failed and ignorePersistErrors swallowed the error. KeyContext.remove also drops the offset the key held, so the partition can commit past state that never reached the store, and the next recovery skips those events.

The unload is now gated on the persist outcome: attemptToPersist returns whether the state was persisted. An unpersisted key stays loaded and keeps holding its offset until a later tick persists it. With ignorePersistErrors = true and a store that keeps failing, keys therefore accumulate instead of being unloaded.

Summary by CodeRabbit

• Bug Fixes

  • Keys are no longer unloaded when persistence fails, including when persistence errors are configured to be ignored.
  • State remains loaded at its last successfully persisted offset until persistence succeeds.

• Tests

  • Added coverage for retaining keys after a failed persistence attempt.

The defect is described under "The existing escape and its cost" in #938; that issue is about the fence, and this fix stands on its own.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Advanced

Run ID: 164a85d3-07bc-4451-9c73-e8660f9d2949

📥 Commits

Reviewing files that changed from the base of the PR and between 2d780f2 and d5ea033.

📒 Files selected for processing (2)

• core/src/main/scala/com/evolutiongaming/kafka/flow/timer/TimerFlowOf.scala
• core/src/test/scala/com/evolutiongaming/kafka/flow/timer/TimerFlowOfSpec.scala

Included review availability: Your plan provides up to 1 included review per hour; 0 remain after this review.

---

📝 Walkthrough

Walkthrough

The timer flow now persists state before unloading orphaned keys. Ignored persistence failures keep keys in memory and retain their last held offsets. A regression test covers this behavior.

Changes

Timer persistence and unloading

Layer / File(s)	Summary
Persist-before-unload control flow core/src/main/scala/com/evolutiongaming/kafka/flow/timer/TimerFlowOf.scala	attemptToPersist returns whether persistence succeeded. onTimer unloads a key only when persistence succeeds. Scaladoc describes the updated behavior.
Ignored failure regression coverage core/src/test/scala/com/evolutiongaming/kafka/flow/timer/TimerFlowOfSpec.scala	The test verifies that an ignored persistence failure keeps the key loaded and preserves its last successfully persisted offset.

Estimated code review effort: 2 (Simple) | ~10 minutes

Merge Risk: ⚪ Minimal · up to d5ea0

Orphaned keys are now retained when persistence fails under ignored-error handling, preventing their offsets from being dropped before state is stored. The intended behavior is covered by regression testing, with no current merge-blocking risk identified.

Suggested reviewers: z1kkurat

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check. Docstring coverage is scoped to functions touched by this diff. Analyzed 0 functions across 0…
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title clearly and concisely describes the main change: a key is unloaded only after its state is persisted.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}