Support Kafka consumer groups #262

akrambek · 2023-05-26T14:58:36Z

Description

Please include a summary of the changes and the related issue. Please also include relevant motivation and context. List any dependencies that are required for this change.

Fixes #215

jfallows · 2023-05-29T15:56:02Z

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

@@ -321,6 +321,7 @@ scope kafka

        struct KafkaGroupBeginEx
        {
+            string16 topic;


Why do we need topic on KafkaGroupBeginEx?
Isn't the consumer group independent of any specific topic?

Forgot to remove it sorry.

akrambek · 2023-05-30T17:02:40Z

...in/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaGroupClientFactory.java

+            final long traceId = data.traceId();
+            final long budgetId = data.budgetId();
+
+            coordinatorClient.doSyncRequest(traceId, budgetId, data.payload());


This is what I am not sure. Whether to start from join request or from doSync

I think this is the case where the leader is making assignments to the members, right?

First, we need to verify and reject if not the leader, as Kafka will reject it anyway.
Then we need to doSyncRequest to update the assignments, but the leader is already in the group, so no need to proactively re-Join the group, i think, agree?

I think this is the case where the leader is making assignments to the members, right?

Yes

yeah that's what I was thinking

jfallows

We'll also need to add KafkaCacheServerGroupFactory and KafkaCacheClientGroupFactory so that this will work from the cache client side of the pipeline.

jfallows · 2023-05-30T17:56:10Z

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

+            {
+               string16 memberId;
+               uint32 length;
+               octets[length] topicPartition;


Given that the format of the assignment is opaque, suggest we call this value instead of topicPartition.

jfallows · 2023-05-30T17:58:35Z

.../aklivity/zilla/specs/binding/kafka/streams/network/consumer.group/full.handshake/client.rpt

+read 10                                              # size
+     ${newRequestId}
+     0                                               # Throttle time
+     0s                                              # No error


Note: all our other scripts use lower case for the comments, please follow the same lowercase pattern for comments in any new scripts you create and fix the comments in these scripts too for completeness when you get a moment.

jfallows · 2023-05-30T17:59:49Z

...a.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/schema/kafka.schema.patch.json

+                                        "groupId":
+                                        {
+                                            "title": "ConsumerGroupId",
+                                            "type": "string"


What is the scenario where we need to configure a route by groupId in zilla.yaml?

Yeah I think we don't really need it and it should be probably the options in mqtt-kafka binding

Sounds good, please remove it, including any associated config parsing and routing code.

jfallows · 2023-05-30T18:08:21Z

...in/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaGroupClientFactory.java

+            final long traceId = data.traceId();
+            final long budgetId = data.budgetId();
+
+            coordinatorClient.doSyncRequest(traceId, budgetId, data.payload());


I think this is the case where the leader is making assignments to the members, right?

First, we need to verify and reject if not the leader, as Kafka will reject it anyway.
Then we need to doSyncRequest to update the assignments, but the leader is already in the group, so no need to proactively re-Join the group, i think, agree?

jfallows · 2023-07-25T21:10:22Z

...kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/KafkaConfiguration.java

+        Configuration config)
+    {
+        return () -> String.format("%s-%s", "zilla", UUID.randomUUID());
+    }


Should the "zilla" part be coming from the client.id so that if the client id is configured, we see the effect here for the group instance id as well?

Note: we have access to the config, so it's easy to extract it as a local final variable before using it in the lambda.

jfallows · 2023-07-25T21:11:36Z

...ain/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaCacheGroupFactory.java

+        sender.accept(reset.typeId(), reset.buffer(), reset.offset(), reset.sizeof());
+    }
+
+    final class KafkaCacheServerGroupNet


Suggest renaming to KafkaCacheGroupNet

jfallows · 2023-07-25T21:12:21Z

...ain/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaCacheGroupFactory.java

+        }
+    }
+
+    private final class KafkaCacheServerGroupApp


Suggest renaming to KafkaCacheGroupApp

jfallows · 2023-07-25T21:19:16Z

...in/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaClientGroupFactory.java

+                        stream.streamCleanup(traceId, traceId);
+                    }
+
+                    KafkaGroupStream kafkaGroupStream = new KafkaGroupStream(


Suggest rename variable to groupStream (or group) as we already in Kafka context.

jfallows · 2023-07-25T21:22:36Z

...in/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaClientGroupFactory.java

+        }
+    }
+
+    private final class GroupIdentifier


Suggest renaming to GroupMembership with fields instanceId and memberIds.

jfallows · 2023-07-25T21:58:38Z

...vity/zilla/specs/binding/kafka/streams/network/group/not.coordinator.for.consumer/client.rpt

@@ -0,0 +1,176 @@
+#


