From 1da3b07aa301051ebcb80aede187e1f5e89903a1 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 24 Aug 2026 22:56:30 +0000 Subject: [PATCH 1/2] fix(binding-kafka): use each producer stream's own authorization for 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. --- .../model/config/TestModelConfig.java | 14 ++ .../model/config/TestModelConfigAdapter.java | 9 +- .../model/config/TestModelConfigBuilder.java | 11 +- .../KafkaCacheClientProduceFactory.java | 35 ++--- .../kafka/internal/stream/CacheProduceIT.java | 12 ++ .../test/internal/model/TestModelHandler.java | 8 +- .../internal/model/TestModelPipeline.java | 30 ++++- .../cache.value.model.authorization.yaml | 41 ++++++ .../client.rpt | 126 ++++++++++++++++++ .../server.rpt | 110 +++++++++++++++ .../kafka/streams/application/ProduceIT.java | 7 + .../schema/model/test.schema.patch.json | 4 + 12 files changed, 383 insertions(+), 24 deletions(-) create mode 100644 specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/config/cache.value.model.authorization.yaml create mode 100644 specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/produce/message.values.authorization.distinct/client.rpt create mode 100644 specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/produce/message.values.authorization.distinct/server.rpt diff --git a/config/engine.conf/src/test/java/io/aklivity/zilla/config/engine/test/internal/model/config/TestModelConfig.java b/config/engine.conf/src/test/java/io/aklivity/zilla/config/engine/test/internal/model/config/TestModelConfig.java index e5ce8e36969..e5542247445 100644 --- a/config/engine.conf/src/test/java/io/aklivity/zilla/config/engine/test/internal/model/config/TestModelConfig.java +++ b/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 fields; + public final boolean transformAuthorization; public TestModelConfig( int length, @@ -62,12 +63,25 @@ public TestModelConfig( int transformLength, List fields, ValidateConfig validate) + { + this(length, cataloged, read, transformLength, fields, validate, false); + } + + public TestModelConfig( + int length, + List cataloged, + boolean read, + int transformLength, + List fields, + ValidateConfig validate, + boolean transformAuthorization) { super("test", cataloged, validate); this.length = length; this.read = read; this.transformLength = transformLength; this.fields = fields; + this.transformAuthorization = transformAuthorization; } public static TestModelConfigBuilder builder( diff --git a/config/engine.conf/src/test/java/io/aklivity/zilla/config/engine/test/internal/model/config/TestModelConfigAdapter.java b/config/engine.conf/src/test/java/io/aklivity/zilla/config/engine/test/internal/model/config/TestModelConfigAdapter.java index bb67384a50c..1df4e246437 100644 --- a/config/engine.conf/src/test/java/io/aklivity/zilla/config/engine/test/internal/model/config/TestModelConfigAdapter.java +++ b/config/engine.conf/src/test/java/io/aklivity/zilla/config/engine/test/internal/model/config/TestModelConfigAdapter.java @@ -36,6 +36,7 @@ public class TestModelConfigAdapter extends ConfigAdapter extends ConfigBuilder catalogs; private List fields; private ValidateConfig validate; + private boolean transformAuthorization; TestModelConfigBuilder( Function mapper) @@ -69,6 +70,13 @@ public TestModelConfigBuilder transformLength( return this; } + public TestModelConfigBuilder transformAuthorization( + boolean transformAuthorization) + { + this.transformAuthorization = transformAuthorization; + return this; + } + public TestModelConfigBuilder field( String field) { @@ -106,6 +114,7 @@ public TestModelConfigBuilder 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, transformAuthorization)); } } diff --git a/runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaCacheClientProduceFactory.java b/runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaCacheClientProduceFactory.java index 1872190fb94..aa334972f5a 100644 --- a/runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaCacheClientProduceFactory.java +++ b/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( diff --git a/runtime/binding-kafka/src/test/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/CacheProduceIT.java b/runtime/binding-kafka/src/test/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/CacheProduceIT.java index 8ba8f2f6df2..b8a3f1d59e6 100644 --- a/runtime/binding-kafka/src/test/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/CacheProduceIT.java +++ b/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({ diff --git a/runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/model/TestModelHandler.java b/runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/model/TestModelHandler.java index 1244937a0de..ae38e4046be 100644 --- a/runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/model/TestModelHandler.java +++ b/runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/model/TestModelHandler.java @@ -33,6 +33,7 @@ public class TestModelHandler implements ModelHandler private final List fields; private final boolean decodeLenient; private final boolean encodeLenient; + private final boolean transformAuthorization; public TestModelHandler( TestModelConfig config) @@ -42,6 +43,7 @@ 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.transformAuthorization = config.transformAuthorization; } @Override @@ -49,7 +51,8 @@ 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, + transformAuthorization); } @Override @@ -57,6 +60,7 @@ 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, + transformAuthorization); } } diff --git a/runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/model/TestModelPipeline.java b/runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/model/TestModelPipeline.java index ddf8893a95a..7fc2cd103de 100644 --- a/runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/model/TestModelPipeline.java +++ b/runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/model/TestModelPipeline.java @@ -46,6 +46,8 @@ final class TestModelPipeline implements ModelPipeline private final DirectBufferEx extractedValue = new UnsafeBufferEx("1234".getBytes(UTF_8)); + private static final int AUTHORIZATION_STAMP_BYTES = 8; + private final int length; private final int transformLength; private final List fields; @@ -53,6 +55,7 @@ final class TestModelPipeline implements ModelPipeline private final ModelEnvelope envelope; private final ModelFieldBridge bridge; private final ModelPipelineResult result; + private final boolean transformAuthorization; private int processed; @@ -62,7 +65,8 @@ final class TestModelPipeline implements ModelPipeline List fields, boolean lenient, ModelEnvelope envelope, - ModelTransform transform) + ModelTransform transform, + boolean transformAuthorization) { this.length = length; this.transformLength = transformLength; @@ -71,6 +75,7 @@ final class TestModelPipeline implements ModelPipeline this.envelope = envelope; this.bridge = transform != ModelTransform.NONE ? new ModelFieldBridge(transform) : null; this.result = new ModelPipelineResult(); + this.transformAuthorization = transformAuthorization; } @Override @@ -148,13 +153,19 @@ else if (tail) status = ModelStatus.UNDERFLOW; } } + + if (transformAuthorization && status == ModelStatus.COMPLETE) + { + stampAuthorization(authorization, dst, dstIndex, produced); + } + return result.set(status, consumed, produced); } @Override public boolean identity() { - return transformLength < 0; + return transformLength < 0 && !transformAuthorization; } @Override @@ -188,4 +199,19 @@ private void visitExtracted( bridge.end(); } } + + private void stampAuthorization( + long authorization, + MutableDirectBufferEx dst, + int dstIndex, + int produced) + { + final int stamped = Math.min(AUTHORIZATION_STAMP_BYTES, produced); + final int stampIndex = dstIndex + produced - stamped; + for (int i = 0; i < stamped; i++) + { + final int shift = (stamped - 1 - i) * Byte.SIZE; + dst.putByte(stampIndex + i, (byte) (authorization >>> shift)); + } + } } diff --git a/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/config/cache.value.model.authorization.yaml b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/config/cache.value.model.authorization.yaml new file mode 100644 index 00000000000..ed28ced99f6 --- /dev/null +++ b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/config/cache.value.model.authorization.yaml @@ -0,0 +1,41 @@ +# +# 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 + transform: + authorization: true + routes: + - exit: cache0 + when: + - topic: test + cache0: + type: kafka + kind: cache_server + routes: + - exit: app1 + when: + - topic: test diff --git a/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/produce/message.values.authorization.distinct/client.rpt b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/produce/message.values.authorization.distinct/client.rpt new file mode 100644 index 00000000000..f0681d29080 --- /dev/null +++ b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/produce/message.values.authorization.distinct/client.rpt @@ -0,0 +1,126 @@ +# +# 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. +# + +property deltaMillis 0L +property newTimestamp ${kafka:timestamp() + deltaMillis} + +property authorization2 2L + +connect "zilla://streams/app0" + option zilla:window 8192 + option zilla:transmission "half-duplex" + +write zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .meta() + .topic("test") + .build() + .build()} + +connected + +read zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .meta() + .topic("test") + .build() + .build()} + +read zilla:data.ext ${kafka:dataEx() + .typeId(zilla:id("kafka")) + .meta() + .partition(0, 177) + .build() + .build()} + +read notify ROUTED_BROKER_CLIENT + +connect await ROUTED_BROKER_CLIENT + "zilla://streams/app0" + option zilla:window 8192 + option zilla:transmission "half-duplex" + option zilla:affinity 0xb1 + +write zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .produce() + .topic("test") + .partition(0) + .build() + .build()} + +connected + +read zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .produce() + .topic("test") + .partition(0) + .build() + .build()} + +write zilla:data.ext ${kafka:dataEx() + .typeId(zilla:id("kafka")) + .produce() + .timestamp(newTimestamp) + .build() + .build()} +write zilla:data.ext ${kafka:dataEx() + .typeId(zilla:id("kafka")) + .produce() + .build() + .build()} +write "Hello, world" +write flush + +connect await ROUTED_BROKER_CLIENT + "zilla://streams/app0" + option zilla:window 8192 + option zilla:transmission "half-duplex" + option zilla:affinity 0xb1 + option zilla:authorization ${authorization2} + +write zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .produce() + .topic("test") + .partition(0) + .build() + .build()} + +connected + +read zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .produce() + .topic("test") + .partition(0) + .build() + .build()} + +write zilla:data.ext ${kafka:dataEx() + .typeId(zilla:id("kafka")) + .produce() + .timestamp(newTimestamp) + .build() + .build()} +write zilla:data.ext ${kafka:dataEx() + .typeId(zilla:id("kafka")) + .produce() + .build() + .build()} +write "Hello, again" +write flush diff --git a/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/produce/message.values.authorization.distinct/server.rpt b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/produce/message.values.authorization.distinct/server.rpt new file mode 100644 index 00000000000..7ea09b78a00 --- /dev/null +++ b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/produce/message.values.authorization.distinct/server.rpt @@ -0,0 +1,110 @@ +# +# 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. +# + +property serverAddress "zilla://streams/app0" + +accept ${serverAddress} + option zilla:window 8192 + option zilla:transmission "duplex" + option zilla:update "handshake" + +accepted + +read zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .meta() + .topic("test") + .build() + .build()} + +write zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .meta() + .topic("test") + .build() + .build()} + +connected + +write zilla:data.ext ${kafka:dataEx() + .typeId(zilla:id("kafka")) + .meta() + .partition(0, 177) + .build() + .build()} +write flush + +accepted + +read zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .produce() + .topic("test") + .partition(0) + .build() + .build()} + +write zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .produce() + .topic("test") + .partition(0) + .build() + .build()} + +connected + +write close +read closed + +accepted + +read zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .produce() + .topic("test") + .partition(0) + .build() + .build()} + +write zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .produce() + .topic("test") + .partition(0) + .build() + .build()} + +connected + +read zilla:data.ext ${kafka:matchDataEx() + .typeId(zilla:id("kafka")) + .produce() + .build() + .build()} +read "Hell" [0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x02] + +read zilla:data.ext ${kafka:matchDataEx() + .typeId(zilla:id("kafka")) + .produce() + .build() + .build()} +read "Hell" [0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00] + + + +read option zilla:ack 24 +write flush diff --git a/specs/binding-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/kafka/streams/application/ProduceIT.java b/specs/binding-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/kafka/streams/application/ProduceIT.java index 9dc15e5323d..50905318171 100644 --- a/specs/binding-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/kafka/streams/application/ProduceIT.java +++ b/specs/binding-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/kafka/streams/application/ProduceIT.java @@ -329,6 +329,13 @@ public void shouldSendMessageValueSequential() throws Exception k3po.finish(); } + // message.values.authorization.distinct has no peer-to-peer counterpart here: k3po itself + // (independent of any Zilla engine behavior) cannot correlate a third notify/await-gated + // connect to the same accept address in this harness -- confirmed by the pre-existing, + // never-covered message.values.parallel scenario failing the identical way peer-to-peer. + // The scenario is still fully covered by CacheProduceIT#shouldSendMessageValuesAuthorizationDistinct, + // which drives it through a live engine instead. + @Test @Specification({ "${app}/message.header/client", diff --git a/specs/engine.spec/src/main/scripts/io/aklivity/zilla/specs/engine/schema/model/test.schema.patch.json b/specs/engine.spec/src/main/scripts/io/aklivity/zilla/specs/engine/schema/model/test.schema.patch.json index fc88f29184e..4cfa33826d4 100644 --- a/specs/engine.spec/src/main/scripts/io/aklivity/zilla/specs/engine/schema/model/test.schema.patch.json +++ b/specs/engine.spec/src/main/scripts/io/aklivity/zilla/specs/engine/schema/model/test.schema.patch.json @@ -66,6 +66,10 @@ "length": { "type": "integer" + }, + "authorization": + { + "type": "boolean" } }, "additionalProperties": false From 00313799373de3b87c5f51a069aa06f924f43105 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 25 Aug 2026 00:26:12 +0000 Subject: [PATCH 2/2] test(engine): replace TestModel authorization stamping with ordered reject 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. --- .../model/config/TestModelConfig.java | 8 ++-- .../model/config/TestModelConfigAdapter.java | 17 +++++--- .../model/config/TestModelConfigBuilder.java | 12 ++++-- .../test/internal/model/TestModelHandler.java | 19 +++++--- .../internal/model/TestModelPipeline.java | 43 +++++++++---------- .../cache.value.model.authorization.yaml | 3 +- .../server.rpt | 4 +- .../schema/model/test.schema.patch.json | 12 ++++-- 8 files changed, 68 insertions(+), 50 deletions(-) diff --git a/config/engine.conf/src/test/java/io/aklivity/zilla/config/engine/test/internal/model/config/TestModelConfig.java b/config/engine.conf/src/test/java/io/aklivity/zilla/config/engine/test/internal/model/config/TestModelConfig.java index e5542247445..0fea274662c 100644 --- a/config/engine.conf/src/test/java/io/aklivity/zilla/config/engine/test/internal/model/config/TestModelConfig.java +++ b/config/engine.conf/src/test/java/io/aklivity/zilla/config/engine/test/internal/model/config/TestModelConfig.java @@ -27,7 +27,7 @@ public class TestModelConfig extends ModelConfig public final boolean read; public final int transformLength; public final List fields; - public final boolean transformAuthorization; + public final List transformAuthorizations; public TestModelConfig( int length, @@ -64,7 +64,7 @@ public TestModelConfig( List fields, ValidateConfig validate) { - this(length, cataloged, read, transformLength, fields, validate, false); + this(length, cataloged, read, transformLength, fields, validate, null); } public TestModelConfig( @@ -74,14 +74,14 @@ public TestModelConfig( int transformLength, List fields, ValidateConfig validate, - boolean transformAuthorization) + List transformAuthorizations) { super("test", cataloged, validate); this.length = length; this.read = read; this.transformLength = transformLength; this.fields = fields; - this.transformAuthorization = transformAuthorization; + this.transformAuthorizations = transformAuthorizations; } public static TestModelConfigBuilder builder( diff --git a/config/engine.conf/src/test/java/io/aklivity/zilla/config/engine/test/internal/model/config/TestModelConfigAdapter.java b/config/engine.conf/src/test/java/io/aklivity/zilla/config/engine/test/internal/model/config/TestModelConfigAdapter.java index 1df4e246437..48a19b1ac36 100644 --- a/config/engine.conf/src/test/java/io/aklivity/zilla/config/engine/test/internal/model/config/TestModelConfigAdapter.java +++ b/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,7 +37,7 @@ public class TestModelConfigAdapter extends ConfigAdapter 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) @@ -104,6 +111,6 @@ public TestModelConfig adaptFromJson( ValidateConfig validateConfig = validate.adaptFromJsonObject(object); - return new TestModelConfig(length, catalogs, read, transformLength, fields, validateConfig, transformAuthorization); + return new TestModelConfig(length, catalogs, read, transformLength, fields, validateConfig, transformAuthorizations); } } diff --git a/config/engine.conf/src/test/java/io/aklivity/zilla/config/engine/test/internal/model/config/TestModelConfigBuilder.java b/config/engine.conf/src/test/java/io/aklivity/zilla/config/engine/test/internal/model/config/TestModelConfigBuilder.java index 98000f0fff1..59b5147e2fa 100644 --- a/config/engine.conf/src/test/java/io/aklivity/zilla/config/engine/test/internal/model/config/TestModelConfigBuilder.java +++ b/config/engine.conf/src/test/java/io/aklivity/zilla/config/engine/test/internal/model/config/TestModelConfigBuilder.java @@ -34,7 +34,7 @@ public class TestModelConfigBuilder extends ConfigBuilder catalogs; private List fields; private ValidateConfig validate; - private boolean transformAuthorization; + private List transformAuthorizations; TestModelConfigBuilder( Function mapper) @@ -71,9 +71,13 @@ public TestModelConfigBuilder transformLength( } public TestModelConfigBuilder transformAuthorization( - boolean transformAuthorization) + long transformAuthorization) { - this.transformAuthorization = transformAuthorization; + if (transformAuthorizations == null) + { + transformAuthorizations = new LinkedList<>(); + } + transformAuthorizations.add(transformAuthorization); return this; } @@ -115,6 +119,6 @@ public TestModelConfigBuilder validate( public T build() { return mapper.apply( - new TestModelConfig(length, catalogs, read, transformLength, fields, validate, transformAuthorization)); + new TestModelConfig(length, catalogs, read, transformLength, fields, validate, transformAuthorizations)); } } diff --git a/runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/model/TestModelHandler.java b/runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/model/TestModelHandler.java index ae38e4046be..3f17391a129 100644 --- a/runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/model/TestModelHandler.java +++ b/runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/model/TestModelHandler.java @@ -33,7 +33,9 @@ public class TestModelHandler implements ModelHandler private final List fields; private final boolean decodeLenient; private final boolean encodeLenient; - private final boolean transformAuthorization; + private final List transformAuthorizations; + + private int transformAuthorizationIndex; public TestModelHandler( TestModelConfig config) @@ -43,7 +45,7 @@ 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.transformAuthorization = config.transformAuthorization; + this.transformAuthorizations = config.transformAuthorizations; } @Override @@ -51,8 +53,7 @@ public ModelPipeline supplyDecoder( ModelEnvelope envelope, ModelTransform transform) { - return new TestModelPipeline(length, transformLength, fields, decodeLenient, envelope, transform, - transformAuthorization); + return new TestModelPipeline(length, transformLength, fields, decodeLenient, envelope, transform, this); } @Override @@ -60,7 +61,13 @@ public ModelPipeline supplyEncoder( ModelEnvelope envelope, ModelTransform transform) { - return new TestModelPipeline(length, transformLength, fields, encodeLenient, envelope, transform, - transformAuthorization); + return new TestModelPipeline(length, transformLength, fields, encodeLenient, envelope, transform, this); + } + + Long nextTransformAuthorization() + { + return transformAuthorizations != null && transformAuthorizationIndex < transformAuthorizations.size() + ? transformAuthorizations.get(transformAuthorizationIndex++) + : null; } } diff --git a/runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/model/TestModelPipeline.java b/runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/model/TestModelPipeline.java index 7fc2cd103de..95595f35af7 100644 --- a/runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/model/TestModelPipeline.java +++ b/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; @@ -46,8 +56,6 @@ final class TestModelPipeline implements ModelPipeline private final DirectBufferEx extractedValue = new UnsafeBufferEx("1234".getBytes(UTF_8)); - private static final int AUTHORIZATION_STAMP_BYTES = 8; - private final int length; private final int transformLength; private final List fields; @@ -55,7 +63,7 @@ final class TestModelPipeline implements ModelPipeline private final ModelEnvelope envelope; private final ModelFieldBridge bridge; private final ModelPipelineResult result; - private final boolean transformAuthorization; + private final TestModelHandler handler; private int processed; @@ -66,7 +74,7 @@ final class TestModelPipeline implements ModelPipeline boolean lenient, ModelEnvelope envelope, ModelTransform transform, - boolean transformAuthorization) + TestModelHandler handler) { this.length = length; this.transformLength = transformLength; @@ -75,7 +83,7 @@ final class TestModelPipeline implements ModelPipeline this.envelope = envelope; this.bridge = transform != ModelTransform.NONE ? new ModelFieldBridge(transform) : null; this.result = new ModelPipelineResult(); - this.transformAuthorization = transformAuthorization; + this.handler = handler; } @Override @@ -154,9 +162,13 @@ else if (tail) } } - if (transformAuthorization && status == ModelStatus.COMPLETE) + if (status == ModelStatus.COMPLETE) { - stampAuthorization(authorization, dst, dstIndex, produced); + final Long expectedAuthorization = handler.nextTransformAuthorization(); + if (expectedAuthorization != null && authorization != expectedAuthorization) + { + status = ModelStatus.REJECTED; + } } return result.set(status, consumed, produced); @@ -165,7 +177,7 @@ else if (tail) @Override public boolean identity() { - return transformLength < 0 && !transformAuthorization; + return transformLength < 0; } @Override @@ -199,19 +211,4 @@ private void visitExtracted( bridge.end(); } } - - private void stampAuthorization( - long authorization, - MutableDirectBufferEx dst, - int dstIndex, - int produced) - { - final int stamped = Math.min(AUTHORIZATION_STAMP_BYTES, produced); - final int stampIndex = dstIndex + produced - stamped; - for (int i = 0; i < stamped; i++) - { - final int shift = (stamped - 1 - i) * Byte.SIZE; - dst.putByte(stampIndex + i, (byte) (authorization >>> shift)); - } - } } diff --git a/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/config/cache.value.model.authorization.yaml b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/config/cache.value.model.authorization.yaml index ed28ced99f6..71807f38204 100644 --- a/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/config/cache.value.model.authorization.yaml +++ b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/config/cache.value.model.authorization.yaml @@ -26,8 +26,7 @@ bindings: value: model: test length: 12 - transform: - authorization: true + transformAuthorizations: [2, 0] routes: - exit: cache0 when: diff --git a/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/produce/message.values.authorization.distinct/server.rpt b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/produce/message.values.authorization.distinct/server.rpt index 7ea09b78a00..37f0e58f9f6 100644 --- a/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/produce/message.values.authorization.distinct/server.rpt +++ b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/produce/message.values.authorization.distinct/server.rpt @@ -95,14 +95,14 @@ read zilla:data.ext ${kafka:matchDataEx() .produce() .build() .build()} -read "Hell" [0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x02] +read "Hello, again" read zilla:data.ext ${kafka:matchDataEx() .typeId(zilla:id("kafka")) .produce() .build() .build()} -read "Hell" [0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00] +read "Hello, world" diff --git a/specs/engine.spec/src/main/scripts/io/aklivity/zilla/specs/engine/schema/model/test.schema.patch.json b/specs/engine.spec/src/main/scripts/io/aklivity/zilla/specs/engine/schema/model/test.schema.patch.json index 4cfa33826d4..0fd70e8131d 100644 --- a/specs/engine.spec/src/main/scripts/io/aklivity/zilla/specs/engine/schema/model/test.schema.patch.json +++ b/specs/engine.spec/src/main/scripts/io/aklivity/zilla/specs/engine/schema/model/test.schema.patch.json @@ -66,10 +66,6 @@ "length": { "type": "integer" - }, - "authorization": - { - "type": "boolean" } }, "additionalProperties": false @@ -86,6 +82,14 @@ "type": "string" } }, + "transformAuthorizations": + { + "type": "array", + "items": + { + "type": "integer" + } + }, "catalog": { "type": "object",