fix(binding-kafka): use each producer stream's own authorization for cache encode

config/engine.conf/src/test/java/io/aklivity/zilla/config/engine/test/internal/model/config/TestModelConfig.java

@@ -27,6 +27,7 @@ public class TestModelConfig extends ModelConfig
     public final boolean read;
     public final int transformLength;
     public final List<String> fields;
+    public final List<Long> transformAuthorizations;
 
     public TestModelConfig(
         int length,
@@ -62,12 +63,25 @@ public TestModelConfig(
         int transformLength,
         List<String> fields,
         ValidateConfig validate)
+    {
+        this(length, cataloged, read, transformLength, fields, validate, null);
+    }
+
+    public TestModelConfig(
+        int length,
+        List<CatalogedConfig> cataloged,
+        boolean read,
+        int transformLength,
+        List<String> fields,
+        ValidateConfig validate,
+        List<Long> transformAuthorizations)
     {
         super("test", cataloged, validate);
         this.length = length;
         this.read = read;
         this.transformLength = transformLength;
         this.fields = fields;
+        this.transformAuthorizations = transformAuthorizations;
     }
 
     public static <T> TestModelConfigBuilder<T> builder(

config/engine.conf/src/test/java/io/aklivity/zilla/config/engine/test/internal/model/config/TestModelConfigAdapter.java

@@ -19,6 +19,7 @@
 
 import jakarta.json.Json;
 import jakarta.json.JsonArray;
+import jakarta.json.JsonNumber;
 import jakarta.json.JsonObject;
 import jakarta.json.JsonString;
 import jakarta.json.JsonValue;
@@ -36,6 +37,7 @@ public class TestModelConfigAdapter extends ConfigAdapter<ModelConfig, JsonValue
     private static final String TEST = "test";
     private static final String LENGTH = "length";
     private static final String TRANSFORM = "transform";
+    private static final String TRANSFORM_AUTHORIZATIONS = "transformAuthorizations";
     private static final String CAPABILITY = "capability";
     private static final String READ = "read";
     private static final String CATALOG_NAME = "catalog";
@@ -62,9 +64,19 @@ public TestModelConfig adaptFromJson(
             : 0;
 
         int transformLength = object.containsKey(TRANSFORM)
-            ? object.getJsonObject(TRANSFORM).getInt(LENGTH)
+            ? object.getJsonObject(TRANSFORM).getInt(LENGTH, -1)
             : -1;
 
+        List<Long> transformAuthorizations = null;
+        if (object.containsKey(TRANSFORM_AUTHORIZATIONS))
+        {
+            transformAuthorizations = new LinkedList<>();
+            for (JsonValue item : object.getJsonArray(TRANSFORM_AUTHORIZATIONS))
+            {
+                transformAuthorizations.add(((JsonNumber) item).longValue());
+            }
+        }
+
         boolean read = object.containsKey(CAPABILITY)
             ? object.getString(CAPABILITY).equals(READ)
             : false;
@@ -99,6 +111,6 @@ public TestModelConfig adaptFromJson(
 
         ValidateConfig validateConfig = validate.adaptFromJsonObject(object);
 
-        return new TestModelConfig(length, catalogs, read, transformLength, fields, validateConfig);
+        return new TestModelConfig(length, catalogs, read, transformLength, fields, validateConfig, transformAuthorizations);
     }
 }

config/engine.conf/src/test/java/io/aklivity/zilla/config/engine/test/internal/model/config/TestModelConfigBuilder.java

@@ -34,6 +34,7 @@ public class TestModelConfigBuilder<T> extends ConfigBuilder<T, TestModelConfigB
     private List<CatalogedConfig> catalogs;
     private List<String> fields;
     private ValidateConfig validate;
+    private List<Long> transformAuthorizations;
 
     TestModelConfigBuilder(
         Function<ModelConfig, T> mapper)
@@ -69,6 +70,17 @@ public TestModelConfigBuilder<T> transformLength(
         return this;
     }
 
+    public TestModelConfigBuilder<T> transformAuthorization(
+        long transformAuthorization)
+    {
+        if (transformAuthorizations == null)
+        {
+            transformAuthorizations = new LinkedList<>();
+        }
+        transformAuthorizations.add(transformAuthorization);
+        return this;
+    }
+
     public TestModelConfigBuilder<T> field(
         String field)
     {
@@ -106,6 +118,7 @@ public TestModelConfigBuilder<T> validate(
     @Override
     public T build()
     {
-        return mapper.apply(new TestModelConfig(length, catalogs, read, transformLength, fields, validate));
+        return mapper.apply(
+            new TestModelConfig(length, catalogs, read, transformLength, fields, validate, transformAuthorizations));
     }
 }

runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaCacheClientProduceFactory.java

@@ -269,15 +269,15 @@ public MessageConsumer newStream(
                 final KafkaCache cache = supplyCache.apply(cacheName);
                 final KafkaCacheTopic topic = cache.supplyTopic(topicName);
                 final KafkaCachePartition partition = topic.supplyProducePartition(partitionId, localIndex);
-                final KafkaTopicType topicType = binding.resolveTopicType(topicName);
                 final KafkaCacheClientProduceFan newFan =
                         new KafkaCacheClientProduceFan(routedId, resolvedId, authorization, budget,
-                            partition, cacheRoute, topicName, topicType);
+                            partition, cacheRoute, topicName);
 
                 cacheRoute.clientProduceFansByTopicPartition.put(partitionKey, newFan);
                 fan = newFan;
             }
 
+            final KafkaTopicType topicType = binding.resolveTopicType(topicName);
             final Int2IntHashMap leadersByPartitionId = cacheRoute.supplyLeadersByPartitionId(topicName);
             final int leaderId = leadersByPartitionId.get(partitionId);
             newStream = new KafkaCacheClientProduceStream(
@@ -287,7 +287,8 @@ public MessageConsumer newStream(
                     routedId,
                     initialId,
                     leaderId,
-                    authorization)::onClientMessage;
+                    authorization,
+                    topicType)::onClientMessage;
         }
 
         return newStream;
@@ -507,8 +508,6 @@ final class KafkaCacheClientProduceFan
         private final long routedId;
         private final long authorization;
         private final int partitionId;
-        private final KafkaCacheModel transformKey;
-        private final KafkaCacheModel transformValue;
 
         private long initialId;
         private long replyId;
@@ -545,8 +544,7 @@ private KafkaCacheClientProduceFan(
             KafkaCacheClientBudget budget,
             KafkaCachePartition partition,
             KafkaCacheRoute cacheRoute,
-            String topicName,
-            KafkaTopicType topicType)
+            String topicName)
         {
             this.originId = originId;
             this.routedId = routedId;
@@ -556,8 +554,6 @@ private KafkaCacheClientProduceFan(
             this.budget = budget;
             this.cacheRoute = cacheRoute;
             this.topicName = topicName;
-            this.transformKey = KafkaCacheModel.encoder(topicType.keyModel, transformBuffer);
-            this.transformValue = KafkaCacheModel.encoder(topicType.valueModel, transformBuffer);
             this.members = new Long2ObjectHashMap<>();
             this.defaultOffset = KafkaOffsetType.LIVE;
             this.cursor = cursorFactory.newCursor(
@@ -718,10 +714,10 @@ private void onClientInitialData(
                     assert partitionOffset >= 0 && partitionOffset >= nextOffset
                         : String.format("%d >= 0 && %d >= %d", partitionOffset, partitionOffset, nextOffset);
 
-                    if (partition.writeProduceEntryStart(traceId, routedId, authorization, partitionOffset, stream.segment,
-                        stream.entryMark, stream.valueMark, stream.valueLimit, timestamp, stream.initialId,
+                    if (partition.writeProduceEntryStart(traceId, routedId, stream.authorization, partitionOffset,
+                        stream.segment, stream.entryMark, stream.valueMark, stream.valueLimit, timestamp, stream.initialId,
                         producerId, producerEpoch, sequence, ackMode, key, valueLength,
-                        headers, trailersSizeMax, valueFragment, transformKey, transformValue) == -1)
+                        headers, trailersSizeMax, valueFragment, stream.transformKey, stream.transformValue) == -1)
                     {
                         error = ERROR_INVALID_RECORD;
                         break init;
@@ -737,9 +733,9 @@ private void onClientInitialData(
 
             if (valueFragment != null && error == NO_ERROR)
             {
-                if (partition.writeProduceEntryContinue(traceId, routedId, authorization, flags, stream.segment,
+                if (partition.writeProduceEntryContinue(traceId, routedId, stream.authorization, flags, stream.segment,
                         stream.entryMark, stream.valueMark, stream.valueLimit,
-                        valueFragment, transformValue) == -1)
+                        valueFragment, stream.transformValue) == -1)
                 {
                     error = ERROR_INVALID_RECORD;
                 }
@@ -798,10 +794,10 @@ private void onClientInitialFlush(
                 assert partitionOffset >= 0 && partitionOffset >= nextOffset
                     : String.format("%d >= 0 && %d >= %d", partitionOffset, partitionOffset, nextOffset);
 
-                partition.writeProduceEntryStart(traceId, routedId, authorization, partitionOffset, stream.segment,
+                partition.writeProduceEntryStart(traceId, routedId, stream.authorization, partitionOffset, stream.segment,
                     stream.entryMark, stream.valueMark, stream.valueLimit, now().toEpochMilli(), stream.initialId,
                     PRODUCE_FLUSH_PRODUCER_ID, PRODUCE_FLUSH_PRODUCER_EPOCH, PRODUCE_FLUSH_SEQUENCE, KafkaAckMode.LEADER_ONLY,
-                    EMPTY_KEY, 0, EMPTY_TRAILERS, trailersSizeMax, EMPTY_OCTETS, transformKey, transformValue);
+                    EMPTY_KEY, 0, EMPTY_TRAILERS, trailersSizeMax, EMPTY_OCTETS, stream.transformKey, stream.transformValue);
                 stream.partitionOffset = partitionOffset;
                 partitionOffset++;
 
@@ -1236,6 +1232,8 @@ private final class KafkaCacheClientProduceStream
         private final long replyId;
         private final long leaderId;
         private final long authorization;
+        private final KafkaCacheModel transformKey;
+        private final KafkaCacheModel transformValue;
 
         private long partitionOffset = DEFAULT_LATEST_OFFSET;
 
@@ -1260,7 +1258,8 @@ private final class KafkaCacheClientProduceStream
             long routedId,
             long initialId,
             long leaderId,
-            long authorization)
+            long authorization,
+            KafkaTopicType topicType)
         {
             this.cursor = cursorFactory.newCursor(
                     cursorFactory
@@ -1277,6 +1276,8 @@ private final class KafkaCacheClientProduceStream
             this.replyId = supplyReplyId.applyAsLong(initialId);
             this.leaderId = leaderId;
             this.authorization = authorization;
+            this.transformKey = KafkaCacheModel.encoder(topicType.keyModel, transformBuffer);
+            this.transformValue = KafkaCacheModel.encoder(topicType.valueModel, transformBuffer);
         }
 
         private void onClientMessage(

runtime/binding-kafka/src/test/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/CacheProduceIT.java

@@ -137,6 +137,18 @@ public void shouldRetryPartitionNotLeaderMessageValues() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Configuration("cache.value.model.authorization.yaml")
+    @Specification({
+        "${app}/message.values.authorization.distinct/client",
+        "${app}/message.values.authorization.distinct/server"})
+    @ScriptProperty("serverAddress \"zilla://streams/app1\"")
+    @Configure(name = KAFKA_CACHE_SERVER_RECONNECT_DELAY_NAME, value = "1")
+    public void shouldSendMessageValuesAuthorizationDistinct() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Test
     @Configuration("cache.yaml")
     @Specification({

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/model/TestModelHandler.java

@@ -33,6 +33,9 @@ public class TestModelHandler implements ModelHandler
     private final List<String> fields;
     private final boolean decodeLenient;
     private final boolean encodeLenient;
+    private final List<Long> transformAuthorizations;
+
+    private int transformAuthorizationIndex;
 
     public TestModelHandler(
         TestModelConfig config)
@@ -42,21 +45,29 @@ public TestModelHandler(
         this.fields = config.fields != null ? config.fields : emptyList();
         this.decodeLenient = config.validate.decode == ValidateMode.LENIENT;
         this.encodeLenient = config.validate.encode == ValidateMode.LENIENT;
+        this.transformAuthorizations = config.transformAuthorizations;
     }
 
     @Override
     public ModelPipeline supplyDecoder(
         ModelEnvelope envelope,
         ModelTransform transform)
     {
-        return new TestModelPipeline(length, transformLength, fields, decodeLenient, envelope, transform);
+        return new TestModelPipeline(length, transformLength, fields, decodeLenient, envelope, transform, this);
     }
 
     @Override
     public ModelPipeline supplyEncoder(
         ModelEnvelope envelope,
         ModelTransform transform)
     {
-        return new TestModelPipeline(length, transformLength, fields, encodeLenient, envelope, transform);
+        return new TestModelPipeline(length, transformLength, fields, encodeLenient, envelope, transform, this);
+    }
+
+    Long nextTransformAuthorization()
+    {
+        return transformAuthorizations != null && transformAuthorizationIndex < transformAuthorizations.size()
+            ? transformAuthorizations.get(transformAuthorizationIndex++)
+            : null;
     }
 }

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/model/TestModelPipeline.java

@@ -39,6 +39,16 @@
 // value when an accepted value completes; the same fields are written to the supplied envelope under their
 // own paths, so a caller supplying a real envelope observes what the model surfaced without wiring a
 // transform at all. State lives on the pipeline so interleaved streams stay isolated.
+//
+// When the handler is configured with an ordered `transformAuthorizations` list, each completed value --
+// across every pipeline this handler ever supplies, not just this one -- consumes the next entry from that
+// list as its expected authorization: message order, not pipeline-instance order, is what the list tracks.
+// A completed value whose `authorization` argument doesn't match is rejected, giving callers a way to assert
+// which authorization value actually reached a given encode/decode call without any extra observability
+// machinery -- the mismatch surfaces as an ordinary REJECTED status, same as a length violation. Binding the
+// check to pipeline-construction order instead would miss exactly the bug this exists to catch: a single
+// pipeline instance shared across multiple producers only ever sees one expected value if it were fixed at
+// construction, even though it processes several messages, each with its own authorization.
 final class TestModelPipeline implements ModelPipeline
 {
     private static final int FLAGS_INIT = 0x02;
@@ -53,6 +63,7 @@ final class TestModelPipeline implements ModelPipeline
     private final ModelEnvelope envelope;
     private final ModelFieldBridge bridge;
     private final ModelPipelineResult result;
+    private final TestModelHandler handler;
 
     private int processed;
 
@@ -62,7 +73,8 @@ final class TestModelPipeline implements ModelPipeline
         List<String> fields,
         boolean lenient,
         ModelEnvelope envelope,
-        ModelTransform transform)
+        ModelTransform transform,
+        TestModelHandler handler)
     {
         this.length = length;
         this.transformLength = transformLength;
@@ -71,6 +83,7 @@ final class TestModelPipeline implements ModelPipeline
         this.envelope = envelope;
         this.bridge = transform != ModelTransform.NONE ? new ModelFieldBridge(transform) : null;
         this.result = new ModelPipelineResult();
+        this.handler = handler;
     }
 
     @Override
@@ -148,6 +161,16 @@ else if (tail)
                 status = ModelStatus.UNDERFLOW;
             }
         }
+
+        if (status == ModelStatus.COMPLETE)
+        {
+            final Long expectedAuthorization = handler.nextTransformAuthorization();
+            if (expectedAuthorization != null && authorization != expectedAuthorization)
+            {
+                status = ModelStatus.REJECTED;
+            }
+        }
+
         return result.set(status, consumed, produced);
     }
 

specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/config/cache.value.model.authorization.yaml

@@ -0,0 +1,40 @@
+#
+# Copyright 2021-2026 Aklivity Inc.
+#
+# Aklivity licenses this file to you under the Apache License,
+# version 2.0 (the "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at:
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+
+---
+name: test
+bindings:
+    app0:
+        type: kafka
+        kind: cache_client
+        options:
+            topics:
+                - name: test
+                  value:
+                      model: test
+                      length: 12
+                      transformAuthorizations: [2, 0]
+        routes:
+            - exit: cache0
+              when:
+                  - topic: test
+    cache0:
+        type: kafka
+        kind: cache_server
+        routes:
+            - exit: app1
+              when:
+                  - topic: test