Consumer group message acknowledgement support by akrambek · Pull Request #538 · aklivity/zilla

akrambek · 2023-10-27T00:12:54Z

Description

This feature lets other kafka mapping ability to start fetching from the last observed message offset

Fixes #588

…artial data frame while computing crc32c value

jfallows · 2023-10-27T05:03:58Z

specs/binding-kafka.spec/src/main/resources/META-INF/zilla/kafka.idl

+            KafkaHeader[] headers;      // INIT + FIN (produce), INIT only (fetch)
+        }
+
+        struct KafkaMergedConsumerDataExKafkaMergedConsumerDataEx


jfallows · 2023-10-27T05:05:37Z

specs/binding-kafka.spec/src/main/resources/META-INF/zilla/kafka.idl

        union KafkaMergedFlushEx switch (uint8)
        {
-            case 252: kafka::stream::KafkaMergedConsumerFlushEx consumer;
+            case 8: kafka::stream::KafkaMergedOffsetCommitFlushEx offsetCommit;


This should stay as KafkaMergedConsumerFlushEx for consistency with KafkaMergedConsumerDataEx, right?

jfallows · 2023-10-27T05:06:23Z

specs/binding-kafka.spec/src/main/resources/META-INF/zilla/kafka.idl

-        struct KafkaConsumerDataEx
+        union KafkaConsumerDataEx switch (uint8)
+        {
+            case 253: kafka::stream::KafkaMergedConsumerDataEx group;


Merged should not be on Consumer, right?

jfallows · 2023-10-27T05:07:27Z

specs/binding-kafka.spec/src/main/resources/META-INF/zilla/kafka.idl


        struct KafkaOffsetCommitDataEx
        {
+            string16 topic;


The idea is to have one offset commit stream per group instead of one per topic?

runtime/binding-kafka/src/main/zilla/protocol.idl

jfallows · 2023-11-14T02:21:37Z

specs/binding-kafka.spec/src/main/resources/META-INF/zilla/kafka.idl

            int64 timestamp = 0;        // INIT only
-            int64 filters = -1;         // INIT only
            KafkaOffset partition;      // INIT only
            KafkaOffset[] progress;     // INIT only


Is progress used by produce?

jfallows · 2023-11-14T02:22:48Z

specs/binding-kafka.spec/src/main/resources/META-INF/zilla/kafka.idl

        {
-            int32 partitionId;
-            int64 partitionOffset;
+            KafkaOffset progress;


progress or partition? (actual question)

jfallows · 2023-11-14T02:23:19Z

specs/binding-kafka.spec/src/main/resources/META-INF/zilla/kafka.idl

+        struct KafkaConsumerOffsetCommitDataEx
+        {
+            KafkaOffset progress;
+            int32 committedLeaderEpoch;


Perhaps just leaderEpoch as we are in the context of Commit already here?

jfallows · 2023-11-14T02:23:24Z

specs/binding-kafka.spec/src/main/resources/META-INF/zilla/kafka.idl

-            int32 partitionId;
-            int64 partitionOffset;
+            KafkaOffset progress;
+            int32 committedLeaderEpoch;


Perhaps just leaderEpoch as we are in the context of Commit already here?

jfallows · 2023-11-14T02:29:27Z

specs/binding-kafka.spec/src/main/resources/META-INF/zilla/kafka.idl

        {
            case 253: kafka::stream::KafkaGroupFlushEx group;
            case 255: kafka::stream::KafkaMergedFlushEx merged;
+            case 8: kafka::stream::KafkaOffsetCommitFlushEx offsetCommit;


Perhaps offset is sufficient here instead of offsetCommit?

jfallows · 2023-11-14T02:29:59Z

specs/binding-kafka.spec/src/main/resources/META-INF/zilla/kafka.idl

+        union KafkaConsumerDataEx switch (uint8)
+        {
+            case 253: kafka::stream::KafkaConsumerGroupDataEx group;
+            case 8: kafka::stream::KafkaConsumerOffsetCommitDataEx offsetCommit;


Perhaps offset is sufficient here instead of offsetCommit?

jfallows · 2023-11-14T02:31:04Z

...illa/specs/binding/kafka/streams/application/consumer/ackknowledge.message.offset/server.rpt

+read zilla:data.ext ${kafka:dataEx()
+                             .typeId(zilla:id("kafka"))
+                             .consumer()
+                               .offsetCommit()


I think offset would read better here, as in

consumer() .offset() ...

jfallows · 2023-11-14T02:33:14Z

...illa/specs/binding/kafka/streams/application/consumer/ackknowledge.message.offset/client.rpt

+write zilla:data.empty
+write flush
+
+read option zilla:ack 0


My concern here is that reserved is 0 when sending the empty DATA frame, so WINDOW ack does not need to elevate the initialAck, making it difficult to tell the difference before or after ack.

Do you think it's an issue?

jfallows · 2023-11-14T02:33:49Z

...ecs/binding/kafka/streams/application/consumer/commit.ackknowledge.message.offset/client.rpt

+                             .typeId(zilla:id("kafka"))
+                             .offsetCommit()
+                                 .progress(0, 2)
+                                 .committedLeaderEpoch(0)


.leaderEpoch(0) reads better here.

jfallows · 2023-11-26T23:49:56Z

specs/binding-kafka.spec/src/main/resources/META-INF/zilla/kafka.idl

        {
            string16 topic;
-            KafkaOffset[] offsets;
+            KafkaTopicPartitionOffset[] partitions;


Since KafkaOffsetFetchBeginEx now includes topic, then KafkaOffsetFetchDataEx is already topic specific and doesn't need topic again, right?

I think that means we should use KafkaTopicPartition[] partitions on KafkaOffsetFetchDataEx and remove KafkaOffsetFetchTopicOffsets from kafka.idl, agree?

jfallows · 2023-11-26T23:51:55Z

specs/binding-kafka.spec/src/main/resources/META-INF/zilla/kafka.idl

-            KafkaKey hashKey;           // INIT only
            KafkaDelta delta;           // INIT + FIN
-            KafkaHeader[] headers;      // INIT + FIN (produce), INIT only (fetch)
+            KafkaHeader[] headers;      // INIT only


I think the comment before might have been incorrect for fetch, since headers are at the end of a kafka message in the wire protocol, so merged directly over client fetch would send headers with FIN, whereas cache always sends headers with INIT, so this comment should probably be INIT + FIN.

jfallows · 2023-11-26T23:53:54Z

...command-log/src/main/java/io/aklivity/zilla/runtime/command/log/internal/LoggableStream.java

-                   format("[merged] (%d) %d %s %d %d %d",
-                           merged.deferred(), merged.timestamp(), asString(key.value()),
-                           partition.partitionId(), partition.partitionOffset(), partition.latestOffset()));
+            format("[merged] (%d) %d %s %d %d %d",


Perhaps [merged] [fetch] ...?

jfallows · 2023-11-26T23:54:07Z

...command-log/src/main/java/io/aklivity/zilla/runtime/command/log/internal/LoggableStream.java

+        final KafkaOffsetFW partition = produce.partition();
+
+        out.printf(verboseFormat, index, offset, timestamp,
+            format("[merged] (%d) %d %s %d %d %d",


Perhaps [merged] [produce] ...?

jfallows · 2023-11-27T00:19:54Z

...java/io/aklivity/zilla/runtime/binding/http/kafka/internal/stream/HttpKafkaProxyFactory.java

                    {
-                        final KafkaMergedDataExFW kafkaMergedDataEx = kafkaDataEx.merged();
-                        final int contentLength = payload.sizeof() + kafkaMergedDataEx.deferred();
+                        final KafkaMergedFetchDataExFW fetch = kafkaDataEx.merged().fetch();


Please rename fetch to kafkaMergedFetchDataEx.

jfallows · 2023-11-27T00:58:12Z

...n/java/io/aklivity/zilla/runtime/binding/sse/kafka/internal/stream/SseKafkaProxyFactory.java

+                        kafkaMergedDataEx.fetch().progress() : null;
+                    key = kafkaMergedDataEx != null ? kafkaMergedDataEx.fetch().key().value() : null;
+                    final Array32FW<KafkaHeaderFW> headers = kafkaMergedDataEx != null ?
+                        kafkaMergedDataEx.fetch().headers() : null;


Define kafkaMergedFetchDataEx to avoid calling kafkaMergedDataEx.fetch() repeatedly.

jfallows · 2023-11-27T01:05:38Z

...specs/binding/kafka/streams/network/group.sasl.f1.j5.s3.l3.h3.handshake.v1/leader/client.rpt

@@ -176,7 +176,7 @@ read 20                                 # size
     -1s                                # authentication bytes
     0L                                 # session lifetime

-write 119                               # size
+write 82                               # size


Comment has vertical misalignment
(also for other similar scripts with ${instanceId} replaced by "zilla")

jfallows · 2023-11-27T01:07:09Z

...c/src/test/java/io/aklivity/zilla/specs/binding/kafka/streams/application/OffsetFetchIT.java

+    @Specification({
+        "${app}/topic.offset.info/client",
+        "${app}/topic.offset.info/server"})
+    public void shouldFetchPartitionOffsetInfo() throws Exception


Test scenario name does not match script scenario name.

jfallows · 2023-11-27T01:08:30Z

....spec/src/test/java/io/aklivity/zilla/specs/binding/kafka/streams/network/OffsetFetchIT.java

+    @Specification({
+        "${net}/topic.offset.info/client",
+        "${net}/topic.offset.info/server"})
+    public void shouldFetchPartitionOffsetInfo() throws Exception


Please check test scenario name.

jfallows · 2023-11-27T01:09:06Z

...c/src/test/java/io/aklivity/zilla/specs/binding/kafka/streams/network/OffsetFetchSaslIT.java

+import org.kaazing.k3po.junit.annotation.Specification;
+import org.kaazing.k3po.junit.rules.K3poRule;
+
+public class OffsetFetchSaslIT


Please check test scenario names.
Also typo Sals should be Sasl, and Scarm should be Scram

jfallows · 2023-11-28T23:33:09Z

...command-log/src/main/java/io/aklivity/zilla/runtime/command/log/internal/LoggableStream.java

+
+

Suggested change

jfallows · 2023-11-28T23:33:44Z

...command-log/src/main/java/io/aklivity/zilla/runtime/command/log/internal/LoggableStream.java

+            format("%s: %s", asString(h.name()), asString(h.value()))));
+    }
+
+


Suggested change

jfallows · 2023-11-28T23:37:45Z

...io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaCacheClientConsumerFactory.java

+                traceId, authorization, initialBud, reserved, extension);
+        }
+
+


Suggested change

jfallows · 2023-11-28T23:39:26Z

...io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaCacheServerConsumerFactory.java

+            final KafkaBeginExFW groupBeginEx = beginEx.typeId() == kafkaTypeId ? extension.get(kafkaBeginExRO::wrap) : null;
+            final KafkaGroupBeginExFW kafkaGroupBeginEx = groupBeginEx != null ? groupBeginEx.group() : null;


Suggested change

final KafkaBeginExFW groupBeginEx = beginEx.typeId() == kafkaTypeId ? extension.get(kafkaBeginExRO::wrap) : null;

final KafkaGroupBeginExFW kafkaGroupBeginEx = groupBeginEx != null ? groupBeginEx.group() : null;

final KafkaBeginExFW kafkaBeginEx = beginEx.typeId() == kafkaTypeId ? extension.get(kafkaBeginExRO::wrap) : null;

final KafkaGroupBeginExFW kafkaGroupBeginEx = groupBeginEx != null ? kafkaBeginEx.group() : null;

jfallows · 2023-11-28T23:43:37Z

...a/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaClientOffsetFetchFactory.java

+                client.errorCode = errorCode;
+                client.decoder = decodeReject;
+            }
+


Suggest switch instead of if-else-if-else.

…ent support (aklivity#538)

akrambek added 7 commits October 25, 2023 13:35

Adjust padding to accommodate good enough headers and don't include p…

2cef977

…artial data frame while computing crc32c value

Merge branch 'develop' into feature/consumer-group-cont

d201582

Merge branch 'feature/consumer-group-cont' into develop

76bf9de

Kafka Idl changes for consumer group offset commit

cadffd1

Merge branch 'develop' into feature/consumer-group-cont

0c2f619

Revert change

63b1232

Revert back change

1965267

jfallows requested changes Oct 27, 2023

View reviewed changes

akrambek and others added 11 commits October 29, 2023 20:25

Merge branch 'aklivity:develop' into develop

29ae79c

Merge branch 'aklivity:develop' into develop

ec1b39e

Merge branch 'aklivity:develop' into develop

51a9f0e

Merge branch 'aklivity:develop' into develop

4394783

Merge branch 'aklivity:develop' into develop

e8696ce

Merge branch 'aklivity:develop' into develop

51c37b1

Merge branch 'aklivity:develop' into develop

5da5f04

Merge branch 'aklivity:develop' into develop

db1e17c

Merge branch 'aklivity:develop' into develop

40f73dc

Merge branch 'develop' into feature/consumer-group-cont

db571f8

Apply feedback from PR

ef9eae6

jfallows previously approved these changes Nov 6, 2023

View reviewed changes

WIP

aa3e3f4

akrambek dismissed jfallows’s stale review via aa3e3f4 November 7, 2023 00:20

akrambek added 3 commits November 6, 2023 18:46

WIP

e5d225a

WIP KafkaFunctionTest passing

27cc8b3

Add spec scripts

8884413

jfallows requested changes Nov 14, 2023

View reviewed changes

akrambek added 4 commits November 14, 2023 16:24

Update spec

d46e2d4

Fix typo

2594787

Fix typo

b705991

Update script

355eafc

akrambek added 5 commits November 17, 2023 18:33

WIP Mqtt-kafka

f8f5297

WIP

868a758

Increase coverage

e0f3100

WIP

955ece0

WIP Offset commit

2482bcd

akrambek changed the title ~~Kafka Idl changes for consumer group offset commit~~ Consumer group message acknowledgement Nov 21, 2023

akrambek and others added 10 commits November 21, 2023 13:27

WIP

ad110bc

WIP

88dca78

WIP

a7b06b2

WIP

ac4cfa2

WIP

d1d7c81

WIP

43a3f17

Merge branch 'aklivity:develop' into develop

d1a0492

Merge branch 'develop' into feature/consumer-group-cont

db9d4fd

WIP Fix known issues

20fb5f0

Fix script

09c98bd

akrambek changed the title ~~Consumer group message acknowledgement~~ Consumer group message acknowledgement support Nov 23, 2023

akrambek marked this pull request as ready for review November 23, 2023 06:19

jfallows requested changes Nov 27, 2023

View reviewed changes

akrambek added 6 commits November 27, 2023 13:54

Apply feedback from PR

78ed2bf

Add more spec test

de26ff4

Bug fixes

325a067

Revert back the change

8954290

Revert back the change

d634384

Include metadata in commit ack

82e1b6d

jfallows requested changes Nov 28, 2023

View reviewed changes

Apply feedback from PR

143fac8

jfallows approved these changes Nov 28, 2023

View reviewed changes

jfallows merged commit c20c047 into aklivity:develop Nov 29, 2023

bmaidics pushed a commit to bmaidics/zilla that referenced this pull request Nov 30, 2023

REVERT BEFORE MERGE WITH UPSTREAM Consumer group message acknowledgem…

710682d

…ent support (aklivity#538)

		format("%s: %s", asString(h.name()), asString(h.value()))));
		}

		final KafkaBeginExFW groupBeginEx = beginEx.typeId() == kafkaTypeId ? extension.get(kafkaBeginExRO::wrap) : null;
		final KafkaGroupBeginExFW kafkaGroupBeginEx = groupBeginEx != null ? groupBeginEx.group() : null;

Conversation

akrambek commented Oct 27, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

akrambek commented Oct 27, 2023 •

edited

Loading