Skip to content

Mqtt-kafka single group support cont #731

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 12 commits into from
Jan 17, 2024
Merged

Mqtt-kafka single group support cont #731

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
0ca1a1e
Kafka single group support continuetion
akrambek Jan 11, 2024
ee72e10
Add more spec
akrambek Jan 13, 2024
f165cd1
Return list of partitions as reply merge begin ex for produce_only as…
akrambek Jan 13, 2024
52b778d
Fix checkstyle
akrambek Jan 13, 2024
5238097
Fix sequence assignment
akrambek Jan 13, 2024
196b0ba
Applied feedback from PR
akrambek Jan 14, 2024
0808b40
Add produce flush support
akrambek Jan 14, 2024
66988b8
Fix kafkafunction
akrambek Jan 14, 2024
89f2857
Add support for replay
akrambek Jan 16, 2024
276ca31
Fix script
akrambek Jan 16, 2024
f9292af
Fix script
akrambek Jan 16, 2024
13a19ff
Fix race in the script
akrambek Jan 16, 2024
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
Prev Previous commit
Next Next commit
Return list of partitions as reply merge begin ex for produce_only as… 
… well
  • Loading branch information
@akrambek
akrambek committed Jan 13, 2024
commit f165cd170eb7add2304e33c31de6a2eb5f92fcbb
25 changes: 20 additions & 5 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 @@ -1592,28 +1592,34 @@ private void doMergedReplyBegin(
if (capabilities == FETCH_ONLY)
{
doBegin(sender, originId, routedId, replyId, replySeq, replyAck, replyMax,
traceId, authorization, affinity, beginExToKafka());
traceId, authorization, affinity, beginExToKafka(beginExToKafkaMergedFetchOnly()));
}
else if (capabilities == PRODUCE_ONLY)
{
doBegin(sender, originId, routedId, replyId, replySeq, replyAck, replyMax,
traceId, authorization, affinity, beginExToKafka(beginExToKafkaMergedProduceOnly()));
}
else
{
doBegin(sender, originId, routedId, replyId, replySeq, replyAck, replyMax,
traceId, authorization, affinity, EMPTY_EXTENSION);
traceId, authorization, affinity, EMPTY_EXTENSION);
}

doUnmergedFetchReplyWindowsIfNecessary(traceId);
}

private Flyweight.Builder.Visitor beginExToKafka()
private Flyweight.Builder.Visitor beginExToKafka(
Consumer<KafkaMergedBeginExFW.Builder> beginExToKafkaMerged)
{
return (buffer, offset, maxLimit) ->
kafkaBeginExRW.wrap(buffer, offset, maxLimit)
.typeId(kafkaTypeId)
.merged(beginExToKafkaMerged())
.merged(beginExToKafkaMerged)
.build()
.limit() - offset;
}

private Consumer<KafkaMergedBeginExFW.Builder> beginExToKafkaMerged()
private Consumer<KafkaMergedBeginExFW.Builder> beginExToKafkaMergedFetchOnly()
{
return builder ->
{
Expand All @@ -1640,6 +1646,15 @@ private Consumer<KafkaMergedBeginExFW.Builder> beginExToKafkaMerged()
};
}

private Consumer<KafkaMergedBeginExFW.Builder> beginExToKafkaMergedProduceOnly()
{
return builder ->
{
builder.capabilities(c -> c.set(PRODUCE_ONLY)).topic(topic);
leadersByPartitionId.intForEach((k, v) -> builder.partitionsItem(i -> i.partitionId(k)));
};
}

private void doMergedReplyData(
long traceId,
int flags,
Expand Down
10 changes: 10 additions & 0 deletions ...cs/binding/kafka/streams/application/merged/merged.group.produce.message.value/client.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,16 @@ write zilla:begin.ext ${kafka:beginEx()

connected

read zilla:begin.ext ${kafka:matchBeginEx()
.typeId(zilla:id("kafka"))
.merged()
.capabilities("PRODUCE_ONLY")
.topic("test")
.partition(0, -1)
.partition(1, -1)
.build()
.build()}

write zilla:data.ext ${kafka:dataEx()
.typeId(zilla:id("kafka"))
.merged()
Expand Down
11 changes: 11 additions & 0 deletions ...cs/binding/kafka/streams/application/merged/merged.group.produce.message.value/server.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,17 @@ read zilla:begin.ext ${kafka:beginEx()

connected

write zilla:begin.ext ${kafka:beginEx()
.typeId(zilla:id("kafka"))
.merged()
.capabilities("PRODUCE_ONLY")
.topic("test")
.partition(0, -1)
.partition(1, -1)
.build()
.build()}
write flush

read zilla:data.ext ${kafka:dataEx()
.typeId(zilla:id("kafka"))
.merged()
Expand Down