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
start of idempotent work
  • Loading branch information
@bmaidics
bmaidics committed Jan 13, 2024
commit f831ff80c8d5c1a272c25acf27f60deee2a64a44
343 changes: 338 additions & 5 deletions ...io/aklivity/zilla/runtime/binding/mqtt/kafka/internal/stream/MqttKafkaSessionFactory.java
View file
Open in desktop

Large diffs are not rendered by default.

3 changes: 3 additions & 0 deletions ...c/main/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/MqttServerFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -476,6 +476,7 @@ public final class MqttServerFactory implements MqttStreamFactory
private final Supplier<String16FW> supplyClientId;
private final MqttValidator validator;
private final CharsetDecoder utf8Decoder;
//TODO: remove
private final ConcurrentMap<String, IntArrayList> unreleasedPacketIdsByClientId;

private Map<String, Validator> validators;
Expand Down Expand Up @@ -2466,9 +2467,11 @@ private final class MqttServer
private int decodableRemainingBytes;
private final Int2ObjectHashMap<MqttSubscribeStream> qos1Subscribes;
private final Int2ObjectHashMap<MqttSubscribeStream> qos2Subscribes;
//TODO Long2LongHashMap
private final LinkedHashMap<Long, Integer> unAckedReceivedQos1PacketIds;
private final LinkedHashMap<Long, Integer> unAckedReceivedQos2PacketIds;

//TODO this will come from session reply begin
private IntArrayList unreleasedPacketIds;

private int version = MQTT_PROTOCOL_VERSION_5;
Expand Down
46 changes: 46 additions & 0 deletions ...fka.spec/src/main/java/io/aklivity/zilla/specs/binding/kafka/internal/KafkaFunctions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2545,6 +2545,18 @@ private KafkaOffsetFetchDataExBuilder()
offsetFetchDataExRW.wrap(writeBuffer, KafkaDataExFW.FIELD_OFFSET_OFFSET_FETCH, writeBuffer.capacity());
}

public KafkaOffsetFetchDataExBuilder partition(
int partitionId,
long partitionOffset,
int leaderEpoch)
{
offsetFetchDataExRW.partitionsItem(o -> o
.partitionId(partitionId)
.partitionOffset(partitionOffset)
.leaderEpoch(leaderEpoch));
return this;
}

