Describe the desired outcome from the user's perspective
As a Zilla operator, I want the kafka cache storage layer's produce path to encode each message with that message's own producer's authorization (#2310), so that a model transform whose behavior depends on that parameter sees the correct, per-message value rather than whichever producer happened to open the shared connection state first.
The gap, concretely
KafkaCacheClientProduceFactory encodes each produced message through a per-topic-partition object (a "fan") shared across every producer writing to that partition. The fan captures and stores the authorization of whichever producer first opens it, and every subsequent produce call — from any producer, including ones that open a stream on that same topic-partition later — encodes using that captured value, not its own. A later producer's own authorization is tracked on its own per-producer state, but the encode call reads the fan's field instead. The fan also persists for a cleanup delay after its last member closes, so a stale captured authorization can outlive the producer that supplied it.
This is inert today for a transform that never varies its behavior by authorization on encode. It is wrong for one that does — a real, if currently unexercised, violation of #2310's own contract for that parameter (a per-message value, not a captured global for the connection group).
Acceptance criteria
- Each encode call uses the authorization of the producer stream actually producing that message, not whatever the fan captured when it was created
- Encoder pipeline state lives per-producer-stream, not shared on the fan, so state carried across separate producers doesn't leak between them the same way authorization does
- A fan reused after its last member's cleanup delay must not carry forward a stale authorization from a previous producer
- Unit and k3po coverage: two producers on the same topic-partition, opened in each order, each encoding with their own authorization on a transform whose behavior depends on it; a fan reused after cleanup delay uses the new producer's own authorization, not the prior one's
Additional context
Related to #2423 (the analogous per-message-authorization gap on the fetch path) — found while examining that issue's implementation, but a distinct code path (KafkaCacheClientProduceFactory, not the fetch side) with a different failure mode: not "no real caller yet," but "the wrong real caller's value used." Scoped to the shared kafka cache storage implementation, the same layer #1688/#1689/#1690/#2423 already scope themselves to.
Describe the desired outcome from the user's perspective
As a Zilla operator, I want the kafka cache storage layer's produce path to encode each message with that message's own producer's
authorization(#2310), so that a model transform whose behavior depends on that parameter sees the correct, per-message value rather than whichever producer happened to open the shared connection state first.The gap, concretely
KafkaCacheClientProduceFactoryencodes each produced message through a per-topic-partition object (a "fan") shared across every producer writing to that partition. The fan captures and stores the authorization of whichever producer first opens it, and every subsequent produce call — from any producer, including ones that open a stream on that same topic-partition later — encodes using that captured value, not its own. A later producer's own authorization is tracked on its own per-producer state, but the encode call reads the fan's field instead. The fan also persists for a cleanup delay after its last member closes, so a stale captured authorization can outlive the producer that supplied it.This is inert today for a transform that never varies its behavior by authorization on encode. It is wrong for one that does — a real, if currently unexercised, violation of #2310's own contract for that parameter (a per-message value, not a captured global for the connection group).
Acceptance criteria
Additional context
Related to #2423 (the analogous per-message-authorization gap on the fetch path) — found while examining that issue's implementation, but a distinct code path (
KafkaCacheClientProduceFactory, not the fetch side) with a different failure mode: not "no real caller yet," but "the wrong real caller's value used." Scoped to the shared kafka cache storage implementation, the same layer #1688/#1689/#1690/#2423 already scope themselves to.