Honour fireEvery in TimerFlowOf#unloadOrphaned

[Z1kkurat]

Problem

TimerFlowOf.unloadOrphaned re-fires onTimer on every timer trigger once a key has been idle longer than fireEvery, instead of once per fireEvery. PartitionFlow.triggerTimers runs at triggerTimersInterval (1s by default) and walks every cached key, so an idle-but-not-yet-unloadable key gets checked once a second.

Waste, not a correctness bug: flush only runs on the canUnload branch and the re-registered instant is the same value each time, so no extra writes and no timer-state growth. But the cost scales with the number of in-memory keys.

Root cause

onTimer re-registered the next check as touchedAt + fireEvery, where touchedAt is when a record was last processed:

processedAt <- timers.processedAt
touchedAt    = processedAt getOrElse committedAt
...
_ <- if (canUnload) { ...flush *> context.remove }
     else register(touchedAt)

touchedAt doesn't move while the key is idle, so it's a fixed instant in the past. Once the clock passes touchedAt + fireEvery, every re-registration lands in the past, Timers.expire drops it immediately (it keeps only instants strictly after now), and the timer is due again on the next trigger.

The spin window is [lastTouched + fireEvery, lastTouched + maxIdle):

• At the defaults (fireEvery = maxIdle = 10.minutes) it is empty — the key becomes unloadable exactly when the check would first re-fire. Hence unnoticed.
• Whenever maxIdle > fireEvery it is real: fireEvery = 5.minutes, maxIdle = 1.hour spins for 55 minutes per idle key.
• Same on the maxOffsetDifference branch.

Fix

Anchor the next check to current, as persistPeriodically and persistPeriodicallyAndUnloadOrphaned already do:

-              register(touchedAt)
+              register(current)

Safe because the decision inputs (expiredAt, offsetDifference) are recomputed from a freshly-read processedAt on every fire, so when the check runs doesn't change what it decides. register's parameter is renamed touchedAt -> checkedAt so the old name stops reading as intentional.

⚠️ One behaviour change worth reviewing

Unload latency is now bounded by fireEvery, not triggerTimersInterval. An idle key used to be unloaded within ~1s of crossing maxIdle — an accident of the spin; now it waits for the next scheduled check, up to fireEvery later. That matches fireEvery's scaladoc ("How often the check should be performed") and both sibling flows, but a large fireEvery against a tight maxIdle will hold keys in memory longer. The fourth test pins it.

Tests

Five tests. A shared helper ticks the clock a minute at a time and counts actual onTimer invocations, by wrapping the flow before trigger; flush counts wouldn't show the spin, since unloadOrphaned only flushes on the canUnload branch. Four also run persistPeriodicallyAndUnloadOrphaned as a live reference, so the expected numbers come from the sibling flow rather than by hand.

test	before	after
does not fire before fireEvery has elapsed	0 ✅	0 ✅
fires once every fireEvery while idle	26 ❌	6 ✅
keeps cadence when a record arrives between checks	22 ❌	6 ✅
unloads at first check after maxIdle	removed at minute 13 ❌	minute 15 ✅
fires on every trigger when fireEvery is zero	5 ✅	5 ✅

The two that pass either way are deliberate guards. The first check is registered at acquire, a path this doesn't touch, so that one pins that the flow doesn't become eager on startup. fireEvery = 0 is a documented configuration ("perfectly fine to set these parameters to zero") that 11 pre-existing tests rely on, and both anchors degenerate to the same thing there, so the guard keeps it meaning "check on every poll". The cadence test pins the semantic choice: a record processed mid-interval must not shift the cadence — the heartbeat follows the previous check, not the last record.

core 94/94, journal 6/6, metrics 5/5, persistence-cassandra 19/19, persistence-kafka 19/19. scalafmtCheck clean. *-it-tests not run locally (need Cassandra/Kafka).

[stasimus]

(default 10 min?) instead of every ~1s trigger, growing the uncommitted-offset window
and idle-key residency with no doc/changelog note.

[Z1kkurat]

I don't think there should be a doc note here as the current doc already states for fireEvery - How often the check should be performed.. So, the doc was misleading before because with fireEvery < maxIdle (e.g. 3m and 10m) the key would be evicted immediately after maxIdle (see One behaviour change worth reviewing in the description).
As for the changelog - I'd put it into the release notes when making a GH release

[stasimus]

Re-triggering required check after org policy rollout fix

[stasimus]

Re-running policy check with updated detector

[stasimus]

Please fix comments according to AI policy

[coderabbitai]

Warning

Review limit reached

Next included review available in 15 minutes.

View limit details

Limit details: You’ve used the included review currently available.

You've used all free OSS reviews for now. Wait for the free limit to reset to keep reviewing this public repository.

Learn how review limits work.

Review configuration:

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: b9ef9c79-9e2c-4e84-a257-fec4e2703517

📥 Commits

Reviewing files that changed from the base of the PR and between 6f76364 and a23ac83.

📒 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

---

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.}