-
[OPIK-6982] Add reaper for orphaned Redis stream consumers (#7154)
frostbyte_neo 发布于
2026-06-19 13:34:09 +00:00 - [OPIK-6982] Add reaper for orphaned Redis stream consumers
Backend processes register a unique consumer-- per process that
Redis only removes on graceful shutdown (XGROUP DELCONSUMER). Non-graceful exits
(OOMKill/SIGKILL/crash/eviction) leak the consumer permanently, so stream consumer
groups accumulate tens of thousands of orphans across deploys and crashes.Add a periodic reaper that, for each registered subscriber's stream, lists groups
(XINFO GROUPS) and removes consumers that are both idle beyond a threshold and have
no pending entries. The pending == 0 guard ensures un-acked work is left for
XAUTOCLAIM to reclaim and never destroyed.- StreamConsumerReaper: reactive reap logic + metrics
- StreamConsumerReaperJob: Quartz job, discovers streams from BaseRedisSubscriber
Guice bindings (no keyspace scan, no hand-maintained list), runs once per cycle
via a distributed lock with hold-until-expiry - StreamConsumerReaperConfig: enabled / jobInterval (1h) / idleThreshold (1d) /
lockDuration (10m), env-overridable - BaseRedisSubscriber exposes getStreamName() for discovery
Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- [OPIK-6982] Address review: shared binding helper, reaper error handling, test config
- Extract GuiceBindings.boundRawTypes() and use it in both StreamConsumerReaperJob
and EventListenerRegistrar so the Guice binding-walk lives in one place (also adds
a missing null-filter in EventListenerRegistrar). - StreamConsumerReaper.reapStream: only swallow "no such key" (stream not created yet)
at debug; route other listGroups errors through reaperErrors + warn, and pass the
throwable as the last log arg so the stack trace is printed. - Disable streamConsumerReaper in config-test.yml (background jobs are disabled in
tests by convention); the job test resolves the bean via the Injector instead.
Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- [OPIK-6982] Address review: log full exception with quoted placeholder in EventListenerRegistrar
Pass the exception to SLF4J (instead of e.getMessage()) so the stack trace is recorded,
and quote the placeholder per the backend logging convention.Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- [OPIK-6982] Address review: document no-such-key match from verified payload, doc/style nits
Verified against a real Redis that XINFO GROUPS on a missing stream raises a generic
org.redisson.client.RedisException with message "ERR no such key..." — no typed exception
exists, so the substring match stays; comment updated to cite the confirmed payload.
Also: reword lockDuration javadoc (TTL suppresses other instances, does not cancel an
in-flight pass) and drop the no-op @Valid on the primitive enabled field.Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- [OPIK-6982] Address review: proper interruption, config record, single result-tagged metric
Address @andrescrz review feedback:
- StreamConsumerReaperJob: Lombok constructor, field/method Javadocs, proper
reactive interruption via Disposable (mirrors ExperimentProjectMigrationJob),
subscribe on boundedElastic, log values at end of message, lock-skip at debug - StreamConsumerReaper: merge reaped/error counters into one counter with a
result attribute, early-return on null/empty input, order-preserving dedup - StreamConsumerReaperConfig: convert to record with @Builder(toBuilder), drop
Java defaults (populated from config yaml), tighter bounds, add startupDelay - GuiceBindings: @UtilityClass, @NonNull on public method param
- OpikGuiceyLifecycleEventListener: schedule reaper with configurable startupDelay
- config.yml: idleThreshold default 1d -> 3d, add startupDelay
- StreamConsumerReaperTest: assert orphan consumer exists before reaping
- fix(reaper): bind reaper config via @Config so the record instance resolves from yaml
The record StreamConsumerReaperConfig has no no-arg/@Inject constructor, so Guice
could not construct it for StreamConsumerReaperJob (ConfigurationException at boot,
failing StreamConsumerReaperJobTest). Inject it with @Config("streamConsumerReaper")
like ExperimentProjectMigrationJob, which binds the parsed yaml instance — this also
fixes a latent bug where the previous @Data class was resolved as a fresh Guice-built
default, ignoring the yaml-configured values.
Co-authored-by: Claude Opus 4.8 (1M context) noreply@anthropic.com
下载附件