public KafkaOffsetFetchDataExBuilder partition(
int partitionId,
long partitionOffset,
Expand Down Expand Up @@ -4026,6 +4038,8 @@ public final class KafkaMergedProduceDataExMatcherBuilder
private Integer deferred;
private Long timestamp;
private Long filters;
private Long producerId;
private Short producerEpoch;
private KafkaOffsetFW.Builder partitionRW;
private Array32FW.Builder<KafkaOffsetFW.Builder, KafkaOffsetFW> progressRW;
private KafkaDeltaFW.Builder deltaRW;
Expand Down Expand Up @@ -4144,6 +4158,24 @@ public KafkaMergedProduceDataExMatcherBuilder hashKey(
return this;
}

public KafkaMergedProduceDataExMatcherBuilder producerId(
long producerId)
{
assert this.producerId == null;
this.producerId = producerId;

return this;
}

public KafkaMergedProduceDataExMatcherBuilder producerEpoch(
short producerEpoch)
{
assert this.producerEpoch == null;
this.producerEpoch = producerEpoch;

return this;
}

public KafkaMergedProduceDataExMatcherBuilder delta(
String delta,
long ancestorOffset)
Expand Down Expand Up @@ -4319,6 +4351,8 @@ private boolean match(
matchDeferred(produce) &&
matchTimestamp(produce) &&
matchKey(produce) &&
matchProducerId(produce) &&
matchProducerEpoch(produce) &&
matchHashKey(produce) &&
matchHeaders(produce);
}
Expand Down Expand Up @@ -4353,6 +4387,18 @@ private boolean matchHashKey(
return hashKeyRW == null || hashKeyRW.build().equals(mergedProduceDataEx.hashKey());
}

private boolean matchProducerId(
final KafkaMergedProduceDataExFW mergedProduceDataEx)
{
return producerId == null || producerId.equals(mergedProduceDataEx.producerId());
}

private boolean matchProducerEpoch(
final KafkaMergedProduceDataExFW mergedProduceDataEx)
{
return producerEpoch == null || producerEpoch.equals(mergedProduceDataEx.producerEpoch());
}

private boolean matchHeaders(
final KafkaMergedProduceDataExFW mergedProduceDataEx)
{
Expand Down
292 changes: 292 additions & 0 deletions .../scripts/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/kafka/publish.qos2/client.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,298 @@ connect "zilla://streams/kafka0"
option zilla:window 8192
option zilla:transmission "duplex"

write zilla:begin.ext ${kafka:beginEx()
.typeId(zilla:id("kafka"))
.merged()
.capabilities("PRODUCE_AND_FETCH")
.topic("mqtt-sessions")
.groupId("mqtt-clients")
.filter()
.key("client#migrate")
.headerNot("sender-id", "sender-1")
.build()
.build()
.build()}

connected

write zilla:data.ext ${kafka:dataEx()
.typeId(zilla:id("kafka"))
.merged()
.produce()
.deferred(0)
.partition(-1, -1)
.key("client#migrate")
.hashKey("client")
.header("sender-id", "sender-1")
.build()
.build()}
write zilla:data.empty
write flush
write notify SENT_MIGRATE_SIGNAL


connect await SENT_MIGRATE_SIGNAL
"zilla://streams/kafka0"
option zilla:window 8192
option zilla:transmission "duplex"

write zilla:begin.ext ${kafka:beginEx()
.typeId(zilla:id("kafka"))
.group()
.groupId("client-session")
.protocol("highlander")
.timeout(1000)
.build()
.build()}

read zilla:begin.ext ${kafka:matchBeginEx()
.typeId(zilla:id("kafka"))
.group()
.groupId("client-session")
.protocol("highlander")
.instanceId("zilla")
.host("localhost")
.port(9092)
.timeout(1000)
.build()
.build()}

connected

read advised zilla:flush ${kafka:matchFlushEx()
.typeId(zilla:id("kafka"))
.group()
.leaderId("consumer-1")
.memberId("consumer-1")
.members("consumer-1")
.build()
.build()}
read notify RECEIVED_LEADER

write zilla:data.empty
write flush


connect await RECEIVED_LEADER
"zilla://streams/kafka0"
option zilla:window 8192
option zilla:transmission "duplex"

write zilla:begin.ext ${kafka:beginEx()
.typeId(zilla:id("kafka"))
.offsetFetch()
.groupId("client")
.host("localhost")
.port(9092)
.topic("mqtt-messages")
.partition(0)
.build()
.build()}
connected

read zilla:data.ext ${kafka:dataEx()
.typeId(zilla:id("kafka"))
.offsetFetch()
.partition(0, 1, 0)
.build()
.build()}
read zilla:data.empty
read notify RECEIVED_INITIAL_OFFSET

write close
read closed


connect await RECEIVED_INITIAL_OFFSET
"zilla://streams/kafka0"
option zilla:window 8192
option zilla:transmission "duplex"

write zilla:begin.ext ${kafka:beginEx()
.typeId(zilla:id("kafka"))
.initProducerId()
.producerId(0)
.producerEpoch(0)
.build()
.build()}

read zilla:begin.ext ${kafka:beginEx()
.typeId(zilla:id("kafka"))
.initProducerId()
.producerId(1)
.producerEpoch(1)
.build()
.build()}
read notify RECEIVED_PRODUCER

connected


connect await RECEIVED_PRODUCER
"zilla://streams/kafka0"
option zilla:window 8192
option zilla:transmission "duplex"

write zilla:begin.ext ${kafka:beginEx()
.typeId(zilla:id("kafka"))
.offsetCommit()
.groupId("client")
.memberId("consumer-1")
.instanceId("zilla")
.build()
.build()}

connected

write zilla:data.ext ${kafka:dataEx()
.typeId(zilla:id("kafka"))
.offsetCommit()
.topic("mqtt-messages")
.progress(0, 1, mqtt:metadata()
.metadata(1, 1)
.build())
.generationId(0)
.leaderEpoch(0)
.build()
.build()}
write zilla:data.empty
write notify SENT_INITIAL_OFFSET_COMMIT

write zilla:data.ext ${kafka:dataEx()
.typeId(zilla:id("kafka"))
.offsetCommit()
.topic("mqtt-messages")
.progress(0, 2, mqtt:metadata()
.metadata(1, 1, 1)
.build())
.generationId(0)
.leaderEpoch(0)
.build()
.build()}
write zilla:data.empty
write flush

read zilla:data.ext ${kafka:dataEx()
.typeId(zilla:id("kafka"))
.offsetCommit()
.topic("mqtt-messages")
.progress(0, 2, mqtt:metadata()
.metadata(1, 1, 1)
.build())
.generationId(0)
.leaderEpoch(0)
.build()
.build()}
read zilla:data.empty

write zilla:data.ext ${kafka:dataEx()
.typeId(zilla:id("kafka"))
.offsetCommit()
.topic("mqtt-messages")
.progress(0, 2, mqtt:metadata()
.metadata(1, 1)
.build())
.generationId(0)
.leaderEpoch(0)
.build()
.build()}
write zilla:data.empty
write flush

read zilla:data.ext ${kafka:dataEx()
.typeId(zilla:id("kafka"))
.offsetCommit()
.topic("mqtt-messages")
.progress(0, 2, mqtt:metadata()
.metadata(1, 1)
.build())
.generationId(0)
.leaderEpoch(0)
.build()
.build()}
read zilla:data.empty


connect await SENT_INITIAL_OFFSET_COMMIT
"zilla://streams/kafka0"
option zilla:window 8192
option zilla:transmission "duplex"

write zilla:begin.ext ${kafka:beginEx()
.typeId(zilla:id("kafka"))
.merged()
.capabilities("PRODUCE_AND_FETCH")
.topic("mqtt-sessions")
.groupId("mqtt-clients")
.filter()
.key("client")
.build()
.filter()
.key("client#migrate")
.headerNot("sender-id", "sender-1")
.build()
.build()
.build()}

connected

write zilla:data.ext ${kafka:dataEx()
.typeId(zilla:id("kafka"))
.merged()
.produce()
.deferred(0)
.partition(-1, -1)
.key("client#will-signal")
.hashKey("client")
.header("type", "will-signal")
.build()
.build()}
write flush

write zilla:data.ext ${kafka:dataEx()
.typeId(zilla:id("kafka"))
.merged()
.produce()
.deferred(0)
.partition(-1, -1)
.key("client#expiry-signal")
.hashKey("client")
.header("type", "expiry-signal")
.build()
.build()}
write flush

write zilla:data.ext ${kafka:dataEx()
.typeId(zilla:id("kafka"))
.merged()
.produce()
.deferred(0)
.partition(-1, -1)
.key("client#expiry-signal")
.hashKey("client")
.header("type", "expiry-signal")
.build()
.build()}
write ${mqtt:sessionSignal()
.expiry()
.instanceId("zilla-1")
.clientId("client")
.delay(1000)
.expireAt(-1)
.build()
.build()}
write flush

read advised zilla:flush
read notify RECEIVED_SESSION_STATE


connect await RECEIVED_SESSION_STATE
"zilla://streams/kafka0"
option zilla:window 8192
option zilla:transmission "duplex"

write zilla:begin.ext ${kafka:beginEx()
.typeId(zilla:id("kafka"))
.merged()
Expand Down
Loading