Skip to content

Qos2 idempotent producer #733

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 44 commits into from
Feb 6, 2024
Merged

Qos2 idempotent producer #733

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
44 commits
Select commit Hold shift + click to select a range
901f0b2
separating publish streams based on qos
bmaidics Jan 10, 2024
79ddafd
Fix test
bmaidics Jan 11, 2024
1265671
Checkpoint
bmaidics Jan 9, 2024
7847580
Checkpoint
bmaidics Jan 9, 2024
7895cd5
check
bmaidics Jan 9, 2024
f831ff8
start of idempotent work
bmaidics Jan 13, 2024
412019a
Merge remote-tracking branch 'upstream/feature/mqtt-kafka' into featu…
bmaidics Jan 15, 2024
167a33a
Optimize memory allocation for mqtt-kafka offset tracking (#694)
bmaidics Jan 15, 2024
cf08e0f
checkpoint
bmaidics Jan 15, 2024
80c0e09
checkpoint
bmaidics Jan 18, 2024
62db2cb
Merge remote-tracking branch 'upstream/feature/mqtt-kafka' into qos2_…
bmaidics Jan 18, 2024
49a7b35
Checkpoint with retained offsetCommit stream
bmaidics Jan 18, 2024
92b262a
checkpoint
bmaidics Jan 19, 2024
29d24f7
checkpoint
bmaidics Jan 22, 2024
f9fe624
mqtt-kafka checkpoint
bmaidics Jan 23, 2024
5195271
checkpoint
bmaidics Jan 24, 2024
99f1c1f
checkpoint
bmaidics Jan 24, 2024
8db664d
Fixes
bmaidics Jan 24, 2024
7038c04
Fix flaky test
bmaidics Jan 24, 2024
e00dd8a
Merge remote-tracking branch 'upstream/feature/mqtt-kafka' into qos2_…
bmaidics Jan 25, 2024
e3d92ee
fixes
bmaidics Jan 25, 2024
31e26ed
fix dump
attilakreiner Jan 25, 2024
f4b27c1
Fix init produce id request
akrambek Jan 25, 2024
9c5fd24
fix
bmaidics Jan 26, 2024
35c6976
Fix bug
bmaidics Jan 26, 2024
a3c7e09
Fix
bmaidics Jan 29, 2024
5cf9068
Don't flush early if the sequence number is not set
akrambek Jan 29, 2024
bc38ac7
Draft
bmaidics Jan 31, 2024
724bd70
checkpoint
bmaidics Feb 1, 2024
7df5123
Fixes
bmaidics Feb 1, 2024
4204bc1
Adrress review comments
bmaidics Feb 1, 2024
2a0204f
Merge remote-tracking branch 'upstream/feature/mqtt-kafka' into qos2_…
bmaidics Feb 1, 2024
ebd9cee
Merge fixes
bmaidics Feb 1, 2024
8988e6b
Include producerId and producerEpoch into cache entry
akrambek Feb 2, 2024
739f79d
fix dump
attilakreiner Feb 2, 2024
878ec78
Fix qos2 large message
bmaidics Feb 3, 2024
0563e13
Fix typo
akrambek Feb 5, 2024
7f2ca2a
Merge branch 'qos2_idempontent' of github.com:bmaidics/zilla into qos…
akrambek Feb 5, 2024
7d9afee
reviews
bmaidics Feb 5, 2024
c274636
more feedback
bmaidics Feb 5, 2024
215791e
checkpoint
bmaidics Feb 6, 2024
d3635b3
Refactor
bmaidics Feb 6, 2024
007e8d1
Adjust code coverage ratio
jfallows Feb 6, 2024
ccae27c
Ignore IT that fails only on GitHub Actions, see issue #786
jfallows Feb 6, 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
Merge fixes
  • Loading branch information
@bmaidics
bmaidics committed Feb 1, 2024
commit ebd9cee7624204ccc732fbd70c705d49aa98362c
4 changes: 2 additions & 2 deletions ...st/java/io/aklivity/zilla/runtime/command/dump/internal/airline/ZillaDumpCommandTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1494,7 +1494,7 @@ public void generateStreamsBuffer() throws Exception
.session()
.flags("CLEAN_START")
.expiry(42)
.qosMax(2)
.subscribeQosMax(2)
.packetSizeMax(42_000)
.capabilities("RETAIN")
.clientId("client-id")
Expand All @@ -1519,7 +1519,7 @@ public void generateStreamsBuffer() throws Exception
.session()
.flags("CLEAN_START", "WILL")
.expiry(42)
.qosMax(2)
.subscribeQosMax(2)
.packetSizeMax(42_000)
.capabilities("RETAIN", "WILDCARD", "SUBSCRIPTION_IDS", "SHARED_SUBSCRIPTIONS")
.clientId("client-id")
Expand Down
Binary file modified ...p/src/test/resources/io/aklivity/zilla/runtime/command/dump/internal/airline/engine/data0
View file
Open in desktop
Binary file not shown.
6 changes: 0 additions & 6 deletions .../aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaCacheClientProduceFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -696,12 +696,6 @@ private void onClientInitialData(
final int valueLength = valueFragment != null ? valueFragment.sizeof() + deferred : -1;
final int maxValueLength = valueLength + headersSizeMax;

// if ((flags & FLAGS_FIN) == 0x00 && deferred == 0)
// {
// error = ERROR_CORRUPT_MESSAGE;
// break init;
// }

if (maxValueLength > partition.segmentBytes())
{
error = ERROR_RECORD_LIST_TOO_LARGE;
Expand Down
1 change: 1 addition & 0 deletions ...ts/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/mqtt/publish.mixture.qos/client.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ write zilla:begin.ext ${mqtt:beginEx()
.flags("CLEAN_START")
.expiry(1)
.publishQosMax(2)
.capabilities("REDIRECT")
.clientId("client")
.build()
.build()}
Expand Down
1 change: 1 addition & 0 deletions ...ts/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/mqtt/publish.mixture.qos/server.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ read zilla:begin.ext ${mqtt:matchBeginEx()
.flags("CLEAN_START")
.expiry(1)
.publishQosMax(2)
.capabilities("REDIRECT")
.clientId("client")
.build()
.build()}
Expand Down
1 change: 1 addition & 0 deletions ...pts/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/mqtt/publish.qos2.abort/client.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ write zilla:begin.ext ${mqtt:beginEx()
.flags("CLEAN_START")
.expiry(1)
.publishQosMax(2)
.capabilities("REDIRECT")
.clientId("client")
.build()
.build()}
Expand Down
1 change: 1 addition & 0 deletions .../specs/binding/mqtt/kafka/streams/mqtt/publish.qos2.offset.commit.abort.phase1/client.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ write zilla:begin.ext ${mqtt:beginEx()
.flags("CLEAN_START")
.expiry(1)
.publishQosMax(2)
.capabilities("REDIRECT")
.clientId("client")
.build()
.build()}
Expand Down
1 change: 1 addition & 0 deletions .../specs/binding/mqtt/kafka/streams/mqtt/publish.qos2.offset.commit.abort.phase1/server.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ read zilla:begin.ext ${mqtt:beginEx()
.flags("CLEAN_START")
.expiry(1)
.publishQosMax(2)
.capabilities("REDIRECT")
.clientId("client")
.build()
.build()}
Expand Down
1 change: 1 addition & 0 deletions .../specs/binding/mqtt/kafka/streams/mqtt/publish.qos2.offset.commit.abort.phase2/client.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ write zilla:begin.ext ${mqtt:beginEx()
.flags("CLEAN_START")
.expiry(1)
.publishQosMax(2)
.capabilities("REDIRECT")
.clientId("client")
.build()
.build()}
Expand Down
1 change: 1 addition & 0 deletions .../specs/binding/mqtt/kafka/streams/mqtt/publish.qos2.offset.commit.abort.phase2/server.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ read zilla:begin.ext ${mqtt:beginEx()
.flags("CLEAN_START")
.expiry(1)
.publishQosMax(2)
.capabilities("REDIRECT")
.clientId("client")
.build()
.build()}
Expand Down
1 change: 1 addition & 0 deletions ...s/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/mqtt/publish.qos2.recover/client.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ write zilla:begin.ext ${mqtt:beginEx()
.flags("CLEAN_START")
.expiry(1)
.publishQosMax(2)
.capabilities("REDIRECT")
.clientId("client")
.build()
.build()}
Expand Down
1 change: 1 addition & 0 deletions ...s/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/mqtt/publish.qos2.recover/server.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ read zilla:begin.ext ${mqtt:matchBeginEx()
.flags("CLEAN_START")
.expiry(1)
.publishQosMax(2)
.capabilities("REDIRECT")
.clientId("client")
.build()
.build()}
Expand Down
1 change: 1 addition & 0 deletions .../io/aklivity/zilla/specs/binding/mqtt/kafka/streams/mqtt/publish.qos2.retained/client.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ write zilla:begin.ext ${mqtt:beginEx()
.flags("CLEAN_START")
.expiry(1)
.publishQosMax(2)
.capabilities("REDIRECT")
.clientId("client")
.build()
.build()}
Expand Down
1 change: 1 addition & 0 deletions .../io/aklivity/zilla/specs/binding/mqtt/kafka/streams/mqtt/publish.qos2.retained/server.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ read zilla:begin.ext ${mqtt:matchBeginEx()
.flags("CLEAN_START")
.expiry(1)
.publishQosMax(2)
.capabilities("REDIRECT")
.clientId("client")
.build()
.build()}
Expand Down
1 change: 1 addition & 0 deletions ...n/scripts/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/mqtt/publish.qos2/client.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ write zilla:begin.ext ${mqtt:beginEx()
.flags("CLEAN_START")
.expiry(1)
.publishQosMax(2)
.capabilities("REDIRECT")
.clientId("client")
.build()
.build()}
Expand Down
1 change: 1 addition & 0 deletions ...n/scripts/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/mqtt/publish.qos2/server.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ read zilla:begin.ext ${mqtt:matchBeginEx()
.flags("CLEAN_START")
.expiry(1)
.publishQosMax(2)
.capabilities("REDIRECT")
.clientId("client")
.build()
.build()}
Expand Down
2 changes: 1 addition & 1 deletion .../zilla/specs/binding/mqtt/streams/application/session.connect.redirect.support/client.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@ read zilla:begin.ext ${mqtt:matchBeginEx()
.typeId(zilla:id("mqtt"))
.session()
.flags("CLEAN_START")
.qosMax(2)
.subscribeQosMax(2)
.packetSizeMax(66560)
.capabilities("RETAIN", "WILDCARD", "SUBSCRIPTION_IDS", "SHARED_SUBSCRIPTIONS", "REDIRECT")
.clientId("client")
Expand Down
2 changes: 1 addition & 1 deletion .../zilla/specs/binding/mqtt/streams/application/session.connect.redirect.support/server.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ write zilla:begin.ext ${mqtt:beginEx()
.typeId(zilla:id("mqtt"))
.session()
.flags("CLEAN_START")
.qosMax(2)
.subscribeQosMax(2)
.packetSizeMax(66560)
.capabilities("RETAIN", "WILDCARD", "SUBSCRIPTION_IDS", "SHARED_SUBSCRIPTIONS", "REDIRECT")
.clientId("client")
Expand Down