Skip to content

fix(binding-kafka): derive the merged produce window from the least budget offered #2546

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jfallows merged 2 commits into from
Sep 4, 2026
+852 −6
Merged

fix(binding-kafka): derive the merged produce window from the least budget offered #2546

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ccc296d
fix(binding-kafka): derive the merged produce window from the least b…
sfr-oc Sep 3, 2026
1a4495b
test(binding-kafka): cover the merged produce window with an idle par…
sfr-oc Sep 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 11 additions & 6 deletions ...main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaMergedFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1798,16 +1798,21 @@ private void doMergedInitialWindow(
initialMaxRW.value = Integer.MAX_VALUE;
produceStreams.forEach(p -> initialNoAckRW.value = Math.max(p.initialNoAck(), initialNoAckRW.value));
produceStreams.forEach(p -> initialPadRW.value = Math.max(p.initialPad, initialPadRW.value));
produceStreams.forEach(p -> initialMaxRW.value = Math.min(p.initialMax, initialMaxRW.value));
// a partition carrying the traffic holds its acknowledge and grows its maximum, while
// idle partitions keep the maximum they started with, so minimizing the maximum here
// would pair the busiest partition's unacknowledged bytes with an idle partition's
// maximum and shrink this window to nothing; minimize the budget each one offers
produceStreams.forEach(p -> initialMaxRW.value =
Math.min(p.initialMax - p.initialNoAck(), initialMaxRW.value));

maxInitialNoAck = initialNoAckRW.value;
maxInitialPad = initialPadRW.value;
minInitialMax = initialMaxRW.value + maxInitialNoAck;

if (producer != null)
{
initialMaxRW.value = Math.max(producer.initialMax, initialMaxRW.value);
minInitialMax = Math.max(producer.initialMax, minInitialMax);
}

maxInitialNoAck = initialNoAckRW.value;
maxInitialPad = initialPadRW.value;
minInitialMax = initialMaxRW.value;
}

final long newInitialAck = Math.max(initialSeq - maxInitialNoAck, initialAck);
Expand Down
10 changes: 10 additions & 0 deletions .../src/test/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/CacheMergedIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -501,6 +501,16 @@ public void shouldProduceMergedMessageValuesDynamicHashKey() throws Exception
k3po.finish();
}

@Test
@Configuration("cache.options.merged.yaml")
@Specification({
"${app}/merged.produce.message.values.partition.idle/client",
"${app}/unmerged.produce.message.values.partition.idle/server"})
public void shouldProduceMergedMessageValuesPartitionIdle() throws Exception
{
k3po.finish();
}

