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

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
Fix sequence assignment
  • Loading branch information
@akrambek
akrambek committed Jan 13, 2024
commit 523809723aec2414ac1ec7959ddffae9a8faa78b
7 changes: 6 additions & 1 deletion ...va/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaClientProduceFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -551,9 +551,14 @@ private int flushRecordInit(
client.doEncodeRequestIfNecessary(traceId, budgetId);
}

if (client.producerId == RECORD_BATCH_PRODUCER_ID_NONE ||
client.producerId != producerId)

@jfallows jfallows Jan 14, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should this be && instead?

Please remove extra whitespace around !=.

{
client.sequence = sequence;

@jfallows jfallows Jan 14, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggest we rename to client.baseSequence.

}

client.producerId = producerId;
client.producerEpoch = producerEpoch;
client.sequence = sequence;

client.doEncodeRecordInit(traceId, timestamp, ackMode, key, payload, headers);
if (client.encodeSlot != NO_SLOT)
Expand Down