feat(offsets) astubbs#255: an opaque rider in the commit-metadata payload, under one envelope magic byte - #460
Draft
astubbs wants to merge 14 commits into
Draft
feat(offsets) astubbs#255: an opaque rider in the commit-metadata payload, under one envelope magic byte#460astubbs wants to merge 14 commits into
astubbs wants to merge 14 commits into
Conversation
Kafka Streams on PC cannot restore its stream time after a restart against a group PC committed, because PC owns the commit metadata field and Streams' own TopicPartitionMetadata is never written there (KTD-S7 of the spike plan). The settled direction is an opaque rider: one generalised extension slot in PC's payload that an embedder fills per partition at commit and reads back on assignment, which PC carries without interpreting. This plan covers the core half only - the envelope encoding, the budget ladder, the supplier hook, the read-back entry point, metrics and the record - as a PR off master; the Streams consumer is a later rung on the Kafka Streams stack. It records the decisions research settled: the rider is charged against the metadata cap and never against back-pressure (charged against the threshold a rider is a floor back-pressure cannot relieve, and on a caught-up partition a permanent block); the ladder sheds the rider, then the drop marker, before the hole map; every released PC crash-loops on an unknown magic byte, so configuring a rider is opt-in with a stated minimum reader version and a recovery procedure; and magic byte 'X' lands ahead of the density work in #306, with the three EncodedOffsetPair switches named as the collision surface. Two things stay open on purpose: whether the rider's write side ships in 0.6.0.0 or the minor after (the graceful-degradation fix it relies on is unreleased), and a tracking issue for the idle-but-punctuating commit gap that blocks the customer's end-to-end result. CONCEPTS.md gains the Rider entry the plan's vocabulary needs. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KkRtYPnDuAPLEmHFYSnAiJ
…he commit metadata payload An embedder that runs on Parallel Consumer needs somewhere to persist a few bytes per partition across a restart, and the only slot a consumer-group commit offers is the metadata string PC already owns for its frontier encoding. This adds the envelope that will carry such a rider: magic byte 'X', an unsigned 16-bit length, the rider bytes, then today's hole encoding unchanged - or nothing, for a caught-up partition. PC never reads the rider; it knows only the length. The type is a standalone codec with no enum constant and no engine wiring, so the format can be reviewed on its own before anything consumes it. Three rules are enforced at the boundary rather than documented: the length is validated against the bytes remaining before anything is allocated, an envelope never nests (an inner first byte of 'X' is corrupt on both read and write), and a zero-length rider is not a value - it is the marker the budget ladder writes when it sheds a rider for size, so a reader can tell "dropped" from "never configured". Every array crosses the boundary as a copy. Two things the plan did not anticipate: the read side declares the checked CorruptOffsetMetadataException, matching EncodedOffsetPair's own decode signature, so the caller that wires it in must route it through the unreadable-metadata policy; and a getter on a rider that is not present throws rather than returning an empty array, because an embedder decoding zero bytes as a real value is exactly the hazard the dropped marker exists to prevent. Test-first: the test compiled red against the missing type, then went green with every layout claim asserted byte for byte, including the boundary lengths, the signed-negative length reading, truncation inside the header, and copy defence in both directions. Plan: docs/plans/2026-09-05-001-feat-offset-metadata-rider-plan.md, U1. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KkRtYPnDuAPLEmHFYSnAiJ
…ld payloads are byte-identical Wires the rider envelope into PC's offset-metadata codec so a payload can carry an embedder's blob beside the frontier's hole encoding. A new OffsetEncoding constant, RiderEnvelope, takes magic byte 'X' at the tail of the list, after the Kafka Streams pair, so the density work in #306 - which inserts its own constants after RunLengthV2Compressed - conflicts on a known line rather than a surprising one. The decode side unwraps the envelope before an EncodedOffsetPair is built and re-enters the same method on the remainder, so the inner magic byte is resolved through maybeDecode and reaches the unreadable-metadata policy exactly as an unknown outer byte does. A rider-only payload answers the same highest-seen offset as an empty payload: one below the committed offset, which is what keeps the committed record from being marked as already done. A malformed envelope - a length past the bytes remaining, a nested envelope, a truncated inner body - is routed through the policy under both IGNORE and FAIL, and the buffer-slicing exceptions are caught as a backstop behind the length validation, so nothing escapes the rebalance callback as a bare runtime exception. All three switches over the encoding gain an arm, including decodeBody's, whose default was the escape hatch. For the caller, one value carries what the codec decoded: DecodedMetadata, the existing HighestOffsetAndIncompletes plus a Rider state. Every public decode overload keeps its signature and return type and projects the rider away; the rider travels on a package-private family underneath. A payload the policy discards reports the rider as unreadable, so "never configured" and "thrown away" never read the same. On the write side the codec gains two entry points the budget ladder in PartitionState will use: one runs the encoder competition exactly once and returns the inner bytes (or the empty body for a caught-up partition, keyed on the same condition tryToEncodeOffsets returns early on), the other assembles inner bytes plus a rider into the string. With no rider the string is byte-identical to today's, asserted per codec. Test-first: the new codec test and the malformed-envelope arms compiled red against the missing shapes, then went green with the wire-format set untouched apart from one enum-source exclusion. Plan: docs/plans/2026-09-05-001-feat-offset-metadata-rider-plan.md, U8. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KkRtYPnDuAPLEmHFYSnAiJ
…e commit metadata, guarded like the retry-delay provider An embedder can now give Parallel Consumer a rider per partition: ParallelConsumerOptions gains riderSupplier, a function from a RiderContext - the partition, the offset the commit is paired with, and the most bytes the supplier may return - to the bytes to carry. The context's offset is the one rule the confluentinc#893 family settled: a payload travels with the offset it describes, so the supplier is told which one that is rather than left to guess. The supplier is user code on the engine's commit thread - the broker poller under the consumer commit modes, the control thread under the transactional producer, and the shutdown commit - so it is guarded the way the retry-delay provider and the meter registry already are: a throw or a null is caught, logged once through a rate limiter, and treated as no rider for that call; a rider longer than its cap is dropped with its own rate-limited warning. Failing the commit instead would turn a persistent supplier fault into a partition that never commits, which is worse than a missing rider. The supplier is called exactly once per encode, after the hole encoding is produced and inside the same snapshot, so a second read of partition state never enters the payload. The cap is derived, not literal: the headroom between the back-pressure threshold and the metadata cap, converted from Base64 characters to raw bytes and bounded by the envelope's own length field - so a rider at its cap can only fail the cap check once the hole map has already crossed the threshold. At the defaults that is 768 bytes. A caught-up partition writes a rider-only envelope and stays unblocked; one whose rider is over the cap writes no metadata at all. Configuring a rider is opt-in, and the option's javadoc and a one-time INFO line at startup say why the whole group must first run a PC that carries the unreadable-metadata policy: every released PC fails assignment on an unknown magic byte, the payload is durable, and the recovery is a consumer-group offset reset. Test-first: the guard test compiled red, then failed on the missing call site, then went green - including the poll-thread arm over a real processor and MockConsumer, and the transactional arm proving the committer calls the supplier on its own caller's thread with the transaction lock held. One thing the tests could not assert is byte identity against a separately built no-rider payload: the encoder competition breaks ties between equal-sized encodings by set iteration order, so the same state encodes as either of two magic bytes across calls. The tests assert the deterministic half and record the measurement. Plan: docs/plans/2026-09-05-001-feat-offset-metadata-rider-plan.md, U3. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KkRtYPnDuAPLEmHFYSnAiJ
… against back-pressure, and the ladder never costs the hole map Back-pressure exists so a partition's metadata can shrink as work completes, and rider bytes do not shrink. Charged against the 0.75 threshold a rider is a floor the mechanism cannot relieve, and on a caught-up partition a permanent block, because nothing is in flight to complete. So updateBlockFromEncodingResult now takes two lengths: the hole encoding's own character length is judged against the threshold, and the assembled string's against the cap. With no rider configured the first number is the assembled string's own length, byte for byte, which is why the existing back-pressure test's block point does not move - pinned over a corpus of hole densities by recomputing the old rule independently. When the assembled payload would exceed the cap, a ladder chooses what to shed, predicted from the Base64 closed form so the outer codec runs once per commit on the winning rung: the rider, then the three-byte drop marker, then the envelope altogether - because a hole map that sits within the marker's cost of the cap must still commit, and configuring a rider may never cost a partition metadata it would otherwise have written. Only after all three does today's strip fire. A rider the supplier's own cap already accepted provably fits alongside its offset map, so in practice the descent starts from the marker; the rung chooser is tested on its own boundary table regardless. The rung enum lives in its own public file only because the Truth subject generator recurses into every type in the state package and cannot see a package-private one. Also fixes a vacuous pair of assertions in the previous unit's guard test: comparing a decoded byte to the magic-byte constant boxed one side to Integer and the other to Byte, so the inequality could never fail. Both sides are ints now. Test-first: two scenarios went red on today's behaviour - a rider over the threshold blocked the partition, and a hole map within the marker's cost of the cap was stripped outright - then green with the whole core unit suite. Plan: docs/plans/2026-09-05-001-feat-offset-metadata-rider-plan.md, U2. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KkRtYPnDuAPLEmHFYSnAiJ
… citations repaired and its stale claims corrected The design record for the rider - one owner per metadata field, the other side riding inside it as an opaque blob - lived only on the Kafka Streams branches, invisible to every master-side search, while the code that implements it was landing here. It now lives beside that code. Every file:line citation became a path plus a greppable anchor, the io.confluent paths became bz.stub ones, and three claims were corrected in place rather than copied: the unreadable-metadata policy defaults to IGNORE since #207, not FAIL; PC's magic bytes include ByteArrayCompressed's 0xEE, which is not a printable letter, so the "always a letter" argument for Kafka Streams' graceful degradation is scoped to the bytes PC can actually emit; and the patched StreamTask method the record credited never existed under that name. A dated section records what shipped: the envelope, the budget rule, the supplier option and its guard, the read-back seam with its explicit policy, and the compatibility rule with its recovery procedure. docs/features gains the rider option as a planned entry - target release per the open question of whether the write side ships with the decoder or one minor later - carrying the minimum reader version, the recovery procedure and its cost, and the operator-facing change that metadata on a caught-up partition no longer means holes in flight. The policy's entry names how an envelope this build cannot read is treated. The pr-207 note records that this is the first encoding the policy was shipped ahead of. Two refactoring lines record what the work found: the encoder competition breaks ties between equal-sized encodings by set iteration order, and serialiseIncompleteOffsetMapToBase64 now has no main-code caller. A third records that the file-ref gate's history-pointer grammar stops at the first slash, so every git-show citation of a feats/ branch here needs a marker instead of the escape the citation guide prescribes. Plan: docs/plans/2026-09-05-001-feat-offset-metadata-rider-plan.md, U7. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KkRtYPnDuAPLEmHFYSnAiJ
…etadata string, and an old reader degrades the way the policy designed An embedder that does not own PC's consumer - Kafka Streams on PC runs the engine over a stub consumer and reads committed metadata through its own - needs a way to get its rider back from the string it holds. decodeRider takes the committed offset, the string and the unreadable-metadata policy, delegates to the codec's single decode choke point so the outer string codec is inherited, and returns the rider's state: none for a payload with no envelope, dropped for the marker the budget ladder writes, present with a copy of the bytes, unreadable when the policy discarded the payload. Under FAIL the policy's typed exception propagates. The policy is a required parameter and a test forbids ever adding an overload without one: the runtime default is IGNORE while the policy-less base64 helper deliberately picks FAIL, so any default here would decide on the caller's behalf between two answers that disagree. The compatibility claim is now proven against a broker rather than argued from code. Two records stuck failing keep holes open while everything around them completes, so the periodic committer writes holes and a rider in one payload - the shape a crash leaves - and the test captures it from the broker. A plain consumer then rewrites the payload's leading byte to one no encoding claims, which is how a reader without the envelope constant is simulated on one classpath, and the two arms are each other's control: under IGNORE a new instance resumes from the committed offset and loses no record; under FAIL assignment dies with the typed exception. An abandoned instance was rejected as the crash shape on a mechanism, not taste: it keeps heartbeating, so the coordinator refuses the rewrite from a non-member. Test-first: the read-back test compiled red against the missing method, and two mutations of the implementation - ignoring the caller's policy, returning none unconditionally - were killed before the tests were trusted. Plan: docs/plans/2026-09-05-001-feat-offset-metadata-rider-plan.md, U4. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KkRtYPnDuAPLEmHFYSnAiJ
…ing supplier, and keep the two payload ratios honest The rider is best-effort by contract, so the only way an operator can tell it is working is a metric. Four per-partition meters join the offset-encoder subsystem: the size of each rider written, a counter for a rider shed to fit the metadata limit (over its own cap, or squeezed out by the offset map), a counter for a payload stripped because the offset map itself did not fit, and a counter for a riderSupplier that threw - the one signal that a rider-based feature has silently stopped working, since a throw is treated as no rider and the commit goes ahead. A null return is deliberately not counted: the option's contract makes it a legitimate "no rider this commit", and counting it would make a healthy embedder look broken. All four are registered and removed with the partition through the guarded path the meter-registry incident established. The two existing ratios were answering the same question once a rider was in the string. The density ratio now records the hole encoding's own length, so it keeps measuring how well the offsets encode; the headroom ratio records the assembled string, rider included, because that is the number that says how close a commit is to the cap. A caught-up commit records neither: its offset range is negative or zero, and Micrometer keeps a zero sample rather than dropping it, so leaving the guard to the library would drag both ratios toward zero on every steady-state commit. The one double-count in the design was guard-then-ladder: a rider over its own cap is counted when the supplier's guard turns it into the drop marker, and the ladder can then shed that marker for size. The shed counts only a present rider, so one lost rider is one count - pinned by an assertion a deliberate mutation of that branch turned red. The metrics table in the README is regenerated from the definitions, never hand-edited; the inflight note on discarded metadata records which half this closes and that a payload discarded on read is still only a warning. Plan: docs/plans/2026-09-05-001-feat-offset-metadata-rider-plan.md, U5. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KkRtYPnDuAPLEmHFYSnAiJ
… whole domain, and pin that back-pressure does not move The forever-format rule here is measure first: a claim about where a payload crosses a size boundary is checked exhaustively through the real pipeline, not sampled on friendly inputs. This test does that for the rider. For every rider length the supplier can be offered, across a spread of inner-encoding lengths, the assembled string's length equals the Base64 closed form the budget ladder predicts from - through the real assemble path, envelope included. Then, for each incumbent encoding over uniform, clustered and all-incomplete hole maps, it searches for the exact hole count at which back-pressure engages and at which the cap engages, with and without a rider, driving real partition-state commits at each boundary and its predecessor. A fixed ladder of ranges would have missed the band where the rider is what crosses the cap: it is a handful of bytes wide out of three thousand. The shape of the result: the back-pressure engagement point does not move at all - reverting the threshold to the pre-change rule turned the test red at the first encoding - and the cap engagement point moves earlier by exactly the rider's Base64 footprint, for every combination that reaches the cap. The free competition ships an encoding of the same size with the same compression decision either way; the test asserts size rather than magic byte because equal-sized ties resolve by set order. Deliberately not excluded from the mutation lane, against the plan: every assertion is an exact value, so the class kills mutants the lane would otherwise count as survivors, and it costs seconds. The design record gains the measured-cost note in place of KTD-S7's budget caveat, and the build guide records a hazard this test found: a nested type inside a test class makes the Truth subject generator emit an import that breaks the next build. Plan: docs/plans/2026-09-05-001-feat-offset-metadata-rider-plan.md, U6. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KkRtYPnDuAPLEmHFYSnAiJ
…and let the rider tests share one fixture Eight workers built this branch one unit at a time, and a simplification pass over the whole found what that leaves behind. The rider parameter threaded through EncodedOffsetPair's decode method was only ever handed "none": the behaviour it was reserved for - an intact envelope keeping its rider when the inner offset map is corrupt - had been solved a level up, where the envelope decoder substitutes the outer rider into the result. The parameter and the branch it fed are gone; the test that proves the behaviour is unchanged. The five new test classes had each restated the same fixture - the Base64 closed form, seeded riders, the remember-and-restore dance around the two mutable statics, a partition state with random holes, decode-and-unwrap of a committed payload. They now share one test-support class, which still restates the arithmetic independently of production so no test agrees with itself. The two hand-rolled logback captures use the LogCapture utility the module already has, and a local byte-array concat gives way to Guava's. The rider javadoc says "offset map", the term the surrounding code has always used, instead of "hole map". One reviewer suggestion was measured and reverted: sharing one module across the overhead test's search probes moved its wall-clock within noise. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KkRtYPnDuAPLEmHFYSnAiJ
…adoc names, and the over-cap warning says what was written Three findings from the pre-PR code review (run 20260907-013655-cce2ea5f), all validated by an independent pass, plus two mutants the PIT lane reported surviving. decodeRider promised CorruptOffsetMetadataException and UnknownOffsetMetadataMagicException under FAIL in its javadoc, but declared only OffsetDecodingError. Both are checked; they reach the caller sneaky-thrown through the same policy handler every other entry point in this family uses. A checked type that is thrown but not declared cannot be caught by name - javac rejects the catch as unreachable - so an embedder could not write the handling the javadoc described. The only test used assertThatThrownBy, whose callable is declared throws Throwable, which is why the compiler never saw the mismatch. The signature now declares both, and a new test is the caller the javadoc describes, written with real catch blocks so it compiles only while the declaration holds. The over-cap rider warning described one outcome where riderFromSupplier produces two. Beside an offset map the rider becomes the dropped marker and a reader sees DROPPED; on a caught-up partition there is no map for the marker to sit beside, no metadata is written at all, and a reader sees NONE. The warning promised the marker in both cases, sending an operator looking for a state the caught-up commit never writes. It now branches on which happened, and the caught-up guard test asserts the text. The feature record's boundary line made the mirror-image error - it said an over-cap rider is "treated as no rider", contradicting its own values section that reserves the marker - and now states both. PIT (offsets lane, PR-scoped against origin/master, 301 of 301 mutants scored): the envelope class itself scored 23/23; two survivors in the rider's own code are closed here - describeEnvelope's unknown-inner-encoding branch had nothing telling it apart from the map branch, and Rider.hashCode was pinned only as "equal riders hash equal", which any constant satisfies. The remaining offsets survivors are in code this branch did not write. Not changed on review advice, deliberately: the reviewer's proposal to split PartitionState at a 1000-line bar (the plan places the ladder there; RiderBudgetRung is split out only for the Truth-generator constraint), and the proposal to hide encodeOffsetsToInnerBytes and assembleMetadataPayload (public because PartitionState calls them cross-package, the same shape as the pre-existing makeOffsetMetadataPayload). Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KkRtYPnDuAPLEmHFYSnAiJ
Owner
Author
|
@claude review this |
4 tasks
|
Claude finished @astubbs's task in 2m 53s —— View job Review: opaque rider envelope (#255)
Note on scope: the review workflow allowlist is resolved from master's copy of OverallThis is a well-reasoned, carefully documented change. The envelope format (
Finding
/**
* One rate-limited warning for a supplier that returned more than the {@link RiderContext#getMaxRiderBytes()}
* it was handed (R8).
* ...
*/
/**
* The one warning for an over-cap rider has to describe two different outcomes, ...
*/
private void warnOversizedRider(int riderLength, int allowance, boolean besideAnOffsetMap) {Only the second block attaches to the method; the first is a dead/orphaned comment left over from an earlier draft (likely predating the Known Residuals sectionRead against the code, the residuals are accurately scoped — I didn't find any of them silently already fixed, nor any additional unflagged gap of the same shape. The "no test drives the rider through PC's own close-time commit and a restart read-back" gap and "no Not reviewed in depthGiven time budget, I traced the core codec/budget/options/module paths closely but did not do a line-by-line pass of the four large test files ( On the release-decision ask in the PR body: that's a product/release call for the author, not something I can weigh in on from a code-correctness review — flagging so it doesn't get lost. |
Dependency Review✅ No vulnerabilities or license issues or OpenSSF Scorecard issues found.Scanned FilesNone |
✅ Duplicate Code ReportTwo engines run in parallel for cross-validation. Each has its own thresholds tuned to its baseline - the real safety net is the per-engine "max increase vs base" check. ✅ PMD CPD
No new clones introduced by this PR. ✅ jscpd (language-agnostic)
|
🧪🔒 Quarantine Lane Report
🔴 expected while the owner PR is open · 🟡🎲 flapper, pass proves nothing · 🚨 a deterministic quarantined test passing means its fix landed: delete its No quarantined test changed outcome since the previous push. Updated for |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #460 +/- ##
============================================
+ Coverage 82.34% 83.34% +0.99%
- Complexity 1486 1574 +88
============================================
Files 95 97 +2
Lines 5189 5469 +280
Branches 508 545 +37
============================================
+ Hits 4273 4558 +285
+ Misses 718 713 -5
Partials 198 198
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
Owner
Author
|
@codex review |
🟢 Throughput — OKThis branch measured about 10% slower than master, on the one test this measures. That is INSIDE this test's own run-to-run spread of about 17%, so read it as a reading and not as a result - re-running the same commit moves it by about as much.
Allowable range 🟢 ≥ 0.70 · 🟡 0.50–0.70 (about a 30% loss) · 🔴 < 0.50 (about a 50% loss) What the numbers mean, and what they cannot tell youThe one that gets misread. Why a shape and not a rate. A rate depends on which runner you drew. A shape does not: every test here processes a fixed number of records, so a runner twice as slow doubles the subject and the controls together and leaves their ratio alone. That is the whole trick, and it is why the reported rate is shown last and labelled as this machine only. Reading the comparison. By conservation, not by correction. Every test in this lane processes a fixed number of records, so within one run the ratio of one test's time to another's is invariant under machine speed — a runner twice as slow doubles both terms and leaves the ratio alone. There is no machine-index correction to be wrong, because nothing needed correcting. Per-method times, not class times. A class time is Reference is the median of 10 recent What this still cannot do. It removes machine-to-machine variance. It does not remove this test's own run-to-run variance, measured at about 30% on a single unchanged commit while its controls stayed within 5%. That is a property of the test, not of the comparison, and no arithmetic here can touch it — which is why the reference is a median and the bounds are deliberately coarse. 🟡 means look at this; only 🔴 is outside the measured spread. Runs used: e8f7beb, 29f6a0f, a75400f, 12bf414, ce6f39a, 70a88bf, 27211b9, c5dde06, 92364a1, 867c407 Since the previous push: ratio 0.989 -> 0.896, share 1.635 -> 1.803, rate 66785 -> 66421 (-0.5%). One push of difference sits inside this test's measured spread - read it as movement, not as a result. Updated for |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 6e85d5768a
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
Comment on lines
+644
to
+646
| // KTD14: the stripped-payload counter only. This escapes the inner-bytes step, which under KTD9 runs | ||
| // BEFORE the supplier is called, so on this path there is no rider to have discarded and nothing to | ||
| // count as dropped - the ladder's own strip rung is where both are counted. |
There was a problem hiding this comment.
When no offset-map encoder can represent the current range, this branch still returns a bare OffsetAndMetadata that is sent as a commit, but it runs before riderFromSupplier and therefore silently omits the configured rider. This contradicts the new option's promise that the supplier is called once per commit per partition and can leave the durable rider stale or absent after exactly the large-range fallback where the hole map is stripped; invoke the supplier and carry a rider-only envelope, or explicitly narrow the public contract.
Useful? React with 👍 / 👎.
Owner
Author
There was a problem hiding this comment.
invoke the supplier and carry a rider-only envelope, or explicitly narrow the public contract.
Fair point, and it is wider than this branch: the ladder's own strip rung has the same shape (over the cap, the rider is shed first and the map stripped after, so nothing rides). Carrying a rider-only envelope on both strip paths would be consistent with the caught-up commit and is the direction I lean, but it changes the ladder's documented last rung and the pc.offsets.rider.dropped accounting the budget tests pin, so I want to make that call deliberately rather than inside a review round. Leaving this open for that decision.
…eclares the whole checked family, and the rider tests speak Truth Codex's first pass on #460, Claude's summary, and the two CI failures that were this branch's own. THE ONE THAT MATTERED. encodeOffsetsToInnerBytes re-read hasIncompleteOffsets() after tryToEncodeOffsets had already decided the caught-up case on the read it commits against. The control thread can complete the last incomplete record between those two reads, and the re-check then answered "nothing to encode" for a commit already pinned to the older offset - no map, and a crash before the next commit replayed records this build had recorded as complete. That is the confluentinc#893/894 tear class the javadoc two lines away warns about, reintroduced one call deeper. The method now encodes whatever the encoder's own single sample holds, as the pre-rider path always did; a map that emptied in between encodes as a complete map, which resumes correctly. The test that pinned the old answer now pins the new one and says why. decodeRider declared OffsetDecodingError and CorruptOffsetMetadataException but not the rest of the checked FAIL family - KafkaStreamsEncodingNotSupported and UnsupportedOffsetEncodingException arrive sneaky-thrown just as the unknown-magic case does. It now declares their parent, EncodingNotSupportedException, so a caller can catch the family in one clause or any member by its own type, and the javadoc names all three. The overhead test read "not engaged at the top of the domain" as "never engages", which would have skipped every boundary assertion for exactly the non-monotone compressed shape it exists to measure. It now walks a 64-point grid before believing the endpoint; the residual is a transition narrower than one grid step, named in the javadoc, rather than an unobserved interior. House rules: the three new rider suites that had followed their AssertJ neighbours now use Google Truth; discarded Set/Map returns in the rider tests are named with their reason; the release decision moved out of the merged #207 note into release-offset-rider-write-side-split.md; the encoding tie-break sighting outgrew a refactoring.md line and is now core-offset-encoding-tie-break-is-not-deterministic.md; two stacked javadocs before warnOversizedRider are one; the setup the two codec tests duplicated (CPD's one new clone) is RiderTestFixtures.moduleWithNoSupplier and stateOver. CI: CloseAndOpenOffsetTest forced every OffsetEncoding as a codec, including the envelope constant that names none; it is skipped like the other non-encoders, and the reset of the forced statics moved to @AfterEach, because the failing arm had leaked forcedCodec into the next test in the class. Infer's BUFFER_OVERRUN_L1 on OffsetRiderEnvelope.wrap and its two callers is a guarded inner[0] read that the interval domain cannot see past; the three identities are ratcheted with that reason. Left open on the PR, deliberately: whether the rider should ride alone on the two strip paths (the no-encoding fallback Codex flagged, and the ladder's own strip rung), which changes the ladder's documented last rung and the dropped-rider accounting. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KkRtYPnDuAPLEmHFYSnAiJ
Owner
Author
Addressed in 6725ad9: the two blocks before |
Master moved 21 commits under this branch in two days. Three touch the rider's own files and none argues with it: #450 registers a record's offset before publishing it to its shard (PartitionState, adjacent to but not inside the commit path this branch changed); #452 makes the codec's encodingCounters a ConcurrentHashMap (the encode-path static this branch's tryToEncodeOffsets calls through); #106 walks only the offsets where completion state changes when no encoder needs every offset (the overhead table's domain is unchanged by it, and its tests pass on the merge). Two text conflicts, both in append-only records, both resolved in master's favour plus this branch's additions: config/infer-known-findings.txt keeps master's retirement of every THREAD_SAFETY_VIOLATION identity (#433) and this branch's three BUFFER_OVERRUN_L1 identities; docs/refactoring.md keeps master's rewritten thread-safety entry for the codec (#452 replaced the "not thread-safe" heading and body) and this branch's two entries above it. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KkRtYPnDuAPLEmHFYSnAiJ
…e single-sample encoder took with it The Infer lane fails both ways, and this is the good direction: one of the three BUFFER_OVERRUN_L1 identities recorded on the previous commit - OffsetMapCodecManager.makeOffsetMetadataPayload - no longer fires, because encodeOffsetsToInnerBytes stopped returning an empty array on its own authority (the emptiness re-read Codex caught), so nothing on that path hands wrap an empty inner any more. The two sightings that remain are the guarded read itself and the caught-up commit that wraps an empty array on purpose. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KkRtYPnDuAPLEmHFYSnAiJ
6 tasks
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
1 participant
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.
Part of #255 - the core half of KTD-S7. It does not close the issue: the Kafka Streams consumer of this slot lands as a rung on
feats/ks-streams-reconciledabove #454, and the idle-but-punctuating gap the issue also names needs its own tracking issue (see Known Residuals).Description
Parallel Consumer's commit metadata gains one generalised slot: an opaque rider an embedder can hand PC per partition, carried in the same
OffsetAndMetadatastring as the offset-hole map, under one new envelope magic byte'X'([u16 len][rider][inner encoding or nothing]). PC never reads the bytes; it needs the length and nothing else.What binds it, in the order a reviewer will want to check them:
OffsetRiderCodecTest.aPayloadWithNoRiderIsWhatPcWritesTodaypins this across every inner encoding.OffsetRiderOverheadTestmeasures the cost over the whole domain and pins that the back-pressure engagement point does not move.RiderBudgetRung,PartitionStateRiderBudgetTest).retryDelayProvider. A throw, a null, or an over-cap rider is counted, rate-limit logged, and the commit still happens (RiderSupplierGuardTest).OffsetMapCodecManager.decodeRider(committedOffset, metadata, policy)- through the codec's single decode choke point, with fix(offsets)! astubbs#197: the metadata policy governs every unreadable payload, and defaults to IGNORE #207's unreadable-metadata policy governing an envelope this build cannot read (OffsetRiderReadBackTest,ForeignOffsetMetadataOnAssignmentTest).OffsetRiderUpgradeDowngradeTestreproduces that against a broker and pins the recovery path; the option's javadoc anddocs/features/offset-metadata-rider.yamlcarry the rule, and PC logs it once at INFO when a supplier is configured.Design record:
docs/solutions/architecture-patterns/one-owner-per-metadata-field-with-an-opaque-rider.md(migrated to master with this PR, citations repaired). Plan:docs/plans/2026-09-05-001-feat-offset-metadata-rider-plan.md- its settled decisions (one envelope byte, not one per encoding; core slot ships alone off master) are the ones a simplify pass would otherwise reverse.Magic byte registry note for #306:
'X'is taken by this envelope, andEncodedOffsetPairnow has three switches that need an arm per encoding, not two.Release decision for the author: the read side (understanding
'X') and the write side (riderSupplier) ship together here, both targeted at 0.6.0.0 alongside #207's policy.docs/features/offset-metadata-rider.yamlrecords the alternative - read side in 0.6.0.0, write side one minor later so every 0.6.0.0 reader predates every writer - underrelease_assumptions. Say which you want before merge.Known Residuals
From the pre-PR review (run
20260907-013655-cce2ea5f, three validated findings applied in6e85d5768) and the first review round on this PR (Codex and Claude; seven of eight threads fixed in6725ad93a, master merged in703591778). None of these is an unapplied actionable finding; they are the design items routed to a human, plus what the mutation lane found:Design call (Codex thread, left open): on the no-encoding fallback - the map cannot be encoded at all, so the payload is stripped - the supplier is never called and no rider is written. The ladder's own strip rung has the same shape. Carrying a rider-only envelope on both strip paths would match the caught-up commit and is the direction I lean, but it changes the ladder's documented last rung and the
pc.offsets.rider.droppedaccounting the budget tests pin.The ladder's first rung is unreachable by construction (PIT sweep of
PartitionState, no coverage onshedRiderForSize): the write-time allowance inriderFromSupplieralready guarantees a rider that honours it fits theRIDERrung, sofitRiderToBudgetcan only descend to the marker when the statics move between the supplier call and the fit. Either the allowance or the rung is redundant; which one to keep is the same decision as the point above.dups: similarityis red on this PR and cannot be made green here: the engine scores the new rider test classes against each other and againstWorkClaimStateMachineTestat 80-90% by whole-file likeness, and it has no allowlist (docs/inflight/ci-dup-similarity-cannot-accept-known-duplication.md). The route the repo has used before is an override at merge.Design call (adversarial, advisory): a same-build group member with no supplier configured overwrites a partition's rider with "none" on its next commit. The docs say every member needs the policy; they do not yet say every member needs the supplier. Decide for the Streams rung whether PC should write the DROPPED marker when it read a rider it will not re-supply, or whether "every member configures the supplier" is the documented rule.
Design call (adversarial, advisory): a supplier that throws on the shutdown commit loses the rider a restart reads, silently (counted, rate-limit logged). Same question: NONE vs DROPPED on supplier failure beside an offset map.
Documented, not enforced (reliability, adversarial): the supplier runs on an engine thread - the broker-poll thread under consumer commit modes, the control thread holding the produce write lock under the transactional producer - with no timeout. A blocking supplier stalls producers; a slow one counts against the rebalance timeout. Same contract as
retryDelayProvider; the Streams rung must publish its rider from a lock-free snapshot.Test gaps worth a follow-up: no test drives the rider through PC's own close-time commit and a restart read-back (the integration test overwrites that commit); no
Error-subtype through the supplier guard; nodecodeRideragainst a Kafka-Streams-shaped inner payload by name.Tracking issue needed: a caught-up, non-dirty partition never refreshes its rider, because the supplier runs only on commits the partition would have made anyway (KTD8 / R12). Kafka Streams: give a Streams topology PC's per-key parallelism #255's idle-but-punctuating case is therefore still open after this PR.
Post-Deploy Monitoring & Validation
parallel-consumer-corePCMetricsDef):pc.offsets.rider.size(distribution - should track the embedder's payload),pc.offsets.rider.dropped(must stay at zero in steady state; a non-zero rate means the rider does not fit beside the hole map),pc.offsets.payload.stripped(should not rise after enabling a rider - if it does, the rider is being blamed for a cap the hole map already hit),pc.offsets.rider.supplier.failed(any increase is a bug in the embedder's supplier: it is the only continuous signal, because the warning is rate limited).announceRiderCompatibilityRequirement); WARNs mentioningriderSupplier(once per 30 s per PC instance: a throw, or an over-cap rider - the message says whether the marker was written or nothing was).riderSupplierWARNs;pc.offsets.rider.droppedflat;pc.offsets.payload.strippedunchanged from before the rider was configured; the existing back-pressure ratio metrics unmoved.UnknownOffsetMetadataMagicExceptionorCorruptOffsetMetadataExceptionfromonPartitionsAssigned, or crash-looping on assignment - a member predates this build. Do NOT roll the writer back: unsettingriderSupplierdoes not heal the group (a caught-up partition never recommits). Follow the recovery indocs/features/offset-metadata-rider.yamlboundaries(kafka-consumer-groups --reset-offsets --to-current).Without a supplier configured this PR changes no wire byte, so a deployment that does not set
riderSupplierneeds no additional operational monitoring.Checklist
docs/features/offset-metadata-rider.yaml,docs/features/invalid-offset-metadata-policy.yaml,CONCEPTS.md(Rider),README.adocmetrics (regenerated fromsrc/docs/README_TEMPLATE.adoc),docs/building.md(Truth-generator nested-type hazard),docs/refactoring.md, the migrated design recorddocs/features/-offset-metadata-rider.yamldocs/inflight/working note (pr-/branch-) started at the PR's first commit - none needed: the two touched notes (pr-207-offset-encoding-policy.md,bug-no-metric-for-discarded-offset-metadata.md) were narrowed to what stays open; the branch's cross-branch state is the plan and the settled KTDs in it, andghshows the restce-simplifyandce-code-reviewlocally - both; review run20260907-013655-cce2ea5f(status complete, verdict "ready with fixes"), fixes in6e85d5768🤖 Generated with Claude Code
https://claude.ai/code/session_01KkRtYPnDuAPLEmHFYSnAiJ