@Test
@Configuration("cache.options.merged.yaml")
@Specification({
Expand Down
154 changes: 154 additions & 0 deletions .../kafka/streams/application/merged/merged.produce.message.values.partition.idle/client.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,154 @@
#
# 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}

connect "zilla://streams/app0"
option zilla:window 8192
option zilla:transmission "half-duplex"

write zilla:begin.ext ${kafka:beginEx()
.typeId(zilla:id("kafka"))
.merged()
.capabilities("PRODUCE_ONLY")
.topic("test")
.ackMode("LEADER_ONLY")
.build()
.build()}

connected

write zilla:data.ext ${kafka:dataEx()
.typeId(zilla:id("kafka"))
.merged()
.produce()
.timestamp(newTimestamp)
.partition(0, 1)
.build()
.build()}
write ${kafka:randomBytes(1024)}
write flush

write zilla:data.ext ${kafka:dataEx()
.typeId(zilla:id("kafka"))
.merged()
.produce()
.timestamp(newTimestamp)
.partition(0, 2)
.build()
.build()}
write ${kafka:randomBytes(1024)}
write flush

write zilla:data.ext ${kafka:dataEx()
.typeId(zilla:id("kafka"))
.merged()
.produce()
.timestamp(newTimestamp)
.partition(0, 3)
.build()
.build()}
write ${kafka:randomBytes(1024)}
write flush

write zilla:data.ext ${kafka:dataEx()
.typeId(zilla:id("kafka"))
.merged()
.produce()
.timestamp(newTimestamp)
.partition(0, 4)
.build()
.build()}
write ${kafka:randomBytes(1024)}
write flush

write zilla:data.ext ${kafka:dataEx()
.typeId(zilla:id("kafka"))
.merged()
.produce()
.timestamp(newTimestamp)
.partition(0, 5)
.build()
.build()}
write ${kafka:randomBytes(1024)}
write flush

write zilla:data.ext ${kafka:dataEx()
.typeId(zilla:id("kafka"))
.merged()
.produce()
.timestamp(newTimestamp)
.partition(0, 6)
.build()
.build()}
write ${kafka:randomBytes(1024)}
write flush

write zilla:data.ext ${kafka:dataEx()
.typeId(zilla:id("kafka"))
.merged()
.produce()
.timestamp(newTimestamp)
.partition(0, 7)
.build()
.build()}
write ${kafka:randomBytes(1024)}
write flush

write zilla:data.ext ${kafka:dataEx()
.typeId(zilla:id("kafka"))
.merged()
.produce()
.timestamp(newTimestamp)
.partition(0, 8)
.build()
.build()}
write ${kafka:randomBytes(1024)}
write flush

write zilla:data.ext ${kafka:dataEx()
.typeId(zilla:id("kafka"))
.merged()
.produce()
.timestamp(newTimestamp)
.partition(0, 9)
.build()
.build()}
write ${kafka:randomBytes(1024)}
write flush

write zilla:data.ext ${kafka:dataEx()
.typeId(zilla:id("kafka"))
.merged()
.produce()
.timestamp(newTimestamp)
.partition(0, 10)
.build()
.build()}
write ${kafka:randomBytes(1024)}
write flush

write zilla:data.ext ${kafka:dataEx()
.typeId(zilla:id("kafka"))
.merged()
.produce()
.timestamp(newTimestamp)
.partition(0, 11)
.build()
.build()}
write "Hello, world #A11"
write flush
134 changes: 134 additions & 0 deletions .../kafka/streams/application/merged/merged.produce.message.values.partition.idle/server.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,134 @@
#
# 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}

accept "zilla://streams/app0"
option zilla:window 8192
option zilla:transmission "half-duplex"

accepted

read zilla:begin.ext ${kafka:beginEx()
.typeId(zilla:id("kafka"))
.merged()
.capabilities("PRODUCE_ONLY")
.topic("test")
.ackMode("LEADER_ONLY")
.build()
.build()}

connected

read zilla:data.ext ${kafka:matchDataEx()
.typeId(zilla:id("kafka"))
.merged()
.produce()
.partition(0, 1)
.build()
.build()}
read [0..1024]

read zilla:data.ext ${kafka:matchDataEx()
.typeId(zilla:id("kafka"))
.merged()
.produce()
.partition(0, 2)
.build()
.build()}
read [0..1024]

read zilla:data.ext ${kafka:matchDataEx()
.typeId(zilla:id("kafka"))
.merged()
.produce()
.partition(0, 3)
.build()
.build()}
read [0..1024]

read zilla:data.ext ${kafka:matchDataEx()
.typeId(zilla:id("kafka"))
.merged()
.produce()
.partition(0, 4)
.build()
.build()}
read [0..1024]

read zilla:data.ext ${kafka:matchDataEx()
.typeId(zilla:id("kafka"))
.merged()
.produce()
.partition(0, 5)
.build()
.build()}
read [0..1024]

read zilla:data.ext ${kafka:matchDataEx()
.typeId(zilla:id("kafka"))
.merged()
.produce()
.partition(0, 6)
.build()
.build()}
read [0..1024]

read zilla:data.ext ${kafka:matchDataEx()
.typeId(zilla:id("kafka"))
.merged()
.produce()
.partition(0, 7)
.build()
.build()}
read [0..1024]

read zilla:data.ext ${kafka:matchDataEx()
.typeId(zilla:id("kafka"))
.merged()
.produce()
.partition(0, 8)
.build()
.build()}
read [0..1024]

read zilla:data.ext ${kafka:matchDataEx()
.typeId(zilla:id("kafka"))
.merged()
.produce()
.partition(0, 9)
.build()
.build()}
read [0..1024]

read zilla:data.ext ${kafka:matchDataEx()
.typeId(zilla:id("kafka"))
.merged()
.produce()
.partition(0, 10)
.build()
.build()}
read [0..1024]

read zilla:data.ext ${kafka:matchDataEx()
.typeId(zilla:id("kafka"))
.merged()
.produce()
.partition(0, 11)
.build()
.build()}
read "Hello, world #A11"
Loading
Loading