[OPIK-8192] [BE] fix: drop undecodable scoring stream messages instead of wedging the stream - #8089
Merged
thiagohora merged 19 commits intoSep 3, 2026
Conversation
…d of wedging the stream OPIK-8164 fixed the codec's Jackson limits and added an opt-in publisher guard, which raised the ceiling but did not remove it. Above the new limit the stream still wedges identically, because the decode happens inside Redisson's CommandDecoder -- below BaseRedisSubscriber, before any StreamMessageId exists. With no id there is nothing to ack, retry or remove, so the entry is redelivered forever (delivery-count 2,915 over 20 days in production) and at consumerBatchSize > 1 strands every healthy entry claimed with it. That is how one oversized trace took two scoring streams to pending == XLEN and 19.66 GiB. The drop machinery already existed: postProcessFailureMessages acks and removes non-retryable failures. The failure simply never reached it. So this changes where the decode failure surfaces, not what happens to it. - RedisStreamCodec.JAVA wraps its value decoder so a payload it cannot decode yields an UndecodableStreamMessage instead of throwing. Encoders are handed through untouched, so the wire format is byte-identical and a rolling upgrade is safe in both directions. - BaseRedisSubscriber.processMessage recognises the sentinel, counts it on a new *_undecodable_messages_total counter, logs the messageId, stream and payload size at error, and fails it as non-retryable so the existing path acks and removes it. Effect: a permanent wedge becomes one dropped scoring job, and streams already wedged self-clear on deploy. maxRetries also becomes reachable on this path, which it never was -- the failure used to happen below the layer that counts deliveries. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
github-actions
Bot
added
java
Contributor
⏱️ pre-commit per-hook timing
⏭️ 42 skipped (no matching files changed)
|
![baz-reviewer[bot]](https://avatars.githubusercontent.com/in/933528?s=60&v=4){kind=small}
…eld-name path The first cut deleted an undecodable entry on first delivery, because IllegalArgumentException is non-retryable. Review found that trades a recoverable wedge for irreversible loss: "this pod cannot decode it" is not "nobody can". The reader's maxStringLength is configuration, and during a rolling upgrade an older pod fails on a payload a newer one reads -- exactly the family FAIL_ON_UNKNOWN_PROPERTIES=false and LenientUUIDDeserializer were added to survive. Blast radius was all seven java-codec streams. - New retryable UndecodablePayloadException, so maxRetries governs and only the exhausted case is acked and removed. Still terminates, so a genuinely poisonous payload cannot wedge the stream. - The wrapper now goes around the CompositeCodec, not its value codec. The composite routes the map-KEY path to LZ4CodecV2 over Kryo5, so the field name was decoded by an unwrapped decoder -- the same wedge, one field over. Wrapping the inner JSON codec never covered it. - Explicit null-payload guard: a missing payload field, or a field name the now-tolerant key decoder could not decode, is retired the same way instead of reaching processEvent as null. - The counter is tagged with error_type and stream, and its description no longer claims to count drops -- it counts detections, once per delivery, which is what a retryable failure produces. - Corrected two overclaims: the sentinel javadoc said such a payload "will never become decodable" (false for the size-limit case), and the wrapper javadoc justified wrapping getValueDecoder with a reason that does not hold -- the two-arg CompositeCodec leaves valueCodec null, so that accessor NPEs before any wrapper is consulted. Tests: 8 codec (map-key tolerance, partial-buffer drain, Error pass-through, which failure was caught), 3 subscriber (not removed before maxRetries, removed once reached, no-payload retired). 28 unit + 7 Testcontainers green. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Contributor
Backend Tests - Integration Group 11 37 files - 4 37 suites - 4 8m 58s ⏱️ + 4m 16s Results for commit 30c30a2. ± Comparison against base commit d18f7e1. This pull request removes 29 and adds 14 tests. Note that renamed tests count towards both.♻️ This comment has been updated with latest results. |
…osing the cause Cycle 1's null-payload guard conflated two failures that deserve opposite treatment, and silently reclassified a pre-existing one. - A field name the map-key decoder could not decode puts the sentinel among the KEYS. Nothing inspected key objects, so the cause was discarded outright -- the very path cycle 1 added the wrapper to close was retried and then deleted with no record of why. Now the key set is inspected, the cause is logged and labelled error_type= undecodable_field_name, and it stays retryable because LZ4/Kryo skew can differ between pods. - A genuinely absent payload field is deterministic, so it keeps its pre-existing non-retryable, remove-on-first-delivery behaviour. Cycle 1 had made it retryable, giving a permanently malformed entry two extra 10-minute pending windows for nothing. - Both branches returned before startMillis, so undecodable entries vanished from the queue-delay histogram on every delivery -- the one signal that shows an entry cycling in a growing PEL. They now record it, and the adjacent comment no longer claims coverage it does not have. - STREAM_KEY was a third private copy of an identical AttributeKey; promoted to ErrorMetricsResolver and shared with StreamConsumerReaper. - Removed a comment that said "Drop it: it will never become decodable" directly above code that deliberately does not drop it. - UndecodablePayloadException's rationale claimed a peer pod with a higher maxStringLength would decode the same bytes. Since #8060 that is false in steady state -- every pod builds its codec after JsonUtils.configure() from one config key. The retryable classification stands on jar skew during a rolling upgrade; the javadoc now says so and admits retrying a size breach is usually wasted work. Tests: the absent-field case asserts listPending is never consulted, which is what separates non-retryable from retryable; the field-name case asserts it is not removed below maxRetries. Mutation-checked: collapsing the two back into one classification fails the first. 36 green (29 unit, 7 Testcontainers). Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…8 GiB test allocation Two high-severity findings, both in tests, both real. The branch was RED on a suite I had not run. BaseRedisSubscriberTest still encoded the pre-OPIK-8192 null-payload contract: shouldHandleNullPayloadSuccessfully asserted processEvent is invoked with null, and shouldContinueProcessingAfterFailedMessages derived its failure count from the NPE that null produced. The new short-circuit means processEvent is never reached for such an entry, so the first timed out and the second failed. Confirmed by running it: 14 tests, 1 failure, 1 error. Rewritten to the new contract -- the payload-less entries leave the stream without reaching processEvent, and the continue-processing test now raises its failure from a real payload, which is what it always meant to test. shippedJavaCodecToleratesUndecodableMapKey fed free text to LZ4CodecV2's decoder, whose first act is to read four bytes as a decompressed-length header and allocate that array with no sanity bound. "plai" is 1,886,151,017, so the test asked for 1.76 GiB. Surefire sets no -Xmx, so on a runner with <= 8 GB RAM the default cap is below that and the allocation throws OutOfMemoryError -- which the wrapper deliberately does not absorb, so the test would fail rather than produce the sentinel it asserts, and where it did pass it allocated 1.76 GiB inside the shared surefire JVM. Now an explicit 4-byte length prefix of 16, with a comment recording why it must stay explicit. 50 green: 14 BaseRedisSubscriberTest, 21 BaseRedisSubscriberUnitTest, 8 FaultTolerantStreamCodecTest, 7 RedisStreamCodecTest. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…l's cause Three of the seven open bot findings; the rest are answered on the PR. - error_type=missing_payload was being reported on *_undecodable_messages_total, so a decoder alert would fire on a deterministic malformed entry where nothing failed to decode. That path now records only queue delay; messageProcessingErrors already counts it. - UndecodableStreamMessage.cause is @nonnull. Only the codec builds it and it always has a cause, so the nullable component was an invalid state the type allowed rather than one anything reaches. - UndecodablePayloadException claimed retirement "always terminates". getDeliveryCount maps a failed listPending to 0, so while a PEL lookup is failing persistently the count never reaches maxRetries and the entry keeps being redelivered. Pre-existing for every retryable failure, but the javadoc now says "terminates while the PEL is readable" rather than overstating it. 50 green. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Contributor
Backend Tests - Integration Group 9 42 files 42 suites 13m 4s ⏱️ Results for commit bdd1895. ♻️ This comment has been updated with latest results. |
Contributor
…try budget - processMessage's doFinally still had its own copy of the queue-delay recording after the helper was extracted, so the normal and early-return paths could drift apart. Both go through recordQueueDelay now. - maxRetries is a fleet-wide budget, not per pod, so an older pod can burn all of it before a newer decoder claims the entry. The retryable classification is therefore a chance for a newer decoder, not a guarantee -- three chances against the one that dropping on first delivery gives. Documented with the numbers (~30 min at the shipped defaults, which normally outlasts a rolling upgrade) rather than left implying more than it delivers. 50 green. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…did not contain Correcting the record. The javadoc edit described in 4c2afe1 ("says 'terminates while the PEL is readable' rather than overstating it") and again in 0ddb90c ("documented with the numbers") never landed -- the scripts that made those edits aborted before reaching this file, and both commit messages went out describing it anyway. The code changes in both commits are as described; only this javadoc was missing. What it now records, both of which weaken claims I made earlier in review: - getDeliveryCount maps a failed listPending to 0, so retirement stalls while a PEL lookup is failing. "Always terminates" is "terminates while the PEL is readable". - maxRetries is a fleet-wide budget, not per pod. An older pod can burn it before a newer decoder claims the entry, so retrying is a chance for a newer decoder rather than a guarantee -- three chances against one. ~30 min at the shipped defaults, which normally outlasts a rolling upgrade but is not guaranteed to. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
'three chances against the one that ...' obscured the comparison; now 'three chances instead of the single chance that deleting on first delivery would give'. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
thiagohora
marked this pull request as ready for review
September 2, 2026 11:47
…o-op annotation Five review findings. Two of them corrected claims I had made and verified against the wrong artifact. @JetoPistola caught that my new javadoc described the wrong class. I had disassembled redisson-3.50.0 -- picked by `find | head -1` out of the 31 redisson jars in ~/.m2 -- while pom.xml sets redisson.version=4.7.0. In 4.7.0 LZ4CodecV2$1.decode is `readInt()` then `newarray byte`, a raw heap array; `ByteBufAllocator.DEFAULT.buffer(int)` is the V1 LZ4Codec$1 pattern, which is not wired here. Re-disassembled 4.7.0 to confirm both halves. Consequences: the OutOfDirectMemoryError parenthetical was impossible on this path and is gone -- a heap array yields OutOfMemoryError: Java heap space only. The ordering claim the guard rests on survives unchanged, and the javadoc now carries the actual offsets (newarray at 6, decompressor at 24, readFully at 44) plus a note distinguishing V1 so the next person re-checking a future version does not conflate them. Also from @JetoPistola: @nonnull on an abstract method parameter generates nothing -- Lombok injects at the start of a method body and an abstract method has none. Verified: the abstract method's bytecode has no Code attribute at all, and since parameter annotations are not inherited, the seven subscribers got nothing from it. Exactly four do not annotate their own override, OnlineScoringBaseScorer among them -- the class my javadoc cites as the one that used to NPE. So it read as protecting the one case it did not. Removed rather than propagated to the overrides: the invariant is established at one place, the guards in processMessage, and adding checks to implementations would only guard a condition the caller excludes. The javadoc now says all of this instead of implying enforcement. Baz findings, all correct: - Inserting the LengthBoundedLz4Codec block left FaultTolerantCodec's javadoc orphaned, documenting the constant that followed it instead. Reordered so every javadoc adjoins its declaration. - MAX_MAP_KEY_DECODED_LENGTH bounds both decoder accessors, not just the field-name one, so the name was misleading -> MAX_LZ4_DECODED_LENGTH, with the error message widened to match. - The two rejection tests differed only in four bytes; combined into one @ParameterizedTest, and widened to Integer.MAX_VALUE and Integer.MIN_VALUE beyond the original two values. 56 green across the four affected suites. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Contributor
|
👋 Re-review summary (head Everything I raised is resolved — five findings across three rounds, and in each case the fix went past what was asked.
On the last two, verified rather than assumed The rewritten The Unprompted improvements in the same commit, neither of which I raised:
Overall The LZ4 length bound stands out across the whole PR: you found a second, independent route to the same OPIK-8164 wedge — an unbounded allocation ahead of frame validation — while fixing the first one, and closed it with boundary tests that assert the guard does not fire at the ceiling and on a short buffer. Guarding against a bound silently widening later is the part most people skip. Inline comments this round: none — nothing new found. One item for a separate ticket, not this PR. Confirming your own reviewer note with measurement: on the No change here, and it should not ride along: swapping the 🤖 Review posted via /review-github-pr |
JetoPistola
previously approved these changes
Sep 3, 2026
JetoPistola
left a comment
JetoPistola
left a comment
Contributor
There was a problem hiding this comment.
Approving. The LZ4 length bound is the standout — a second, independent route to the same OPIK-8164 wedge, found while fixing the first one and closed with boundary tests that assert the guard does not fire at the ceiling or on a short buffer.
Two non-blocking follow-ups, neither for this PR:
-
Uncompressed trace payloads on the
JAVAcodec.CompositeCodecroutes map-values throughJsonJacksonCodecand only map-keys through LZ4, so LZ4 compresses the seven-character field name while the multi-MB trace payload goes to the wire uncompressed (confirmed by round-trip — plain readable JSON in the buffer). Consistent with OPIK-8164's 19.66 GiB. Fixing it means swapping theCompositeCodecarguments, which is a wire-format change needing a two-phase rollout — worth its own ticket. -
Terminal bookkeeping for dropped entries.
ExperimentItemProcessingSubscriberdecrements its batch counter insideprocessEvent, so an entry retired before that point never finalizes its batch. Pre-existing —postProcessFailureMessageshas always dropped non-retryable failures without terminal handling — and correctly out of scope here, but it outlives this PR.
Also worth removing the now-redundant onlineScoring.dropOversizedPayloads publisher guard from #8060 deliberately, rather than leaving two mechanisms in place.
…lared length Replaces LengthBoundedLz4Codec (net -102 lines). The ceiling it enforced was an undocumented number of my own choosing: nothing in Redis or Redisson constrains a stream field name, so 4096 rested only on the local observation that every field name in this codebase is the constant "message" -- an assumption that would silently discard data the day someone picks a longer one. Measured what the raw LZ4CodecV2 decoder actually does per declared length, which showed the pre-check was doing much less than its javadoc implied (9 GB heap): negative -> NegativeArraySizeException (Exception, already caught) 7 (legitimate) -> IOException (Exception, already caught) 1,886,151,017 -> IOException (allocated 1.8 GiB, then failed) Integer.MAX_VALUE -> OutOfMemoryError (Error, escaped) Only the last row needed guarding, and which row a length lands in moves with -Xmx: 1.8 GiB is an IOException on a 9 GB heap and an OutOfMemoryError on a 1 GB container. So the fix is to widen the catch to Exception | OutOfMemoryError, which makes the behaviour heap-independent without inventing a limit. No other Error is absorbed -- StackOverflowError and an OOM from a legitimate multi-hundred-MB payload document still propagate. Two residual risks, documented in the javadoc rather than papered over: an allocation large enough to fail can starve a different thread, which then throws where nothing catches it; and a length the heap can satisfy is still allocated in full before it fails. Both are properly fixed by the CompositeCodec argument-order change, which takes LZ4 off the field-name path and dissolves the hazard instead of bounding it. Tests: the four ceiling tests collapse into one parameterized case asserting the sentinel for negative, MIN_VALUE, 1.8 GiB and MAX_VALUE lengths, plus a direct OOM-absorption test and a StackOverflowError propagation test for the boundary. Mutation check on the OOM arm does not fail an assertion -- it kills the forked JVM (Tests run: 0, BUILD FAILURE), which is a stronger signal but easy to misread, so noting it here. 55 green across the four affected suites. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Prompted by two review questions I could not answer from the code, so I
measured instead of reasoning. Both changed what the javadoc claims.
1. Is the OPIK-8164 failure an OOM? NO. Through the shipped JAVA codec's
value decoder at the real 20,000,000 limit:
19,999,999 chars -> decodes fine
20,000,001 chars -> StreamConstraintsException, isError=false
An ordinary IOException, matching the production stack trace verbatim.
The Exception arm has always handled the literal incident; my previous
commit framed the OOM arm as if it closed a hole related to it, which
was wrong.
2. But both modes are real, on the SAME path. A payload UNDER
maxStringLength but larger than the heap OOMs inside Jackson's own
String materialization. On a -Xmx64m fork, 12,000,000 chars (well under
the 20,000,000 limit) gave cause=java.lang.OutOfMemoryError,
isError=true, absorbed into the sentinel. This is the case that makes
the OOM arm matter in production, where maxStringLength ships at 100 MB
and consumerBatchSize is 10.
3. Does absorbing an OOM recover? For this shape, yes, measured: three
consecutive rounds of oversized-then-ordinary on a 64 MB heap each
absorbed the OOM and then decoded the ordinary message correctly, free
heap stable, no cumulative degradation. The array that failed was never
allocated, so the failure consumed nothing.
The operational argument settles it: the helm chart ships
-XX:+UseG1GC -XX:MaxRAMPercentage=80.0 with NO
-XX:+ExitOnOutOfMemoryError, so the process already survives an OOM
today. Not absorbing does not buy a clean restart -- it buys a
still-running pod with a permanently wedged stream. If the team ever
adds ExitOnOutOfMemoryError the JVM exits at throw time and this arm
becomes unreachable, which is fine.
Documented what it does NOT claim: a JVM under genuine heap exhaustion
is not healthy, this arm can absorb an OOM that was a symptom rather
than a cause, and the recovery measured is single-threaded.
Tests: the over-limit case now asserts the cause is StreamConstraintsException
and is NOT an Error, which is the distinction both questions turned on. Also
removed a shipped-codec test I had just added -- it needed an assumeTrue on
the memoized limit and skipped in the real suite, because RedisStreamCodecTest
calls JsonUtils.configure in the same JVM. The small-limit mapper exercises
the identical path; the real-limit numbers live in the javadoc as measured
evidence instead.
58 green, 0 skipped.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…nd to end
You were right to push on this, and my earlier wording was wrong. I said
the Exception arm "has always handled the literal incident" -- I meant
since the first commit OF THIS PR, which introduced the wrapper. Pre-PR
there was no try/catch in the codec at all, so the StreamConstraintsException
propagated up through CommandDecoder exactly as the original OPIK-8164
analysis said. Nothing of ours caught it.
And the drop mechanism was not actually tested for a real decode failure.
Every existing test either injected a pre-built UndecodableStreamMessage
into a mocked readGroup, or exercised the decoder in isolation. None
reproduced the thing that happened: written successfully, then failing on
the way back out below the subscriber.
New BaseRedisSubscriberTest.UndecodablePayloadTests does that on real
Redis -- a JAVA-shaped codec with a 512-byte string limit standing in for
production's 100 MB, so the write/read asymmetry reproduces at a few
hundred bytes. It asserts the entry writes fine, the healthy entry behind
it still processes, the undecodable one is NOT removed on first delivery,
and that it does leave once the delivery count reaches maxRetries.
Mutation-checked, and the mutant is the incident: making tolerant() a
no-op produces
io.netty.handler.codec.DecoderException:
com.fasterxml.jackson.core.exc.StreamConstraintsException:
String value length (513) exceeds the maximum allowed (512, ...)
which is the production stack trace's exact shape, and the test fails at
the FIRST await -- the healthy entry behind it never processes either.
That is the batch-stranding half of the wedge, now covered.
Two things the run exposed, both documented in the test:
- Retirement is driven by the delivery count, which only advances when
XAUTOCLAIM redelivers after pendingMessageDuration. At the shipped
online-scoring values (10m, maxRetries 3) an undecodable entry sits in
the stream for ~30 minutes, re-read and re-decoded each cycle, before
removal. Bounded, not eliminated -- and at production payload sizes each
cycle re-attempts a large materialization. The test compresses the
timers to 500ms/2 to stay fast.
- My first attempt at this test published before subscriber.start(), so
the consumer group was created at '$' and never saw the entries. That
was a test bug, not a code one, but it is why the ordering is now
explicit with a comment.
RedisStreamCodec.faultTolerant widens from package-private to public
@VisibleForTesting so the subscriber test package can build the codec.
59 green, 0 skipped.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
You are right that this was circling, and the cause was my own pattern: each question got a scratch probe, a reported finding, a DELETED probe, and a javadoc paragraph. Prose does not fail when it goes stale, so every question reopened settled ground. Audited all 18 assumptions established by measurement in this PR. 14 were already pinned by tests. Of the four that were not: - CompositeCodec argument order / map values being uncompressed JSON. Two reviewers queried what encodedBytes measures and the answer depends entirely on this, so it is now a test: the value encoder must produce plain readable JSON, the key encoder must not, and the key must still round-trip. Swapping those arguments -- the pre-existing fix this PR keeps deferring -- now fails a test and forces the encodedBytes contract to be revisited with it. - Real heap exhaustion, and recovery across repeated rounds. Both need a JVM small enough to exhaust and the surefire fork is not, so they stay manual evidence -- now labelled as such in the javadoc, with the -DargLine="-Xmx64m" reproduction, instead of reading like verified claims. What IS pinned is the arm (heapExhaustionDuringMaterializationIsAbsorbed, via a stub) and its boundary (nonOomErrorStillPropagates). - The ~30 minute retirement window at shipped config values. Deliberately not added: that is a configuration-value assertion, and @andrescrz dismissed exactly that pattern on #8038 ("we typically don't unit test configuration"). It stays documented in the test that compresses the timers. 60 green, 0 skipped. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Contributor
Backend Tests - Integration Group 14 51 files 51 suites 3m 28s ⏱️ Results for commit 316db5a. ♻️ This comment has been updated with latest results. |
…ytes in tests Three fixes; the reported failure and one review finding were both my own mistakes. The reported failure (expected 1 but was 2) was a race I wrote, not a code defect -- the log shows the sentinel handled correctly. It asserted size()==1 immediately after successCount==1, but ack and remove are asynchronous and batched (postProcessSuccessMessages runs after bufferTimeout), so processEvent completing does not mean the healthy entry's XACK/XDEL has landed. Removed rather than made timing-dependent: that the undecodable entry survives its first delivery is already pinned deterministically by the mocked unit tests, which control the delivery count instead of inferring it from wall-clock ordering. Ran 5x clean. Review: the parameterized LZ4 case fed 1_886_151_017 and Integer.MAX_VALUE to the real decoder, which allocates them for real -- a transient 1.8 GiB inside the shared surefire JVM. That is the exact hazard an earlier commit in this PR fixed and I reintroduced. Bounded to lengths that fail without a large allocation (negatives throw before any array exists, 1024 allocates a kilobyte then fails parsing). The OOM arm stays covered by heapExhaustionDuringMaterializationIsAbsorbed and the measured outcomes stay in the javadoc. Review: the integration test asserted XLEN reaches zero, which proves XDEL but not XACK -- a broken ack leaves the id in the consumer group's PEL where XAUTOCLAIM keeps reclaiming it, a wedge invisible to XLEN. Now captures the id from add() and asserts the PEL is empty, via listPending rather than pendingRange so the assertion does not itself have to decode the undecodable payload. Note Redisson's reactive listPending COMPLETES EMPTY rather than emitting an empty list, so block() returns null on the passing path -- defaultIfEmpty is required and the first version of this assertion failed on exactly that. 56 tests, 3 consecutive clean runs, 0 skipped. Counted with surefire-reports cleared first: earlier totals in this PR were inflated by stale reports from prior runs. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
JetoPistola
reviewed
Sep 3, 2026
| * a cause, and keep consuming — masking it. The recovery measured above is single-threaded and | ||
| * proves the decode path, not a loaded service; an allocation large enough to fail can still starve | ||
| * a different thread, which throws where nothing catches it. The counter to that is a size guard at | ||
| * publish time, which is what {@code onlineScoring.dropOversizedPayloads} does, not anything this |
Contributor
There was a problem hiding this comment.
💡 suggestion | Documentation accuracy
onlineScoring.dropOversizedPayloads does not exist. git grep -rn "dropOversizedPayloads" over the repo returns exactly one hit — this sentence. It is not a field on OnlineScoringConfig (I read the class), not in config.yml, not anywhere in src.
I checked whether it might have been renamed or landed elsewhere: #8060 was merged (09211d68d6), but as "apply Jackson stream-read limits to the Redis stream codec" — the Guice/Dropwizard ordering fix that made the codec actually receive maxStringLength. It never added a publisher-side drop guard, under this or any name.
That matters more than a stale reference usually would, because of the job this sentence is doing. The paragraph above it makes the most serious admission in the design: under real heap pressure this arm will absorb an OOM that was a symptom rather than a cause, and keep consuming — masking it. This sentence is the discharge for that admission: the residual risk is acceptable because a publish-time size guard covers it. With no such guard, that risk is open, not delegated, and someone auditing the decision later would go looking for a control that was never built.
Worth stressing that this is the only thing I would change here. The OOM-absorbing call itself I checked and agree with: values.yaml:179 ships -XX:+UseG1GC -XX:MaxRAMPercentage=80.0 with no ExitOnOutOfMemoryError anywhere in the repo, so the process already survives an OOM — not absorbing buys a live pod with a wedged stream, which is strictly worse. The multi-catch compiles and StackOverflowError still propagates, as nonOomErrorStillPropagates pins.
Suggested change
| * publish time, which is what {@code onlineScoring.dropOversizedPayloads} does, not anything this | |
| * publish time. No such guard ships today -- #8060 applied the codec's stream-read limits but added | |
| * no publisher-side size check -- so this residual risk is open rather than delegated. It is not | |
| * something this codec can address on read. |
🤖 Review posted via /review-github-pr
Contributor
|
👋 Re-review summary (head This round reviewed a genuine design reversal, so it got more scrutiny than a polish pass: the LZ4 length bound added in I checked the reversal rather than taking it on trust, and it holds.
Two things I nearly flagged and withdrew after checking properly — recording them because the checking is the point:
Removing the bound was also the right call, and worth saying since I praised it last round. 4096 was an undocumented ceiling resting on the local observation that every field name is the constant What the new commits add
Inline comments: 1 suggestion — nothing blocking. The one finding: 🤖 Review posted via /review-github-pr |
JetoPistola
approved these changes
Sep 3, 2026
JetoPistola
left a comment
JetoPistola
left a comment
Contributor
There was a problem hiding this comment.
Approving — nothing blocking.
Reviewed the OOM-absorption reversal carefully since it inverts the earlier Exception-not-Throwable boundary, and the argument holds: the helm chart ships no ExitOnOutOfMemoryError, so the process already survives an OOM. Not absorbing buys a live pod with a wedged stream, which is worse than dropping one message. The four-row failure-mode table and the behavioural pin on the CompositeCodec argument order are the strongest artifacts here.
Three non-blocking follow-ups, none for this PR:
-
onlineScoring.dropOversizedPayloadsis cited but does not exist (left inline). It is the stated mitigation for the design's own admitted masking risk, so that risk is currently open rather than delegated. Worth either building the guard or restating the position — a comment fix, not a code one. -
Uncompressed trace payloads on the
JAVAcodec.CompositeCodecroutes values throughJsonJacksonCodecand only keys through LZ4, so LZ4 compresses the seven-character field name while multi-MB payloads go to the wire uncompressed (confirmed by round-trip). Consistent with OPIK-8164's 19.66 GiB. Swapping the arguments is a wire-format change needing a two-phase rollout —compositeCodecPutsLz4OnTheKeyPathNotTheValuePathalready fails if anyone does it without revisiting theencodedBytescontract. -
Terminal bookkeeping for dropped entries.
ExperimentItemProcessingSubscriberdecrements its batch counter insideprocessEvent, so an entry retired before that point never finalizes its batch. Pre-existing and correctly out of scope, but it outlives this PR.
2 tasks
thiagohora
deleted the
thiagohora/OPIK-8192-drop-undecodable-scoring-messages
branch
September 3, 2026 20:51
2 tasks
thiagohora
added a commit
that referenced
this pull request
Sep 4, 2026
…time guard ships Folded in from #8145 (closed in favour of carrying it here), addressing review feedback on #8089 that landed after that PR merged. The javadoc justified absorbing an OutOfMemoryError by pointing at a publish-time size guard as the control covering the residual risk, naming onlineScoring.dropOversizedPayloads. That key does not exist: a repo-wide grep returns exactly one hit, the sentence itself. It never shipped. #8060 was merged as the Guice/Dropwizard ordering fix that made the codec actually receive maxStringLength; the publisher-side drop guard it originally carried was cut from that PR before merge, under this or any other name. That matters more than a stale reference normally would, because of the job the sentence was doing. The paragraph above it makes the most serious admission in the design -- under real heap pressure this arm absorbs an OOM that was a symptom rather than a cause and keeps consuming, masking it -- and this sentence was its discharge. With no such guard the risk is open, not delegated, and someone auditing the decision later would go looking for a control that was never built. Wording as suggested in review. Comment-only: the OOM-absorbing call itself is unchanged and not in question. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
thiagohora
added a commit
that referenced
this pull request
Sep 4, 2026
…time guard ships Folded in from #8145 (closed in favour of carrying it here), addressing review feedback on #8089 that landed after that PR merged. The javadoc justified absorbing an OutOfMemoryError by pointing at a publish-time size guard as the control covering the residual risk, naming onlineScoring.dropOversizedPayloads. That key does not exist: a repo-wide grep returns exactly one hit, the sentence itself. It never shipped. #8060 was merged as the Guice/Dropwizard ordering fix that made the codec actually receive maxStringLength; the publisher-side drop guard it originally carried was cut from that PR before merge, under this or any other name. That matters more than a stale reference normally would, because of the job the sentence was doing. The paragraph above it makes the most serious admission in the design -- under real heap pressure this arm absorbs an OOM that was a symptom rather than a cause and keeps consuming, masking it -- and this sentence was its discharge. With no such guard the risk is open, not delegated, and someone auditing the decision later would go looking for a control that was never built. Wording as suggested in review. Comment-only: the OOM-absorbing call itself is unchanged and not in question. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
thiagohora
added a commit
that referenced
this pull request
Sep 4, 2026
Per review: andrescrz's feedback was mostly about the retry logic, so the two fixes are being separated rather than reviewed together. Removed from this PR, moved to a follow-up: - ChatCompletionService's status-code retryability split and its tests. Removed entirely, filed as OPIK-8262: - preferRetryable / the sibling-failure aggregation change. The real fix is refactoring the trace-thread scorers to emit per-message ProcessingResults so the base subscriber's existing per-message ack/remove granularity is used, rather than picking a less-bad victim from a collapsed batch. Restored errors.getFirst(), which is what main already did -- the defect pre-dates this PR. Also dropped the RedisStreamCodec javadoc correction that had been folded in from #8145: it belongs to #8089, which is already merged, and does not need to ride along here. What remains is the XAUTOCLAIM cursor fix alone. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
thiagohora
added a commit
that referenced
this pull request
Sep 4, 2026
* [OPIK-8240] [BE] fix: split provider-error retryability by status, resume XAUTOCLAIM from its cursor Two independent defects that each leave a permanently failing online-scoring message cycling instead of retiring it. 1. A permanent provider 4xx was retried as if transient. ChatCompletionService.scoreTrace answered every unmappable provider failure with a blanket InternalServerErrorException, which sits outside BaseRedisSubscriber.NON_RETRYABLE_EXCEPTIONS -- so a request that can never succeed was replayed maxRetries times, once per pendingMessageDuration. The real status was available the whole time on the HttpException in the cause chain; it went unread because the existing recovery path classifies by family, and mapping all of 4xx to ClientErrorException would drop transient 408/429 after one attempt. Classify by status code instead: 400/401/403 and the rest of 4xx are permanent, 408/425/429 and all 5xx stay retryable. scoreTrace gets its own mapping so create() and the streaming handler keep returning the provider's status verbatim to HTTP callers. This also closes a pre-existing over-eager drop on the parseable branch, where a JSON-parseable 429 already became ClientErrorException and was dropped after one attempt. 2. XAUTOCLAIM never scanned past the first ~100 pending entries. claimPendingMessages passed StreamMessageId.MIN as the scan start every call and discarded the cursor Redis returns. Redis caps each XAUTOCLAIM at COUNT * 10 PEL entries *examined*, so at consumerBatchSize=10 a call inspects only the first 100 -- and restarting at MIN means nothing past that window is ever examined. Any backlog above ~100 grows a permanently unreachable tail. Carry getNextId() forward, resetting to MIN on Redis's 0-0 end-of-pass reply. A failed scan deliberately does not advance the cursor, so its window is retried rather than skipped. The sentinel is compared numerically: StreamMessageId.MIN/.ALL serialize to "-" and "0" and neither is equals() to the StreamMessageId(0, 0) Redisson parses 0-0 into (verified against redisson 4.7.0), so matching on the constants would never fire. Addressed from review: - Status precedence was backwards. The provider mappers synthesize a status when they cannot read one off the body (CustomLlmErrorMessage defaults to 400, OpenAiErrorMessage to 500), and preferring the mapped code let a synthetic 400 mask a real upstream 503 -- classifying a transient failure as permanent and dropping it on first delivery. The cause-chain HttpException now wins, with the mapped code as fallback, matching the precedence findProviderHttpStatus already documents for its own chain walk. - Sibling aggregation could discard retryable work. The trace-thread scorers re-emit one error for a message that fans out over many thread ids, and errors.getFirst() made that an arrival-order race. Harmless while every failure was a blanket 500; not once permanent and transient were split, as a ClientErrorException arriving first would ack and remove the entry with its retryable siblings. Both scorers now share representativeError(), which prefers a retryable sibling: a bounded replay is recoverable, silently dropped work is not. - Test-only: unconditional assertion flow in the parameterized status test, real-mapper-path coverage for the precedence rule, display-name wording. Tests: 145 green across the affected suites. Mutation-checked -- collapsing the status split to the whole 4xx family fails 5, removing it fails 7, reverting the cursor to always-MIN fails 3, inverting the status precedence fails 4, reverting aggregation to getFirst() fails 1. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * docs: correct the OOM-arm discharge in RedisStreamCodec — no publish-time guard ships Folded in from #8145 (closed in favour of carrying it here), addressing review feedback on #8089 that landed after that PR merged. The javadoc justified absorbing an OutOfMemoryError by pointing at a publish-time size guard as the control covering the residual risk, naming onlineScoring.dropOversizedPayloads. That key does not exist: a repo-wide grep returns exactly one hit, the sentence itself. It never shipped. #8060 was merged as the Guice/Dropwizard ordering fix that made the codec actually receive maxStringLength; the publisher-side drop guard it originally carried was cut from that PR before merge, under this or any other name. That matters more than a stale reference normally would, because of the job the sentence was doing. The paragraph above it makes the most serious admission in the design -- under real heap pressure this arm absorbs an OOM that was a symptom rather than a cause and keeps consuming, masking it -- and this sentence was its discharge. With no such guard the risk is open, not delegated, and someone auditing the decision later would go looking for a control that was never built. Wording as suggested in review. Comment-only: the OOM-absorbing call itself is unchanged and not in question. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix: fold sibling failures pairwise instead of collecting them Second review round on OPIK-8240. Fan-out is not chunked at either call site -- a manual evaluation passes every resolved thread id, and the streaming path every thread that closed in the window -- so collectList() could hold every sibling Throwable and its cause chain in memory at once during a provider outage, only to discard all but one. Reduced pairwise instead, keeping a single accumulator, in both trace-thread scorers. It also drops the isEmpty() check and the second pass: an empty Flux reduces to an empty Mono, which is already the no-failures case. Selection semantics are unchanged -- first retryable wins, else first failure. The trade is that the count of failed siblings is no longer recoverable for reporting; nothing reports it today. Tests: the two order-variant cases are consolidated into one @ParameterizedTest driven through Flux.reduce, so they exercise the accumulator the way the scorers actually use it rather than by direct call, and cover the empty sequence. Added a case pinning order-stability across three siblings, which the previous pair could not distinguish -- mutation-checked: always keeping the incumbent fails 2, always preferring a retryable candidate fails 1. Also: cursor test asserts isEqualTo rather than isSameAs. The contract is that the position is carried forward, not the identity of the object carrying it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * test: make the no-failures case a plain @test Parameterizing a single fixed case bought nothing and left an unused testName parameter -- an artefact of mechanically applying the same shape as the consolidated ordering cases, which do have something to vary. Inlined as Flux.empty(), which also says what the case is more directly than an empty list threaded through a MethodSource. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * test: cover the claim cursor on real redis, drop method-only test classes Review feedback on #8137, from andrescrz. The claim cursor was pinned by mocked XAUTOCLAIM replies plus a class-per-method test of the mapping function. Neither proves the behaviour: the reply those tests assert on is the one the test itself wrote. Redis's own PEL scan is what produces it in production, and its COUNT * 10 examine budget is the whole reason the cursor exists. BaseRedisSubscriberTest now drives a pending list deeper than one scan window against the real container. Restoring StreamMessageId.MIN makes it time out on the tail, which is the regression it is there to catch. The two mocked unit tests it subsumes are gone; the failed-scan one stays, because a container cannot be made to fail one XAUTOCLAIM and succeed on the next. The scoreTrace cases carried a helper that branched on isPermanentFailure to pick its assertions -- a test deriving its expectation from the classifier under test cannot fail when the classifier is wrong. Split into assertNonRetryable / assertRetryable, with each parameterised case running one unconditional flow over rows partitioned by a literal status set. That keeps the no-branching shape an earlier review asked for. Also restores the status assertion the transient case had lost, adds the permanent half of the parseable branch, turns the isPermanentFailure table into input-vs-expected rows, and randomises the request and workspace the helper builds. nextCursor goes private rather than gaining @VisibleForTesting: with its only direct test removed, there is no test for the annotation to document. isPermanentFailure, which is still called directly, gets the annotation in place of its comment. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * narrow to the claim-cursor fix; move retry classification to its own PR Per review: andrescrz's feedback was mostly about the retry logic, so the two fixes are being separated rather than reviewed together. Removed from this PR, moved to a follow-up: - ChatCompletionService's status-code retryability split and its tests. Removed entirely, filed as OPIK-8262: - preferRetryable / the sibling-failure aggregation change. The real fix is refactoring the trace-thread scorers to emit per-message ProcessingResults so the base subscriber's existing per-message ack/remove granularity is used, rather than picking a less-bad victim from a collapsed batch. Restored errors.getFirst(), which is what main already did -- the defect pre-dates this PR. Also dropped the RedisStreamCodec javadoc correction that had been folded in from #8145: it belongs to #8089, which is already merged, and does not need to ride along here. What remains is the XAUTOCLAIM cursor fix alone. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix: reset the claim cursor whenever the consumer group is recreated recoverFromNoGroup is reached from both readMessages and claimPendingMessages, but the cursor reset sat at the claim call site only. The read path is the likelier of the two to notice NOGROUP first, since reads run on every tick that is not a claim tick -- so the common case left the cursor pointing into the pending list of a group that no longer exists. Not self-correcting on a busy stream. A scan starting above the recreated PEL's entries only wraps once it exhausts the list, and entries arriving after the stale position keep giving it work at the high end, so the wrap can be deferred indefinitely while the oldest entries go unexamined -- the exact starvation this PR exists to remove. Reset moved into recoverFromNoGroup so both paths get identical treatment and there is one place that owns it. Regression test covers the read path specifically. Mutation-checked: moving the reset back to the claim site alone fails it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Details
A payload the Redis stream codec cannot decode used to throw inside Redisson's
CommandDecoder— belowBaseRedisSubscriber, before anyStreamMessageIdexists — so the entry could never be acked, retried or removed and the stream wedged permanently. This moves where that failure surfaces, so the entry arrives with its id and can be counted, logged, and retired throughmaxRetries.RedisStreamCodec.JAVAwraps the wholeCompositeCodec, so all three decode paths are covered: the payload (map value), the field name (map key — which resolves toLZ4CodecV2over Kryo5, and was the remaining hole), and plain values. An undecodable input yieldsUndecodableStreamMessageinstead of throwing.ExceptionplusOutOfMemoryError. Both arms are load-bearing and were measured: an over-limit string is aStreamConstraintsException, but a payload undermaxStringLengthand over the heap OOMs inside Jackson's own String materialization — andmaxStringLengthships at 100 MB withconsumerBatchSize: 10. No otherErroris absorbed.BaseRedisSubscriber.processMessagerecognises the sentinel, increments a new*_undecodable_messages_totaltagged witherror_typeandstream, logs the messageId / stream /encodedBytesat warn, and fails it as a retryableUndecodablePayloadException.maxRetriesbounds it, so it still terminates. A genuinely absent payload field is deterministic and stays non-retryable, removed on first delivery.postProcessFailureMessageshas always acked and removed failures. The failure simply never reached it.javacodec, so a decode failure wedges any of them today.Is the original problem solved?
Yes, and it is now verified end to end rather than argued.
BaseRedisSubscriberTest.UndecodablePayloadTestswrites an entry that succeeds on write and fails on read through real Redis, then asserts the healthy entry behind it still processes and the bad one leaves the stream. Mutation-checked: disabling the wrapper reproduces the production signature exactly —— and the test then fails at the first assertion, because the healthy entry behind it never processes either. That is the batch-stranding half of the wedge.
Three residuals, none of them the original bug:
XAUTOCLAIMredelivers afterpendingMessageDuration. At shipped values (10m × 3 retries) an undecodable entry occupies the stream for roughly 30 minutes, re-read each cycle, before removal. "Permanent" becomes "~30 minutes" — and at production payload sizes each cycle re-attempts a large materialization.getDeliveryCountmaps a failedlistPendingto0, so while that lookup fails persistently the count never reachesmaxRetries. Pre-existing for every retryable failure, not specific to this one, but it means "always terminates" is really "terminates while the PEL is readable".postProcessFailureMessagesdoes no subscriber-specific terminal bookkeeping, so a discardedDatasetExportMessageleaves its jobPROCESSINGand anExperimentItemToProcesscounter overstated. Pre-existing — there is a standing// TODO: Send to the dead letter queueathandleMaxRetriesReached— but this PR makes it reachable where before the stream simply wedged. Wants its own ticket, alongside version-scoped retirement; three separate review findings point at it.Behaviour change beyond undecodable payloads
Flagging this because the framing above does not imply it, and a reviewer skimming would not expect it:
processEventno longer receivesnull, for any of the seven subscribers.On
main, a stream entry with nothing under the configured payload field reachedprocessEvent(null)—messageContext(null)ran and the subscriber got the null. It is now short-circuited inprocessMessageand retired as a non-retryable failure, removed on first delivery.That is a base-class contract change, not just a test update, so it is worth being explicit that it fixes rather than removes behaviour:
OnlineScoringBaseScorer.processEventdereferencesmessage.workspaceName()on its first line, so a null was already a liveNullPointerExceptionfor the online-scoring subscribers onmain. The oldBaseRedisSubscriberTestonly passed becauseTestRedisSubscribertolerated a null where the real subscribers do not.processEvent's javadoc now states the guarantee and the parameter carries@NonNull.A field name that fails to decode lands in the same "no payload" symptom but is treated as retryable, since LZ4/Kryo skew can differ between pods — see the two branches in
processMessage.Change checklist
Behaviour change: a permanent stream wedge becomes one dropped scoring job, and streams already wedged self-clear on deploy.
maxRetriesalso becomes reachable on this path for the first time — the failure used to happen below the layer that counts deliveries. No new configuration keys.Issues
AI-WATERMARK
AI-WATERMARK: yes
Testing
New coverage:
oversizedPayloadYieldsSentinelwellFormedPayloadStillDecodesmalformedJsonYieldsSentinelencodersAreUntouchedshippedJavaCodecIsFaultTolerantshouldDropUndecodableMessageAndKeepConsumingprocessEvent, and healthy messages behind it still processMutation-checked, both directions:
JAVAcodec →shippedJavaCodecIsFaultTolerantfails. The guard against the wedge returning is real.UndecodableStreamMessagebranch inprocessMessage→ nothing fails. The sentinel still reachesprocessEvent, fails the generic cast, and is dropped as a non-retryableClassCastException. So that branch is not what makes the drop work; it adds the counter and a log naming the stream and payload size, which is the difference between "we discarded a 20 MB trace" and an opaque cast error. Asserting the counter needs OTel metric test infrastructure this module does not have. This is written into the test's javadoc rather than left to be rediscovered.Not run, with reason:
mvn verify— relying on CI.Documentation
No documentation change. No new configuration keys; the new metric is described in its own OTel description string.
Reviewer notes
CompositeCodec's two-arg constructor is(mapKeyCodec, mapValueCodec), andRStreamdecodes the field with the key codec and the value with the value codec. If that reading is right,CompositeCodec(new LZ4CodecV2(), ...)compresses the literal field name"payload"and stores the multi-megabyte trace JSON uncompressed — which would fit OPIK-8164's ~56 MB average entry and 19.66 GiB of growth. This PR is unaffected either way (encoders untouched), and I have not confirmed it against observed bytes. Worth a round-trip test before anyone acts on it; it would be a wire-format change needing a two-phase rollout.onlineScoring.dropOversizedPayloadsfrom [OPIK-8164] [BE] fix: apply Jackson stream-read limits to the Redis stream codec #8060 is largely redundant — the consumer can now survive anything that was written. Worth removing deliberately rather than leaving two mechanisms.Map<String, M>by erasure. That is how the existing OPIK-5647Collections.emptyList()handling works too, so it is the established shape here, but it is worth a reviewer's eye.🤖 Generated with Claude Code