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..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,6 +27,7 @@ public class TestModelConfig extends ModelConfig public final boolean read; public final int transformLength; public final List fields; + public final List transformAuthorizations; public TestModelConfig( int length, @@ -62,12 +63,25 @@ public TestModelConfig( int transformLength, List fields, ValidateConfig validate) + { + this(length, cataloged, read, transformLength, fields, validate, null); + } + + public TestModelConfig( + int length, + List cataloged, + boolean read, + int transformLength, + List fields, + ValidateConfig validate, + List transformAuthorizations) { super("test", cataloged, validate); this.length = length; this.read = read; this.transformLength = transformLength; this.fields = fields; + 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 bb67384a50c..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,6 +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) : 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); } } 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 6e901023bd7..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,6 +34,7 @@ public class TestModelConfigBuilder extends ConfigBuilder catalogs; private List fields; private ValidateConfig validate; + private List transformAuthorizations; TestModelConfigBuilder( Function mapper) @@ -69,6 +70,17 @@ public TestModelConfigBuilder transformLength( return this; } + public TestModelConfigBuilder transformAuthorization( + long transformAuthorization) + { + if (transformAuthorizations == null) + { + transformAuthorizations = new LinkedList<>(); + } + transformAuthorizations.add(transformAuthorization); + return this; + } + public TestModelConfigBuilder field( String field) { @@ -106,6 +118,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, transformAuthorizations)); } } 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..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,6 +33,9 @@ public class TestModelHandler implements ModelHandler private final List fields; private final boolean decodeLenient; private final boolean encodeLenient; + private final List transformAuthorizations; + + private int transformAuthorizationIndex; public TestModelHandler( TestModelConfig config) @@ -42,6 +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.transformAuthorizations = config.transformAuthorizations; } @Override @@ -49,7 +53,7 @@ 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 @@ -57,6 +61,13 @@ 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; } } 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..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; @@ -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 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); } 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..71807f38204 --- /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,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 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..37f0e58f9f6 --- /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 "Hello, again" + +read zilla:data.ext ${kafka:matchDataEx() + .typeId(zilla:id("kafka")) + .produce() + .build() + .build()} +read "Hello, world" + + + +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..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 @@ -82,6 +82,14 @@ "type": "string" } }, + "transformAuthorizations": + { + "type": "array", + "items": + { + "type": "integer" + } + }, "catalog": { "type": "object",