Suggest renaming scenario to coordinator.reject.invalid.consumer.

jfallows · 2023-07-25T22:01:13Z

...aklivity/zilla/specs/binding/kafka/streams/network/group/no.coordinator.available/client.rpt

@@ -0,0 +1,150 @@
+#


Suggest renaming scenario to coordinator.not.available.

jfallows · 2023-07-25T22:03:59Z

.../kafka/streams/network/group/client.sent.write.abort.before.coordinator.connected/client.rpt

@@ -0,0 +1,41 @@
+#


What does connected mean here?

Suggest renaming scenario to client.sent.write.abort.before.coordinator.response.

Makes sesnse

jfallows · 2023-07-25T22:04:42Z

...lla/specs/binding/kafka/streams/network/group/highlander.protocol.stream.takeover/client.rpt

@@ -0,0 +1,195 @@
+#


Suggest rename scenario to rebalance.protocol.highlander

jfallows · 2023-07-25T22:10:24Z

...n/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/group.sasl/leader/client.rpt

@@ -0,0 +1,165 @@
+#


Note: each of the other network script categories indicate the versions of the Kafka wire protocol api(s) in use, even if multiple - for example unmerged.p3.f5.d0.m5. 😄

Suggest doing the same for group category, also for group.sasl category.

jfallows · 2023-07-25T23:14:23Z

...kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/KafkaConfiguration.java

@@ -100,6 +105,11 @@ public class KafkaConfiguration extends Configuration
        KAFKA_CACHE_CLIENT_TRAILERS_SIZE_MAX = config.property("cache.client.trailers.size.max", 256);
        KAFKA_CLIENT_SASL_SCRAM_NONCE = config.property(NonceSupplier.class, "client.sasl.scram.nonce",
                KafkaConfiguration::decodeNonceSupplier, KafkaConfiguration::defaultNonceSupplier);
+        KAFKA_CLIENT_GROUP_INSTANCE_ID = config.property("client.group.instance.id", UUID.randomUUID().toString());
+        KAFKA_CLIENT_GROUP_REBALANCE_TIMEOUT = config.property("client.group.rebalance.timeout", 4000);


This should be of type Duration, with default PT4S (period time 4 seconds).

jfallows · 2023-07-26T18:41:22Z

...kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/KafkaConfiguration.java

+    {
+        return Duration.parse(timeout);
+    }
+


Please inline this lambda as (c, v) -> Duration.parse(v) in the _PROPERTY assignment above and remove this method.

jfallows · 2023-07-26T18:43:15Z

...kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/KafkaConfiguration.java

