You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2026/10/15 23:16:37 UTC
(kafka) branch trunk updated: MINOR: Migrate PlaintextConsumerAssignorsTest to Java (#23708)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new f7a6dfe0769 MINOR: Migrate PlaintextConsumerAssignorsTest to Java (#23708)
f7a6dfe0769 is described below
commit f7a6dfe076929c3b761bfb9fbb97f786ff9503e8
Author: Andy Huang <ky...@gmail.com>
AuthorDate: Fri Oct 16 07:16:28 2026 +0800
MINOR: Migrate PlaintextConsumerAssignorsTest to Java (#23708)
Migrates `PlaintextConsumerAssignorsTest` from Scala in `core` to Java
in `clients-integration-tests` using `ClusterTestExtensions`, as part of
KAFKA-19042.
Preserves all nine assignor scenarios and reuses the existing
`ConsumerAssignmentPoller` and `ClientsTestUtils` helpers. The group
assignment checks now fail fast when a poller thread hits an exception,
instead of waiting for the timeout. Pollers and consumers are closed
after each test. Removes the original Scala test.
Reviewers: Chia-Ping Tsai
<[chia7712@gmail.com](mailto:chia7712@gmail.com)>
---
.../consumer/PlaintextConsumerAssignorsTest.java | 381 ++++++++++++++++++++
.../kafka/api/PlaintextConsumerAssignorsTest.scala | 391 ---------------------
2 files changed, 381 insertions(+), 391 deletions(-)
diff --git a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerAssignorsTest.java b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerAssignorsTest.java
new file mode 100644
index 00000000000..0c29376c314
--- /dev/null
+++ b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerAssignorsTest.java
@@ -0,0 +1,381 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer;
+
+import org.apache.kafka.clients.ClientsTestUtils;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.UnsupportedAssignorException;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.ClusterTestDefaults;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.test.NoRetryException;
+import org.apache.kafka.test.TestUtils;
+
+import org.junit.jupiter.api.Timeout;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Integration tests for client-side and server-side consumer assignors.
+ */
+@Timeout(600)
+@ClusterTestDefaults(
+ types = {Type.KRAFT},
+ brokers = 3,
+ serverProperties = {
+ @ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "3"),
+ @ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+ @ClusterConfigProperty(key = "group.min.session.timeout.ms", value = "100"),
+ @ClusterConfigProperty(key = "group.max.session.timeout.ms", value = "60000"),
+ @ClusterConfigProperty(key = "group.initial.rebalance.delay.ms", value = "10")
+ }
+)
+public class PlaintextConsumerAssignorsTest {
+ private final ClusterInstance cluster;
+
+ public PlaintextConsumerAssignorsTest(ClusterInstance cluster) {
+ this.cluster = cluster;
+ }
+
+ private boolean isPartitionAssignmentValid(List<Set<TopicPartition>> assignments,
+ Set<TopicPartition> partitions,
+ List<Set<TopicPartition>> expectedAssignments) {
+ if (assignments.stream().anyMatch(Set::isEmpty)
+ || assignments.stream().mapToInt(Set::size).sum() != partitions.size()) {
+ return false;
+ }
+ var assigned = new HashSet<TopicPartition>();
+ assignments.forEach(assigned::addAll);
+ return assigned.equals(partitions)
+ && (expectedAssignments.isEmpty()
+ || (assignments.size() == expectedAssignments.size()
+ && new HashSet<>(assignments).equals(new HashSet<>(expectedAssignments))));
+ }
+
+ @ClusterTest
+ public void testRoundRobinAssignment() throws Exception {
+ testSubscriptionChanges(Map.of(
+ ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name,
+ ConsumerConfig.GROUP_ID_CONFIG, "roundrobin-group",
+ ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, RoundRobinAssignor.class.getName()
+ ));
+ }
+
+ @ClusterTest
+ public void testRemoteAssignorRange() throws Exception {
+ testSubscriptionChanges(Map.of(
+ ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CONSUMER.name,
+ ConsumerConfig.GROUP_ID_CONFIG, "range-group",
+ ConsumerConfig.GROUP_REMOTE_ASSIGNOR_CONFIG, "range",
+ ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "30000"
+ ));
+ }
+
+ private void testSubscriptionChanges(Map<String, String> configs) throws Exception {
+ try (var consumer = createConsumer(configs)) {
+ var expected = createTopicAndSendRecords("topic1", 2);
+ expected.addAll(createTopicAndSendRecords("topic2", 2));
+ assertTrue(consumer.assignment().isEmpty());
+ consumer.subscribe(List.of("topic1", "topic2"));
+ ClientsTestUtils.awaitAssignment(consumer, expected);
+
+ var expanded = new HashSet<>(expected);
+ expanded.addAll(createTopicAndSendRecords("topic3", 2));
+ consumer.subscribe(List.of("topic1", "topic2", "topic3"));
+ ClientsTestUtils.awaitAssignment(consumer, expanded);
+ consumer.subscribe(List.of("topic1", "topic2"));
+ ClientsTestUtils.awaitAssignment(consumer, expected);
+ consumer.unsubscribe();
+ assertTrue(consumer.assignment().isEmpty());
+ }
+ }
+
+ @ClusterTest
+ public void testRemoteAssignorInvalid() throws Exception {
+ try (var consumer = createConsumer(Map.of(
+ ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CONSUMER.name,
+ ConsumerConfig.GROUP_ID_CONFIG, "invalid-assignor-group",
+ ConsumerConfig.GROUP_REMOTE_ASSIGNOR_CONFIG, "invalid"
+ ))) {
+ createTopicAndSendRecords("topic1", 2);
+ assertTrue(consumer.assignment().isEmpty());
+ consumer.subscribe(List.of("topic1"));
+ var failure = new AtomicReference<UnsupportedAssignorException>();
+ TestUtils.waitForCondition(() -> {
+ try {
+ consumer.poll(Duration.ofMillis(100));
+ return false;
+ } catch (UnsupportedAssignorException e) {
+ failure.set(e);
+ return true;
+ }
+ }, "Consumer did not reject the invalid remote assignor");
+ assertTrue(failure.get().getMessage().startsWith(
+ "ServerAssignor invalid is not supported. Supported assignors: "));
+ }
+ }
+
+ @ClusterTest
+ public void testMultiConsumerRoundRobinAssignor() throws Exception {
+ var consumers = new ArrayList<Consumer<byte[], byte[]>>();
+ var pollers = new ArrayList<ConsumerAssignmentPoller>();
+ try {
+ var configs = Map.of(
+ ConsumerConfig.GROUP_ID_CONFIG, "roundrobin-group",
+ ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, RoundRobinAssignor.class.getName()
+ );
+ var partitions = createTopicAndSendRecords("topic1", 5);
+ partitions.addAll(createTopicAndSendRecords("topic2", 8));
+ var topics = List.of("topic1", "topic2");
+ addConsumers(consumers, pollers, 10, topics, configs);
+ awaitGroupAssignment(pollers, partitions);
+ addConsumers(consumers, pollers, 1, topics, configs);
+ awaitGroupAssignment(pollers, partitions);
+ } finally {
+ closeConsumers(consumers, pollers);
+ }
+ }
+
+ @ClusterTest
+ public void testMultiConsumerStickyAssignor() throws Exception {
+ var consumers = new ArrayList<Consumer<byte[], byte[]>>();
+ var pollers = new ArrayList<ConsumerAssignmentPoller>();
+ try {
+ var configs = Map.of(
+ ConsumerConfig.GROUP_ID_CONFIG, "sticky-group",
+ ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, StickyAssignor.class.getName()
+ );
+ // Adding the tenth consumer should move exactly one tenth of the partitions.
+ int partitionsPerConsumer = ThreadLocalRandom.current().nextInt(1, 11);
+ var partitions = createTopicAndSendRecords("single-topic", partitionsPerConsumer * 10);
+ var topics = List.of("single-topic");
+ addConsumers(consumers, pollers, 9, topics, configs);
+ awaitGroupAssignment(pollers, partitions);
+ var previousOwners = partitionOwners(pollers);
+ addConsumers(consumers, pollers, 1, topics, configs);
+ awaitGroupAssignment(pollers, partitions);
+ var newOwners = partitionOwners(pollers);
+ long changes = partitions.stream()
+ .filter(partition -> previousOwners.get(partition) != newOwners.get(partition)).count();
+ assertEquals(partitionsPerConsumer, changes,
+ "Only the partitions needed by the tenth consumer should change owners");
+ } finally {
+ closeConsumers(consumers, pollers);
+ }
+ }
+
+ private Map<TopicPartition, ConsumerAssignmentPoller> partitionOwners(List<ConsumerAssignmentPoller> pollers) {
+ var owners = new HashMap<TopicPartition, ConsumerAssignmentPoller>();
+ pollers.forEach(poller -> poller.consumerAssignment().forEach(partition -> owners.put(partition, poller)));
+ return owners;
+ }
+
+ @ClusterTest
+ public void testMultiConsumerDefaultAssignorAndVerifyAssignment() throws Exception {
+ var consumers = new ArrayList<Consumer<byte[], byte[]>>();
+ var pollers = new ArrayList<ConsumerAssignmentPoller>();
+ try {
+ cluster.createTopic("topic1", 3, (short) 3);
+ cluster.createTopic("topic2", 3, (short) 3);
+ var first = Set.of(new TopicPartition("topic1", 0), new TopicPartition("topic1", 1),
+ new TopicPartition("topic2", 0), new TopicPartition("topic2", 1));
+ var second = Set.of(new TopicPartition("topic1", 2), new TopicPartition("topic2", 2));
+ var partitions = new HashSet<>(first);
+ partitions.addAll(second);
+ // Leave the assignment strategy unset to exercise the default Range assignor.
+ addConsumers(consumers, pollers, 2, List.of("topic1", "topic2"), Map.of());
+ awaitGroupAssignment(pollers, partitions, List.of(first, second));
+ } finally {
+ closeConsumers(consumers, pollers);
+ }
+ }
+
+ @ClusterTest
+ public void testMultiConsumerDefaultAssignor() throws Exception {
+ var consumers = new ArrayList<Consumer<byte[], byte[]>>();
+ var pollers = new ArrayList<ConsumerAssignmentPoller>();
+ try {
+ var partitions = createTopicAndSendRecords("topic", 2);
+ partitions.addAll(createTopicAndSendRecords("topic1", 5));
+ var topics = List.of("topic", "topic1");
+ addConsumers(consumers, pollers, 2, topics, Map.of());
+ awaitGroupAssignment(pollers, partitions);
+ addConsumers(consumers, pollers, 2, topics, Map.of());
+ awaitGroupAssignment(pollers, partitions);
+ var expanded = new HashSet<>(partitions);
+ expanded.addAll(createTopicAndSendRecords("topic2", 3));
+ changeSubscriptions(pollers, List.of("topic", "topic1", "topic2"), expanded);
+ changeSubscriptions(pollers, topics, partitions);
+ } finally {
+ closeConsumers(consumers, pollers);
+ }
+ }
+
+ @ClusterTest
+ public void testRebalanceAndRejoinWithRangeAssignor() throws Exception {
+ testRebalanceAndRejoin(RangeAssignor.class.getName(), 1);
+ }
+
+ @ClusterTest
+ public void testRebalanceAndRejoinWithCooperativeStickyAssignor() throws Exception {
+ testRebalanceAndRejoin(CooperativeStickyAssignor.class.getName(), 2);
+ }
+
+ private void testRebalanceAndRejoin(String strategy, int expectedGenerations) throws Exception {
+ var configs = Map.of(
+ ConsumerConfig.GROUP_ID_CONFIG, "rebalance-and-rejoin-group",
+ ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, strategy,
+ ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true"
+ );
+ var pollers = new ArrayList<ConsumerAssignmentPoller>();
+ try (var consumer1 = createConsumer(configs);
+ var consumer2 = createConsumer(configs)) {
+ try {
+ var partitions = createTopicAndSendRecords("topic1", 2);
+ assertTrue(consumer1.assignment().isEmpty());
+ assertTrue(consumer2.assignment().isEmpty());
+ var metadata = new AtomicReference<ConsumerGroupMetadata>();
+ var listener = new ConsumerRebalanceListener() {
+ @Override
+ public void onPartitionsRevoked(Collection<TopicPartition> revoked) {
+ }
+
+ @Override
+ public void onPartitionsAssigned(Collection<TopicPartition> assigned) {
+ metadata.set(consumer1.groupMetadata());
+ }
+ };
+ var poller1 = startPolling(pollers, consumer1, List.of("topic1"), listener);
+ TestUtils.waitForCondition(() -> {
+ checkPollerExceptions(pollers);
+ return poller1.consumerAssignment().equals(partitions) && metadata.get() != null;
+ }, "First consumer did not complete its initial assignment");
+ var stableMetadata = metadata.get();
+ var poller2 = startPolling(pollers, consumer2, List.of("topic1"), null);
+ TestUtils.waitForCondition(() -> {
+ checkPollerExceptions(pollers);
+ return poller1.consumerAssignment().size() == 1
+ && poller2.consumerAssignment().size() == 1
+ && metadata.get().generationId() >= stableMetadata.generationId() + expectedGenerations;
+ }, "Consumers did not complete the expected rebalance");
+ assertEquals(stableMetadata.generationId() + expectedGenerations, metadata.get().generationId());
+ assertEquals(stableMetadata.memberId(), metadata.get().memberId());
+ } finally {
+ assertAll(pollers.stream().map(poller -> poller::shutdown));
+ }
+ }
+ }
+
+ private Consumer<byte[], byte[]> createConsumer(Map<String, String> overrides) {
+ var configs = new HashMap<String, Object>();
+ configs.put(ConsumerConfig.GROUP_ID_CONFIG, "my-test");
+ configs.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name);
+ configs.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
+ configs.put(ConsumerConfig.METADATA_MAX_AGE_CONFIG, "100");
+ configs.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "6000");
+ configs.putAll(overrides);
+ return cluster.consumer(configs);
+ }
+
+ private Set<TopicPartition> createTopicAndSendRecords(String topic, int partitionCount) throws Exception {
+ cluster.createTopic(topic, partitionCount, (short) 3);
+ var partitions = new HashSet<TopicPartition>();
+ try (Producer<byte[], byte[]> producer = cluster.producer()) {
+ for (int partition = 0; partition < partitionCount; partition++) {
+ var tp = new TopicPartition(topic, partition);
+ partitions.add(tp);
+ ClientsTestUtils.sendRecords(producer, tp, 100, System.currentTimeMillis());
+ }
+ }
+ return partitions;
+ }
+
+ private void addConsumers(List<Consumer<byte[], byte[]>> consumers, List<ConsumerAssignmentPoller> pollers,
+ int count, List<String> topics, Map<String, String> configs) {
+ for (int i = 0; i < count; i++) {
+ var consumer = createConsumer(configs);
+ consumers.add(consumer);
+ startPolling(pollers, consumer, topics, null);
+ }
+ }
+
+ private ConsumerAssignmentPoller startPolling(List<ConsumerAssignmentPoller> pollers,
+ Consumer<byte[], byte[]> consumer, List<String> topics,
+ ConsumerRebalanceListener listener) {
+ var poller = new ConsumerAssignmentPoller(consumer, topics, Set.of(), listener);
+ pollers.add(poller);
+ poller.start();
+ return poller;
+ }
+
+ private void changeSubscriptions(List<ConsumerAssignmentPoller> pollers, List<String> topics, Set<TopicPartition> partitions) throws InterruptedException {
+ pollers.forEach(poller -> poller.subscribe(topics));
+ TestUtils.waitForCondition(() -> {
+ checkPollerExceptions(pollers);
+ return pollers.stream().allMatch(ConsumerAssignmentPoller::isSubscribeRequestProcessed);
+ }, "Not all consumers processed the subscription change");
+ awaitGroupAssignment(pollers, partitions);
+ }
+
+ private void awaitGroupAssignment(List<ConsumerAssignmentPoller> pollers, Set<TopicPartition> partitions) throws InterruptedException {
+ awaitGroupAssignment(pollers, partitions, List.of());
+ }
+
+ private void awaitGroupAssignment(List<ConsumerAssignmentPoller> pollers, Set<TopicPartition> partitions,
+ List<Set<TopicPartition>> expectedAssignments) throws InterruptedException {
+ TestUtils.waitForCondition(() -> {
+ checkPollerExceptions(pollers);
+ var assignments = new ArrayList<Set<TopicPartition>>();
+ for (var poller : pollers) {
+ assignments.add(poller.consumerAssignment());
+ }
+ return isPartitionAssignmentValid(assignments, partitions, expectedAssignments);
+ }, "Consumers did not reach the expected assignment for " + partitions);
+ }
+
+ private void checkPollerExceptions(List<ConsumerAssignmentPoller> pollers) {
+ for (var poller : pollers) {
+ poller.getThrownException().ifPresent(exception -> {
+ throw new NoRetryException(exception);
+ });
+ }
+ }
+
+ private void closeConsumers(List<Consumer<byte[], byte[]>> consumers, List<ConsumerAssignmentPoller> pollers) {
+ try {
+ assertAll(pollers.stream().map(poller -> poller::shutdown));
+ } finally {
+ assertAll(consumers.stream().map(consumer -> consumer::close));
+ }
+ }
+}
diff --git a/core/src/test/scala/integration/kafka/api/PlaintextConsumerAssignorsTest.scala b/core/src/test/scala/integration/kafka/api/PlaintextConsumerAssignorsTest.scala
deleted file mode 100644
index 6173a2987ef..00000000000
--- a/core/src/test/scala/integration/kafka/api/PlaintextConsumerAssignorsTest.scala
+++ /dev/null
@@ -1,391 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE
- * file distributed with this work for additional information regarding copyright ownership. The ASF licenses this file
- * to You under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the
- * License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
- * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations under the License.
- */
-package kafka.api
-
-import kafka.utils.{TestInfoUtils, TestUtils}
-import org.apache.kafka.clients.consumer._
-import org.apache.kafka.common.TopicPartition
-import org.apache.kafka.common.errors.UnsupportedAssignorException
-import org.junit.jupiter.api.Assertions._
-import org.junit.jupiter.api.Timeout
-import org.junit.jupiter.params.ParameterizedTest
-import org.junit.jupiter.params.provider.{CsvSource, MethodSource}
-
-import java.util
-import java.util.concurrent.TimeUnit
-import java.util.concurrent.locks.ReentrantLock
-import scala.collection.mutable
-
-/**
- * Integration tests for the consumer that covers assignors logic (client and server side assignors)
- */
-@Timeout(600)
-class PlaintextConsumerAssignorsTest extends AbstractConsumerTest {
-
- // Only the classic group protocol supports client-side assignors
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedGroupProtocolNames)
- @MethodSource(Array("getTestGroupProtocolParametersClassicGroupProtocolOnly"))
- def testRoundRobinAssignment(groupProtocol: String): Unit = {
- // 1 consumer using round-robin assignment
- this.consumerConfig.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "roundrobin-group")
- this.consumerConfig.setProperty(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, classOf[RoundRobinAssignor].getName)
- val consumer = createConsumer()
-
- // create two new topics, each having 2 partitions
- val topic1 = "topic1"
- val topic2 = "topic2"
- val producer = createProducer()
- val expectedAssignment = createTopicAndSendRecords(producer, topic1, 2, 100) ++
- createTopicAndSendRecords(producer, topic2, 2, 100)
-
- assertEquals(0, consumer.assignment().size)
-
- // subscribe to two topics
- consumer.subscribe(java.util.List.of(topic1, topic2))
- awaitAssignment(consumer, expectedAssignment)
-
- // add one more topic with 2 partitions
- val topic3 = "topic3"
- createTopicAndSendRecords(producer, topic3, 2, 100)
-
- val newExpectedAssignment = expectedAssignment ++ Set(new TopicPartition(topic3, 0), new TopicPartition(topic3, 1))
- consumer.subscribe(java.util.List.of(topic1, topic2, topic3))
- awaitAssignment(consumer, newExpectedAssignment)
-
- // remove the topic we just added
- consumer.subscribe(java.util.List.of(topic1, topic2))
- awaitAssignment(consumer, expectedAssignment)
-
- consumer.unsubscribe()
- assertEquals(0, consumer.assignment().size)
- }
-
- // Only the classic group protocol supports client-side assignors
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedGroupProtocolNames)
- @MethodSource(Array("getTestGroupProtocolParametersClassicGroupProtocolOnly"))
- def testMultiConsumerRoundRobinAssignor(groupProtocol: String): Unit = {
- this.consumerConfig.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "roundrobin-group")
- this.consumerConfig.setProperty(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, classOf[RoundRobinAssignor].getName)
-
- // create two new topics, total number of partitions must be greater than number of consumers
- val topic1 = "topic1"
- val topic2 = "topic2"
- val producer = createProducer()
- val subscriptions = createTopicAndSendRecords(producer, topic1, 5, 100) ++
- createTopicAndSendRecords(producer, topic2, 8, 100)
-
- // create a group of consumers, subscribe the consumers to all the topics and start polling
- // for the topic partition assignment
- val (consumerGroup, consumerPollers) = createConsumerGroupAndWaitForAssignment(10, List(topic1, topic2), subscriptions)
- try {
- validateGroupAssignment(consumerPollers, subscriptions)
-
- // add one more consumer and validate re-assignment
- addConsumersToGroupAndWaitForGroupAssignment(1, consumerGroup, consumerPollers,
- List(topic1, topic2), subscriptions, "roundrobin-group")
- } finally {
- consumerPollers.foreach(_.shutdown())
- }
- }
-
- /**
- * This test runs the following scenario to verify sticky assignor behavior.
- * Topics: single-topic, with random number of partitions, where #par is 10, 20, 30, 40, 50, 60, 70, 80, 90, or 100
- * Consumers: 9 consumers subscribed to the single topic
- * Expected initial assignment: partitions are assigned to consumers in a round robin fashion.
- * - (#par mod 9) consumers will get (#par / 9 + 1) partitions, and the rest get (#par / 9) partitions
- * Then consumer #10 is added to the list (subscribing to the same single topic)
- * Expected new assignment:
- * - (#par / 10) partition per consumer, where one partition from each of the early (#par mod 9) consumers
- * will move to consumer #10, leading to a total of (#par mod 9) partition movement
- */
- // Only the classic group protocol supports client-side assignors
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedGroupProtocolNames)
- @MethodSource(Array("getTestGroupProtocolParametersClassicGroupProtocolOnly"))
- def testMultiConsumerStickyAssignor(groupProtocol: String): Unit = {
-
- def reverse(m: Map[Long, Set[TopicPartition]]) =
- m.values.toSet.flatten.map(v => (v, m.keys.filter(m(_).contains(v)).head)).toMap
-
- this.consumerConfig.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "sticky-group")
- this.consumerConfig.setProperty(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, classOf[StickyAssignor].getName)
-
- // create one new topic
- val topic = "single-topic"
- val rand = 1 + scala.util.Random.nextInt(10)
- val producer = createProducer()
- val partitions = createTopicAndSendRecords(producer, topic, rand * 10, 100)
-
- // create a group of consumers, subscribe the consumers to the single topic and start polling
- // for the topic partition assignment
- val (consumerGroup, consumerPollers) = createConsumerGroupAndWaitForAssignment(9, List(topic), partitions)
- validateGroupAssignment(consumerPollers, partitions)
- val prePartition2PollerId = reverse(consumerPollers.map(poller => (poller.getId, poller.consumerAssignment())).toMap)
-
- // add one more consumer and validate re-assignment
- addConsumersToGroupAndWaitForGroupAssignment(1, consumerGroup, consumerPollers, List(topic), partitions, "sticky-group")
-
- val postPartition2PollerId = reverse(consumerPollers.map(poller => (poller.getId, poller.consumerAssignment())).toMap)
- val keys = prePartition2PollerId.keySet.union(postPartition2PollerId.keySet)
- var changes = 0
- keys.foreach { key =>
- val preVal = prePartition2PollerId.get(key)
- val postVal = postPartition2PollerId.get(key)
- if (preVal.nonEmpty && postVal.nonEmpty) {
- if (preVal.get != postVal.get)
- changes += 1
- } else
- changes += 1
- }
-
- consumerPollers.foreach(_.shutdown())
-
- assertEquals(rand, changes, "Expected only two topic partitions that have switched to other consumers.")
- }
-
- /**
- * This test re-uses BaseConsumerTest's consumers.
- * As a result, it is testing the default assignment strategy set by BaseConsumerTest
- * It tests the assignment results is expected using default assignor (i.e. Range assignor)
- */
- // Only the classic group protocol supports client-side assignors
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedGroupProtocolNames)
- @MethodSource(Array("getTestGroupProtocolParametersClassicGroupProtocolOnly"))
- def testMultiConsumerDefaultAssignorAndVerifyAssignment(groupProtocol: String): Unit = {
- // create two new topics, each having 3 partitions
- val topic1 = "topic1"
- val topic2 = "topic2"
-
- createTopic(topic1, 3)
- createTopic(topic2, 3)
-
- val consumersInGroup = mutable.Buffer[Consumer[Array[Byte], Array[Byte]]]()
- consumersInGroup += createConsumer()
- consumersInGroup += createConsumer()
-
- val tp1_0 = new TopicPartition(topic1, 0)
- val tp1_1 = new TopicPartition(topic1, 1)
- val tp1_2 = new TopicPartition(topic1, 2)
- val tp2_0 = new TopicPartition(topic2, 0)
- val tp2_1 = new TopicPartition(topic2, 1)
- val tp2_2 = new TopicPartition(topic2, 2)
-
- val subscriptions = Set(tp1_0, tp1_1, tp1_2, tp2_0, tp2_1, tp2_2)
- val consumerPollers = subscribeConsumers(consumersInGroup, List(topic1, topic2))
-
- val expectedAssignment = mutable.Buffer(Set(tp1_0, tp1_1, tp2_0, tp2_1), Set(tp1_2, tp2_2))
-
- try {
- validateGroupAssignment(consumerPollers, subscriptions, expectedAssignment = expectedAssignment)
- } finally {
- consumerPollers.foreach(_.shutdown())
- }
- }
-
- /**
- * This test re-uses BaseConsumerTest's consumers.
- * As a result, it is testing the default assignment strategy set by BaseConsumerTest
- */
- // Only the classic group protocol supports client-side assignors
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedGroupProtocolNames)
- @MethodSource(Array("getTestGroupProtocolParametersClassicGroupProtocolOnly"))
- def testMultiConsumerDefaultAssignor(groupProtocol: String): Unit = {
- // use consumers and topics defined in this class + one more topic
- val producer = createProducer()
- sendRecords(producer, numRecords = 100, tp)
- sendRecords(producer, numRecords = 100, tp2)
- val topic1 = "topic1"
- val subscriptions = Set(tp, tp2) ++ createTopicAndSendRecords(producer, topic1, 5, 100)
-
- // subscribe all consumers to all topics and validate the assignment
-
- val consumersInGroup = mutable.Buffer[Consumer[Array[Byte], Array[Byte]]]()
- consumersInGroup += createConsumer()
- consumersInGroup += createConsumer()
-
- val consumerPollers = subscribeConsumers(consumersInGroup, List(topic, topic1))
- try {
- validateGroupAssignment(consumerPollers, subscriptions)
-
- // add 2 more consumers and validate re-assignment
- addConsumersToGroupAndWaitForGroupAssignment(2, consumersInGroup, consumerPollers, List(topic, topic1), subscriptions)
-
- // add one more topic and validate partition re-assignment
- val topic2 = "topic2"
- val expandedSubscriptions = subscriptions ++ createTopicAndSendRecords(producer, topic2, 3, 100)
- changeConsumerGroupSubscriptionAndValidateAssignment(consumerPollers, List(topic, topic1, topic2), expandedSubscriptions)
-
- // remove the topic we just added and validate re-assignment
- changeConsumerGroupSubscriptionAndValidateAssignment(consumerPollers, List(topic, topic1), subscriptions)
-
- } finally {
- consumerPollers.foreach(_.shutdown())
- }
- }
-
- // Remote assignors only supported with consumer group protocol
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedGroupProtocolNames)
- @MethodSource(Array("getTestGroupProtocolParametersConsumerGroupProtocolOnly"))
- def testRemoteAssignorInvalid(groupProtocol: String): Unit = {
- // 1 consumer using invalid remote assignor
- this.consumerConfig.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "invalid-assignor-group")
- this.consumerConfig.setProperty(ConsumerConfig.GROUP_REMOTE_ASSIGNOR_CONFIG, "invalid")
- val consumer = createConsumer()
-
- // create two new topics, each having 2 partitions
- val topic1 = "topic1"
- val producer = createProducer()
- val expectedAssignment = createTopicAndSendRecords(producer, topic1, 2, 100)
-
- assertEquals(0, consumer.assignment().size)
-
- // subscribe to two topics
- consumer.subscribe(java.util.List.of(topic1))
-
- val e: UnsupportedAssignorException = assertThrows(
- classOf[UnsupportedAssignorException],
- () => awaitAssignment(consumer, expectedAssignment)
- )
-
- assertTrue(e.getMessage.startsWith("ServerAssignor invalid is not supported. " +
- "Supported assignors: "))
- }
-
- // Remote assignors only supported with consumer group protocol
- @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedGroupProtocolNames)
- @MethodSource(Array("getTestGroupProtocolParametersConsumerGroupProtocolOnly"))
- def testRemoteAssignorRange(groupProtocol: String): Unit = {
- // 1 consumer using range assignment
- this.consumerConfig.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "range-group")
- this.consumerConfig.setProperty(ConsumerConfig.GROUP_REMOTE_ASSIGNOR_CONFIG, "range")
- this.consumerConfig.setProperty(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "30000")
- this.consumerConfig.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
- val consumer = createConsumer()
-
- // create two new topics, each having 2 partitions
- val topic1 = "topic1"
- val topic2 = "topic2"
- val producer = createProducer()
- val expectedAssignment = createTopicAndSendRecords(producer, topic1, 2, 100) ++
- createTopicAndSendRecords(producer, topic2, 2, 100)
-
- assertEquals(0, consumer.assignment().size)
-
- // subscribe to two topics
- consumer.subscribe(java.util.List.of(topic1, topic2))
- awaitAssignment(consumer, expectedAssignment)
-
- // add one more topic with 2 partitions
- val topic3 = "topic3"
- val additionalAssignment = createTopicAndSendRecords(producer, topic3, 2, 100)
-
- val newExpectedAssignment = expectedAssignment ++ additionalAssignment
- consumer.subscribe(java.util.List.of(topic1, topic2, topic3))
- awaitAssignment(consumer, newExpectedAssignment)
-
- // remove the topic we just added
- consumer.subscribe(java.util.List.of(topic1, topic2))
- awaitAssignment(consumer, expectedAssignment)
-
- consumer.unsubscribe()
- assertEquals(0, consumer.assignment().size)
- }
-
- // Only the classic group protocol supports client-side assignors
- @ParameterizedTest(name = "{displayName}.groupProtocol={0}.assignmentStrategy={1}")
- @CsvSource(Array(
- "classic, org.apache.kafka.clients.consumer.CooperativeStickyAssignor",
- "classic, org.apache.kafka.clients.consumer.RangeAssignor"
- ))
- def testRebalanceAndRejoin(groupProtocol: String, assignmentStrategy: String): Unit = {
- // create 2 consumers
- this.consumerConfig.setProperty(ConsumerConfig.GROUP_PROTOCOL_CONFIG, groupProtocol)
- this.consumerConfig.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "rebalance-and-rejoin-group")
- this.consumerConfig.setProperty(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, assignmentStrategy)
- this.consumerConfig.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")
- val consumer1 = createConsumer()
- val consumer2 = createConsumer()
-
- // create a new topic, have 2 partitions
- val topic = "topic1"
- val producer = createProducer()
- val expectedAssignment = createTopicAndSendRecords(producer, topic, 2, 100)
-
- assertEquals(0, consumer1.assignment().size)
- assertEquals(0, consumer2.assignment().size)
-
- val lock = new ReentrantLock()
- var generationId1 = -1
- var memberId1 = ""
- val customRebalanceListener = new ConsumerRebalanceListener {
- override def onPartitionsRevoked(partitions: util.Collection[TopicPartition]): Unit = {
- }
-
- override def onPartitionsAssigned(partitions: util.Collection[TopicPartition]): Unit = {
- if (!lock.tryLock(3000, TimeUnit.MILLISECONDS)) {
- fail(s"Time out while awaiting for lock.")
- }
- try {
- generationId1 = consumer1.groupMetadata().generationId()
- memberId1 = consumer1.groupMetadata().memberId()
- } finally {
- lock.unlock()
- }
- }
- }
- val consumerPoller1 = new ConsumerAssignmentPoller(consumer1, List(topic), Set.empty, customRebalanceListener)
- consumerPoller1.start()
- TestUtils.waitUntilTrue(() => consumerPoller1.consumerAssignment() == expectedAssignment,
- s"Timed out while awaiting expected assignment change to $expectedAssignment.")
-
- // Since the consumer1 already completed the rebalance,
- // the `onPartitionsAssigned` rebalance listener will be invoked to set the generationId and memberId
- var stableGeneration = -1
- var stableMemberId1 = ""
- if (!lock.tryLock(3000, TimeUnit.MILLISECONDS)) {
- fail(s"Time out while awaiting for lock.")
- }
- try {
- stableGeneration = generationId1
- stableMemberId1 = memberId1
- } finally {
- lock.unlock()
- }
-
- val consumerPoller2 = subscribeConsumerAndStartPolling(consumer2, List(topic))
- TestUtils.waitUntilTrue(() => consumerPoller1.consumerAssignment().size == 1,
- s"Timed out while awaiting expected assignment size change to 1.")
- TestUtils.waitUntilTrue(() => consumerPoller2.consumerAssignment().size == 1,
- s"Timed out while awaiting expected assignment size change to 1.")
-
- if (!lock.tryLock(3000, TimeUnit.MILLISECONDS)) {
- fail(s"Time out while awaiting for lock.")
- }
- try {
- if (assignmentStrategy.equals(classOf[CooperativeStickyAssignor].getName)) {
- // cooperative rebalance should rebalance twice before finally stable
- assertEquals(stableGeneration + 2, generationId1)
- } else {
- // eager rebalance should rebalance once before finally stable
- assertEquals(stableGeneration + 1, generationId1)
- }
- assertEquals(stableMemberId1, memberId1)
- } finally {
- lock.unlock()
- }
-
- consumerPoller1.shutdown()
- consumerPoller2.shutdown()
- }
-
-}