You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by jy...@apache.org on 2026/09/11 15:03:45 UTC
(kafka) branch 4.3 updated: MINOR: Align consumer assignor loading with share and streams (#22929)
This is an automated email from the ASF dual-hosted git repository.
m1a2st pushed a commit to branch 4.3
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.3 by this push:
new 4c2e5ae3c58 MINOR: Align consumer assignor loading with share and streams (#22929)
4c2e5ae3c58 is described below
commit 4c2e5ae3c580903fcc57024cfa2c7c0b885d0e8a
Author: Matthias J. Sax <ma...@confluent.io>
AuthorDate: Mon Jul 27 22:35:26 2026 -0700
MINOR: Align consumer assignor loading with share and streams (#22929)
group.consumer.assignors is a broker config, and thus its type is always
String. The class loader nevertheless accepted Class objects
unnecessarily.
This PR drops the Class branch to simplify the code. Additionally, this
PR improves some error messages.
Reviewers: Andrew Schofield <as...@confluent.io>
---
.../coordinator/group/GroupCoordinatorConfig.java | 57 +++++++++--------
.../group/GroupCoordinatorConfigTest.java | 74 +++++++++++++++-------
2 files changed, 83 insertions(+), 48 deletions(-)
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java
index b3c25dbfb88..372311a7df8 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java
@@ -20,6 +20,7 @@ import org.apache.kafka.common.Configurable;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.config.AbstractConfig;
import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.record.internal.CompressionType;
import org.apache.kafka.common.record.internal.Records;
import org.apache.kafka.common.utils.Utils;
@@ -756,28 +757,24 @@ public class GroupCoordinatorConfig {
List<ConsumerGroupPartitionAssignor> assignors = new ArrayList<>();
try {
- for (Object object : config.getList(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG)) {
- ConsumerGroupPartitionAssignor assignor;
-
- if (object instanceof String klass) {
- assignor = defaultAssignors.get(klass);
- if (assignor == null) {
- try {
- assignor = Utils.newInstance(klass, ConsumerGroupPartitionAssignor.class);
- } catch (ClassNotFoundException e) {
- throw new KafkaException("Class " + klass + " cannot be found", e);
- } catch (ClassCastException e) {
- throw new KafkaException(klass + " is not an instance of " + ConsumerGroupPartitionAssignor.class.getName());
- }
- }
- } else if (object instanceof Class<?> klass) {
- Object o = Utils.newInstance((Class<?>) klass);
- if (!(o instanceof ConsumerGroupPartitionAssignor)) {
- throw new KafkaException(klass + " is not an instance of " + ConsumerGroupPartitionAssignor.class.getName());
+ // `configuredAssignor` is either the name of a built-in assignor,
+ // or a fully qualified class name of a custom assignor
+ for (String configuredAssignor : config.getList(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG)) {
+ ConsumerGroupPartitionAssignor assignor = defaultAssignors.get(configuredAssignor);
+ if (assignor == null) {
+ try {
+ assignor = Utils.newInstance(configuredAssignor, ConsumerGroupPartitionAssignor.class);
+ } catch (ClassNotFoundException e) {
+ throw new ConfigException(CONSUMER_GROUP_ASSIGNORS_CONFIG, configuredAssignor,
+ "Class cannot be found");
+ } catch (ClassCastException e) {
+ throw new ConfigException(CONSUMER_GROUP_ASSIGNORS_CONFIG, configuredAssignor,
+ "Class is not an instance of " + ConsumerGroupPartitionAssignor.class.getName());
+ } catch (KafkaException e) {
+ // Utils#newInstance reports instantiation failures, for example a missing
+ // public no-argument constructor, without naming the config that caused them.
+ throw new ConfigException(CONSUMER_GROUP_ASSIGNORS_CONFIG, configuredAssignor, e.getMessage());
}
- assignor = (ConsumerGroupPartitionAssignor) o;
- } else {
- throw new KafkaException("Unexpected element of type " + object.getClass().getName() + ", expected String or Class");
}
assignors.add(assignor);
@@ -802,16 +799,24 @@ public class GroupCoordinatorConfig {
List<ShareGroupPartitionAssignor> assignors = new ArrayList<>();
try {
- for (String kclass : config.getList(GroupCoordinatorConfig.SHARE_GROUP_ASSIGNORS_CONFIG)) {
+ // `configuredAssignor` is either the name of a built-in assignor,
+ // or a fully qualified class name of a custom assignor
+ for (String configuredAssignor : config.getList(GroupCoordinatorConfig.SHARE_GROUP_ASSIGNORS_CONFIG)) {
ShareGroupPartitionAssignor assignor = SHARE_GROUP_BUILTIN_ASSIGNOR;
- if (!Objects.equals(kclass, SHARE_GROUP_ASSIGNORS_DEFAULT)) {
+ if (!Objects.equals(configuredAssignor, SHARE_GROUP_ASSIGNORS_DEFAULT)) {
try {
- assignor = Utils.newInstance(kclass, ShareGroupPartitionAssignor.class);
+ assignor = Utils.newInstance(configuredAssignor, ShareGroupPartitionAssignor.class);
} catch (ClassNotFoundException e) {
- throw new KafkaException("Class " + kclass + " cannot be found", e);
+ throw new ConfigException(SHARE_GROUP_ASSIGNORS_CONFIG, configuredAssignor,
+ "Class cannot be found");
} catch (ClassCastException e) {
- throw new KafkaException(kclass + " is not an instance of " + ShareGroupPartitionAssignor.class.getName());
+ throw new ConfigException(SHARE_GROUP_ASSIGNORS_CONFIG, configuredAssignor,
+ "Class is not an instance of " + ShareGroupPartitionAssignor.class.getName());
+ } catch (KafkaException e) {
+ // Utils#newInstance reports instantiation failures, for example a missing
+ // public no-argument constructor, without naming the config that caused them.
+ throw new ConfigException(SHARE_GROUP_ASSIGNORS_CONFIG, configuredAssignor, e.getMessage());
}
}
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfigTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfigTest.java
index 3444e7cd7c6..67a229b151f 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfigTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfigTest.java
@@ -17,7 +17,6 @@
package org.apache.kafka.coordinator.group;
import org.apache.kafka.common.Configurable;
-import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.config.AbstractConfig;
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.record.internal.CompressionType;
@@ -68,6 +67,24 @@ public class GroupCoordinatorConfigTest {
}
}
+ public static class NoDefaultConstructorAssignor implements ConsumerGroupPartitionAssignor, ShareGroupPartitionAssignor {
+ public NoDefaultConstructorAssignor(String unused) {
+ }
+
+ @Override
+ public String name() {
+ return "NoDefaultConstructorAssignor";
+ }
+
+ @Override
+ public GroupAssignment assign(
+ GroupSpec groupSpec,
+ SubscribedTopicDescriber subscribedTopicDescriber
+ ) throws PartitionAssignorException {
+ return null;
+ }
+ }
+
@Test
public void testConsumerGroupAssignorFullClassNames() {
// The full class name of the assignors is part of our public api. Hence,
@@ -104,15 +121,7 @@ public class GroupCoordinatorConfigTest {
assertInstanceOf(CustomAssignor.class, assignors.get(0));
assertNotNull(((CustomAssignor) assignors.get(0)).configs);
- // Test with classes.
- configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(RangeAssignor.class, CustomAssignor.class));
- config = createConfig(configs);
- assignors = config.consumerGroupAssignors();
- assertEquals(2, assignors.size());
- assertInstanceOf(RangeAssignor.class, assignors.get(0));
- assertInstanceOf(CustomAssignor.class, assignors.get(1));
-
- // Test combination of short name and class.
+ // Test combination of short name and class name as a comma-separated string.
configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, "uniform, " + CustomAssignor.class.getName());
config = createConfig(configs);
assignors = config.consumerGroupAssignors();
@@ -120,7 +129,7 @@ public class GroupCoordinatorConfigTest {
assertInstanceOf(UniformAssignor.class, assignors.get(0));
assertInstanceOf(CustomAssignor.class, assignors.get(1));
- // Test combination of short name and class.
+ // Test combination of short name and class name as a list of strings.
configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of("uniform", CustomAssignor.class.getName()));
config = createConfig(configs);
assignors = config.consumerGroupAssignors();
@@ -169,6 +178,24 @@ public class GroupCoordinatorConfigTest {
configs.put(GroupCoordinatorConfig.SHARE_GROUP_ASSIGNORS_CONFIG, "simple, " + CustomAssignor.class.getName());
assertEquals("group.share.assignors must contain exactly one assignor, but found 2",
assertThrows(IllegalArgumentException.class, () -> createConfig(configs)).getMessage());
+
+ // Test unknown class.
+ configs.put(GroupCoordinatorConfig.SHARE_GROUP_ASSIGNORS_CONFIG, "foo");
+ assertEquals("Invalid value foo for configuration group.share.assignors: Class cannot be found",
+ assertThrows(ConfigException.class, () -> createConfig(configs)).getMessage());
+
+ // Test class that is not an assignor.
+ configs.put(GroupCoordinatorConfig.SHARE_GROUP_ASSIGNORS_CONFIG, Object.class.getName());
+ assertEquals("Invalid value java.lang.Object for configuration group.share.assignors: " +
+ "Class is not an instance of org.apache.kafka.coordinator.group.api.assignor.ShareGroupPartitionAssignor",
+ assertThrows(ConfigException.class, () -> createConfig(configs)).getMessage());
+
+ // Test class that cannot be instantiated.
+ configs.put(GroupCoordinatorConfig.SHARE_GROUP_ASSIGNORS_CONFIG, NoDefaultConstructorAssignor.class.getName());
+ assertEquals("Invalid value " + NoDefaultConstructorAssignor.class.getName() +
+ " for configuration group.share.assignors: Could not find a public no-argument constructor for " +
+ NoDefaultConstructorAssignor.class.getName(),
+ assertThrows(ConfigException.class, () -> createConfig(configs)).getMessage());
}
@Test
@@ -180,7 +207,7 @@ public class GroupCoordinatorConfigTest {
configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_SESSION_TIMEOUT_MS_CONFIG, 555);
configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_HEARTBEAT_INTERVAL_MS_CONFIG, 200);
configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_MAX_SIZE_CONFIG, 55);
- configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(RangeAssignor.class));
+ configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(RangeAssignor.class.getName()));
configs.put(GroupCoordinatorConfig.OFFSETS_TOPIC_SEGMENT_BYTES_CONFIG, 2222);
configs.put(GroupCoordinatorConfig.OFFSET_METADATA_MAX_SIZE_CONFIG, 3333);
configs.put(GroupCoordinatorConfig.GROUP_MAX_SIZE_CONFIG, 60);
@@ -303,20 +330,23 @@ public class GroupCoordinatorConfigTest {
assertEquals("Invalid value class java.lang.Object for configuration group.consumer.assignors: Expected a comma separated list.",
assertThrows(ConfigException.class, () -> createConfig(configs)).getMessage());
- configs.clear();
- configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(Object.class));
- assertEquals("class java.lang.Object is not an instance of org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor",
- assertThrows(KafkaException.class, () -> createConfig(configs)).getMessage());
-
configs.clear();
configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, Object.class.getName());
- assertEquals("java.lang.Object is not an instance of org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor",
- assertThrows(KafkaException.class, () -> createConfig(configs)).getMessage());
+ assertEquals("Invalid value java.lang.Object for configuration group.consumer.assignors: " +
+ "Class is not an instance of org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor",
+ assertThrows(ConfigException.class, () -> createConfig(configs)).getMessage());
configs.clear();
configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, "foo");
- assertEquals("Class foo cannot be found",
- assertThrows(KafkaException.class, () -> createConfig(configs)).getMessage());
+ assertEquals("Invalid value foo for configuration group.consumer.assignors: Class cannot be found",
+ assertThrows(ConfigException.class, () -> createConfig(configs)).getMessage());
+
+ configs.clear();
+ configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, NoDefaultConstructorAssignor.class.getName());
+ assertEquals("Invalid value " + NoDefaultConstructorAssignor.class.getName() +
+ " for configuration group.consumer.assignors: Could not find a public no-argument constructor for " +
+ NoDefaultConstructorAssignor.class.getName(),
+ assertThrows(ConfigException.class, () -> createConfig(configs)).getMessage());
configs.clear();
configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, "foobar");
@@ -624,7 +654,7 @@ public class GroupCoordinatorConfigTest {
configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_HEARTBEAT_INTERVAL_MS_CONFIG, 5);
configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_MIN_HEARTBEAT_INTERVAL_MS_CONFIG, 5);
configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_MAX_SIZE_CONFIG, Integer.MAX_VALUE);
- configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(RangeAssignor.class));
+ configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(RangeAssignor.class.getName()));
configs.put(GroupCoordinatorConfig.OFFSETS_TOPIC_SEGMENT_BYTES_CONFIG, 1000);
configs.put(GroupCoordinatorConfig.OFFSET_METADATA_MAX_SIZE_CONFIG, offsetMetadataMaxSize);
configs.put(GroupCoordinatorConfig.GROUP_MAX_SIZE_CONFIG, Integer.MAX_VALUE);