+        KAFKA_CLIENT_GROUP_REBALANCE_TIMEOUT = config.property(Duration.class, "client.group.rebalance.timeout",
+            KafkaConfiguration::rebalanceTimeoutDuration, "PT4S");
+        KAFKA_CLIENT_ID = config.property("client.id", "zilla");
+        KAFKA_CLIENT_INSTANCE_ID_SUPPLIER = config.property(InstanceIdSupplier.class, "client.instance.id.supplier",


Please remove .supplier from property name, remove _SUPPLIER from constant name, and remove Supplier suffix from property accessor method below.

What is the difference between "client.group.instance.id" and "client.instance.id.supplier"?

If they represent the same concept then we just need "client.group.instance.id" of type InstanceIdSupplier and default to a supplier that generates UUID.randomUUID().toString().

jfallows · 2023-07-26T18:46:44Z

runtime/engine/src/main/java/io/aklivity/zilla/runtime/engine/Configuration.java

@@ -20,6 +20,7 @@

 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.time.Duration;


Please remove this import.

jfallows · 2023-07-26T19:12:36Z

...specs/binding/kafka/streams/application/group/highlander.protocol.stream.takeover/client.rpt

@@ -0,0 +1,69 @@
+#


Please see comment above.

jfallows · 2023-07-26T19:13:28Z

...in/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/group/leader/client.rpt

+read zilla:data.ext ${kafka:dataEx()
+                             .typeId(zilla:id("kafka"))
+                             .group()
+                                 .leaderId("memberid-1")


Sorry to nitpick, but this should be memberId-1 not memberid-1, right?

jfallows · 2023-07-26T19:14:20Z

...ng/kafka/streams/network/group.f1.j5.s3.l3.h3/highlander.protocol.stream.takeover/client.rpt

@@ -0,0 +1,195 @@
+#


Rename scenario?

jfallows · 2023-07-26T19:15:46Z

runtime/binding-kafka/pom.xml

+      <dependency>
+          <groupId>org.jboss.byteman</groupId>
+          <artifactId>byteman</artifactId>
+      </dependency>


Please remove this dependency, it should be unnecessary. We only use it for TCP binding to force behavior of the socket APIs when simulating different network interactions.

Akram Yakubov added 23 commits April 28, 2023 17:10

Checkpoint

e49a458

Checkpoint

1dea9a5

Checkpoint

4209230

Chechpoint

8f9ae54

Combine consumer group handshake

872762e

Rename

860a780

Checkpoint

b8c4a5a

Checkpoint

a955eee

Checkpoint

7107968

Checkpoint

6f8e5fb

Add unit tests

2e67dd0

Merge branch 'feature/consumer-group-idl' into feature/consumer-group

ad7444d

Checkpoint

919a0d4

Checkpoint

306543b

Merge branch 'develop' into feature/consumer-group

c1ce5a3

Checkpoint

b8dd0bb

Checkpoint

0f542fe

Checkpoint

175bf8b

Checkpoint

ed3e339

Checkpoint

802cff7

Merge branch 'develop' into feature/consumer-group

5e6a742

Checkpoint

df8da88

Initial implementation of consumer group

9729be6

jfallows requested changes May 29, 2023

View reviewed changes

Akram Yakubov added 2 commits May 30, 2023 19:10

Fix typo

30ba518

Revert back changes

af4eaa9

akrambek commented May 30, 2023

View reviewed changes

jfallows requested changes May 30, 2023

View reviewed changes

Akram Yakubov added 2 commits June 8, 2023 07:31

Checkpoint

d5beb2c

Checkpoint

2471375

Akram Yakubov added 9 commits July 24, 2023 21:26

Handle the same group client coming to the same core id

c1c0e51

Checkpoint

e9f2d44

Checkpoint

9df825a

Add more tests

6f2a664

Merge branch 'develop' into feature/consumer-group

5b431c4

Fix typo

8c435cb

Add sasl test

2a52135

Fix typo

2aeceec

Fix script bug

8e80f42

jfallows changed the title ~~Consumer group~~ Support Kafka consumer groups Jul 25, 2023

jfallows requested changes Jul 25, 2023

View reviewed changes

Akram Yakubov added 5 commits July 26, 2023 18:48

Checkpoint

85d03ce

Checkpoint

9add48f

Checkpoint

23ffbfe

Checkpoint

344dee5

Fix configuration decoder

002d4c1

jfallows requested changes Jul 26, 2023

View reviewed changes

Akram Yakubov added 5 commits July 27, 2023 16:10

Apply changes

8c85e39

Apply feedback from PR

bb07f7d

Remove uncessary code

ef5090b

Fix remaining issues

eddfbea

Remove extra configuration

ae6eef7

jfallows previously approved these changes Jul 27, 2023

View reviewed changes

Fix typo

a71ace4

akrambek dismissed jfallows’s stale review via a71ace4 July 28, 2023 12:24

Fix another typo

d87d676

jfallows approved these changes Jul 28, 2023

View reviewed changes

jfallows merged commit 80e2091 into aklivity:develop Jul 28, 2023
2 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Support Kafka consumer groups #262

Support Kafka consumer groups #262

akrambek commented May 26, 2023 •

edited by jfallows

Loading

jfallows May 29, 2023

akrambek May 30, 2023

akrambek May 30, 2023 •

edited

Loading

jfallows May 30, 2023

akrambek May 31, 2023 •

edited

Loading

jfallows left a comment

jfallows May 30, 2023

jfallows May 30, 2023

jfallows May 30, 2023

akrambek May 31, 2023

jfallows Jun 23, 2023

jfallows May 30, 2023

jfallows Jul 25, 2023

jfallows Jul 25, 2023

jfallows Jul 25, 2023

jfallows Jul 25, 2023

jfallows Jul 25, 2023

jfallows Jul 25, 2023

jfallows Jul 25, 2023

jfallows Jul 25, 2023

akrambek Jul 26, 2023

jfallows Jul 25, 2023

jfallows Jul 25, 2023

jfallows Jul 25, 2023

jfallows Jul 26, 2023

jfallows Jul 26, 2023

jfallows Jul 26, 2023

jfallows Jul 26, 2023

jfallows Jul 26, 2023

jfallows Jul 26, 2023

jfallows Jul 26, 2023

Support Kafka consumer groups #262

Support Kafka consumer groups #262

Conversation

akrambek commented May 26, 2023 • edited by jfallows Loading

Description

Choose a reason for hiding this comment

Choose a reason for hiding this comment

akrambek May 30, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

akrambek May 31, 2023 • edited Loading

Choose a reason for hiding this comment

jfallows left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

akrambek commented May 26, 2023 •

edited by jfallows

Loading

akrambek May 30, 2023 •

edited

Loading

akrambek May 31, 2023 •

edited

Loading