kafka cache storage: honor per-message authorization on the fetch path, not just at cache population

[jfallows]

Describe the desired outcome from the user's perspective

As a Zilla operator, I want the kafka cache storage layer's fetch path to honor the per-message authorization parameter #2310 already added to ModelPipeline.transform(...), so that an installed model transform whose behavior depends on that parameter gets the contract #2310 established, rather than only ever seeing whatever value was in effect when the cache was populated.

The gap, concretely

KafkaCacheServerFetchFactory decodes a fetched record's value once, while populating the local cache, with authorization hardcoded to a fixed sentinel (NO_AUTHORIZATION) — there is no requesting consumer yet at that point, since this runs ahead of and independent of any consumer's own fetch. The result is stored in the partition's convertedFile and served to every subsequent reader verbatim by the cursor that reads it back — including the per-consumer fetch stream itself, which constructs no model pipeline of its own at all today.

This is correct and efficient for a transform whose result never depends on the sentinel value at all — the existing view: json conversion is exactly this kind of transform, and caching it once genuinely is the right behavior, not a bug to fix. It's a real gap for any transform whose behavior is defined in terms of the caller's own authorization: evaluating it once, against a fixed sentinel, and sharing that single answer with every future reader forever doesn't honor #2310's own contract for that parameter — the parameter exists specifically to let behavior vary per caller, and this path never gives it a real, caller-specific value at all.

Acceptance criteria

• Give the per-consumer fetch stream a real model-decode invocation of its own, carrying that consumer's live authorization, rather than only ever serving whatever the population-time decode already computed and cached
• The population-time decode and cache remain exactly as they are for whatever portion of a model's composed pipeline does not depend on the calling consumer's authorization — this issue is about adding a missing per-consumer invocation, not removing or replacing the existing shared one
• Open design question, not assumed by this issue: how does a model's composed pipeline (Declare AvroModelExtFactorySpi and wire provider-supplied pipeline stages in avro model #1681's fold) distinguish an installed extension stage that's safe to compute once and share from one that must be recomputed per requesting consumer? Today's fold treats every installed stage identically. Resolving this — whether by a property a stage declares, a property of the pipeline as configured, or some other mechanism — is part of this issue's scope, not a premise to design around
• A consumer whose authorization changes mid-stream (reauthorize, expire) sees the new authorization reflected starting with its next fetched message, not the authorization in effect when the stream opened
• A model with no authorization-dependent stage installed shows no behavior change and no added cost from this issue — the existing shared, cached path remains untouched in that case
• Unit and k3po coverage: two consumers with different authorization fetching the same underlying record get correspondingly different results; a mid-stream authorization change is reflected on the next message, not retroactively or with a delay beyond that; a model with only authorization-independent stages installed is unaffected, byte for byte, from today's behavior

Additional context

Split out of #1689, which is narrower (wiring ModelEnvelope into the existing read/write paths) and doesn't need this. Scoped to the shared kafka cache storage implementation specifically — the same implementation layer #1688/#1689/#1690 already scope themselves to — not to any other binding's own request-handling path.