fix(binding-kafka): use each producer stream's own authorization for cache encode - #2425
Open
jfallows wants to merge 2 commits into
Open
fix(binding-kafka): use each producer stream's own authorization for cache encode#2425jfallows wants to merge 2 commits into
jfallows wants to merge 2 commits into
Conversation
…cache encode
KafkaCacheClientProduceFan captured the authorization of whichever
producer stream first created the per-topic-partition fan, then reused
that single value (and a shared transformKey/transformValue encoder
pair) for every subsequent producer writing to the same partition.
KafkaCacheClientProduceStream already tracked its own authorization but
it went unused by the encode call sites. Move transformKey/transformValue
onto the stream and thread stream.authorization through
writeProduceEntryStart/Continue so each message is encoded using the
authorization of the stream that actually produced it.
Extend the engine's generic TestModel with an opt-in
transform: { authorization: true } flag so tests can observe which
authorization value reached the model encoder via stamped output
bytes, and add a k3po scenario (two sequential producers, distinct
authorization) plus engine-driven and peer-to-peer IT coverage.
…eject checks
TestModel previously exposed a transform: { authorization: true } flag that
stamped the authorization value into output bytes so a test could recover it
after the fact. Replace it with transform.authorizations: [N, M] -- an
ordered list of expected authorization values -- and reject the value
outright when a mismatch is observed. The check is applied per completed
value, keyed off a shared counter on the handler rather than the position of
the pipeline instance that observed it, since a shared/hoisted pipeline
handling multiple messages must still be checked against each message's own
expected value.
Fix the corresponding k3po scenario's message ordering to match the actual,
now-verified arrival order, and hoist the new schema property to a top-level
sibling of transform/fields rather than nesting it under transform, which
a two-properties-deep array-typed nested property does not validate under
the current JSON schema patch mechanism.
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
2 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.
Description
KafkaCacheClientProduceFancaptured theauthorizationof whichever producer stream first created the per-topic-partition fan, then reused that single value — along with a sharedtransformKey/transformValueencoder pair — for every subsequent producer writing to the same partition.KafkaCacheClientProduceStreamalready tracked its ownauthorization, but the encode call sites (onClientInitialData×2,onClientInitialFlush) never used it.This moves
transformKey/transformValuefrom the fan onto the stream and threadsstream.authorizationthroughwriteProduceEntryStart/writeProduceEntryContinue, so each message is encoded using the authorization of the producer stream that actually produced it, not the first producer to touch the partition.To make this observable in an oss-only test, the engine's generic
TestModelgains an opt-intransformAuthorizations: [N, M]config list: an ordered list of the authorization values expected to reach the model, in message-arrival order. Each completed value consumes the next entry from this list (tracked on the sharedModelHandler, not the pipeline instance that observed it — a shared/hoisted pipeline handling multiple messages must still be checked against each message's own expected value) and is rejected outright on mismatch, surfacing as an ordinaryREJECTED→ streamRESET, the same signal already used by the existingmessage.value.rejected/message.values.rejectedscenarios. A new k3po scenario (message.values.authorization.distinct) drives two producers with distinct authorization values through a live engine and asserts each message reaches the model with its own value, not the other's.Verified test-first: the new
CacheProduceITtest fails with a clean, fast rejection (not a hang/timeout) against the pre-fix code, and passes against the fix. FullProduceITandCacheProduceITsuites,runtime/engineandconfig/engine.confunit tests, and checkstyle/license checks all pass with no regressions.The new scenario's peer-to-peer counterpart is intentionally omitted, with the reasoning documented inline in
ProduceIT.java: a pre-existing k3po-harness limitation (unrelated to this change, and already present for the never-coveredmessage.values.parallelscenario) prevents a third notify/await-gated connect from being peer-verified in this harness.Fixes # (issue)
Generated by Claude Code