KAFKA-18632: Added few share consumer multibroker tests. #18679

smjn · 2025-01-23T17:45:22Z

Added test to check share consumer behavior across share coordinator restart.
Another simple test based on a complicated share consumer.

AndrewJSchofield

Thanks for the PR. A handful of comments to address.

AndrewJSchofield · 2025-01-24T10:22:32Z

core/src/test/java/kafka/test/api/ShareConsumerTest.java

+            @ClusterConfigProperty(key = "group.share.record.lock.duration.ms", value = "15000"),
+            @ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "3"),
+            @ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "3"),
+            @ClusterConfigProperty(key = "share.coordinator.state.topic.min.isr", value = "1"),


It seems to me that this override is unnecessary with 3 brokers and the default of 2 is better.

AndrewJSchofield · 2025-01-24T10:22:51Z

core/src/test/java/kafka/test/api/ShareConsumerTest.java

+            @ClusterConfigProperty(key = "share.coordinator.state.topic.min.isr", value = "1"),
+            @ClusterConfigProperty(key = "share.coordinator.state.topic.num.partitions", value = "3"),
+            @ClusterConfigProperty(key = "share.coordinator.state.topic.replication.factor", value = "3"),
+            @ClusterConfigProperty(key = "transaction.state.log.min.isr", value = "1"),


Leave this as the default of 2 also I think.

AndrewJSchofield · 2025-01-24T10:23:45Z

core/src/test/java/kafka/test/api/ShareConsumerTest.java

+            @ClusterConfigProperty(key = "share.coordinator.state.topic.num.partitions", value = "3"),
+            @ClusterConfigProperty(key = "share.coordinator.state.topic.replication.factor", value = "3"),
+            @ClusterConfigProperty(key = "transaction.state.log.min.isr", value = "1"),
+            @ClusterConfigProperty(key = "transaction.state.log.replication.factor", value = "3"),


Actually, the default replication factor of all three of these internal topics is 3 also, which should be ideal for this test with 3 brokers.

It is necessary because the class level ClusterTestDefaults defines it to be 1. If we do not add the override here - it will be set to 1 and the re-election will not happen as I shut down a broker in the test.

AndrewJSchofield · 2025-01-24T10:26:20Z

core/src/test/java/kafka/test/api/ShareConsumerTest.java

+        setup();
+        String topicName = "multipart";
+        String groupId = "multipartGrp";
+        createTopic(topicName, 3, 3);


Actually this createTopic method knows the topic ID and discards it. I suggest changing the signature of this method to return a Uuid and then the topic ID is known without the extra call to Admin.describeTopics.

AndrewJSchofield · 2025-01-24T10:28:03Z

core/src/test/java/kafka/test/api/ShareConsumerTest.java

+            // consume messages
+            try (ShareConsumer<byte[], byte[]> shareConsumer = createShareConsumer(groupId)) {
+                shareConsumer.subscribe(List.of(topicName));
+                alterShareAutoOffsetReset(groupId, "earliest");


Experience with other tests tells us that it's best to set the auto-offset reset earlier in the test. It's an asynchronous action that may or may not take effect by the time consumption begins.

AndrewJSchofield · 2025-01-24T10:31:57Z

core/src/test/java/kafka/test/api/ShareConsumerTest.java

+            // get current share coordinator node
+            SharePartitionKey key = SharePartitionKey.getInstance(groupId, new TopicIdPartition(topicId, tpMulti));
+            int shareGroupStateTp = Utils.abs(key.asCoordinatorKey().hashCode()) % 3;
+            List<Integer> curShareCoordNodeId = admin.describeTopics(List.of(Topic.SHARE_GROUP_STATE_TOPIC_NAME)).topicNameValues().get(Topic.SHARE_GROUP_STATE_TOPIC_NAME).get()


I'd prefer admin.describeTopics(...).allTopicNames().get(Topic.SHARE_GROUP_TOPIC_NAME). I think that yields the same result more tersely.

AndrewJSchofield · 2025-01-24T10:33:24Z

core/src/test/java/kafka/test/api/ShareConsumerTest.java

+            @ClusterConfigProperty(key = "group.share.record.lock.duration.ms", value = "15000"),
+            @ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "3"),
+            @ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "3"),
+            @ClusterConfigProperty(key = "share.coordinator.state.topic.min.isr", value = "1"),


Same comments as previous test regarding min.isr and replication factor defaults which are appropriate without overrides.

AndrewJSchofield · 2025-01-24T10:35:15Z

core/src/test/java/kafka/test/api/ShareConsumerTest.java

+
+        // produce messages until we want
+        executer.execute(() -> {
+            while (!prodDone.get()) {


This makes a separate producer per loop iteration. I suggest re-using the producer across loop iterations. You can still use try-with-resources.

AndrewJSchofield · 2025-01-24T10:36:25Z

core/src/test/java/kafka/test/api/ShareConsumerTest.java

@@ -2015,4 +2204,88 @@ private void alterShareAutoOffsetReset(String groupId, String newValue) {
                .get(60, TimeUnit.SECONDS), "Failed to alter configs");
        }
    }
+
+    private static class ComplexShareConsumer<K, V> implements Runnable {


This really needs a comment about what it's up to.

AndrewJSchofield · 2025-01-24T10:37:38Z

core/src/test/java/kafka/test/api/ShareConsumerTest.java

+
+    private static class ComplexShareConsumer<K, V> implements Runnable {
+        public static final int POLL_TIMEOUT_MS = 15000;
+        public static final int MAX_DELIVERY_COUNT = 5;


This 5 is really ShareGroupConfig.SHARE_GROUP_DELIVERY_COUNT_LIMIT_DEFAULT.

smjn added 2 commits January 23, 2025 22:25

KAFKA-18632: Multibroker test to validate operation after coord move.

d5e90f9

Added complex share consumer test.

769df3a

github-actions bot added triage PRs from the community core Kafka Broker tests Test fixes (including flaky tests) labels Jan 23, 2025

smjn added 2 commits January 24, 2025 02:42

generified complex consumer.

223d616

minor refactoring

3a14c24

AndrewJSchofield added KIP-932 Queues for Kafka ci-approved and removed triage PRs from the community labels Jan 24, 2025

AndrewJSchofield requested changes Jan 24, 2025

View reviewed changes

incroporated comments.

df5e5f3

smjn requested a review from AndrewJSchofield January 24, 2025 11:55

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-18632: Added few share consumer multibroker tests. #18679

KAFKA-18632: Added few share consumer multibroker tests. #18679

smjn commented Jan 23, 2025

AndrewJSchofield left a comment

AndrewJSchofield Jan 24, 2025

AndrewJSchofield Jan 24, 2025

AndrewJSchofield Jan 24, 2025

smjn Jan 24, 2025 •

edited

Loading

AndrewJSchofield Jan 24, 2025

AndrewJSchofield Jan 24, 2025

AndrewJSchofield Jan 24, 2025

AndrewJSchofield Jan 24, 2025

AndrewJSchofield Jan 24, 2025

AndrewJSchofield Jan 24, 2025

AndrewJSchofield Jan 24, 2025

KAFKA-18632: Added few share consumer multibroker tests. #18679

Are you sure you want to change the base?

KAFKA-18632: Added few share consumer multibroker tests. #18679

Conversation

smjn commented Jan 23, 2025

AndrewJSchofield 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

smjn Jan 24, 2025 • edited Loading

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

smjn Jan 24, 2025 •

edited

Loading