This is an automated email from the ASF dual-hosted git repository.
mjsax 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 ec89db257f0 MINOR: Align consumer assignor loading with share and
streams (#22929)
ec89db257f0 is described below
commit ec89db257f0ebe3ba80542df30ebefcec26eca7a
Author: Matthias J. Sax <[email protected]>
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 <[email protected]>
---
.../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 35ec327c31e..0501a7c9e9c 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;
@@ -859,28 +860,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);
@@ -905,16 +902,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 297019d70f1..c53c297d22e 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;
@@ -73,6 +72,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,
@@ -109,15 +126,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();
@@ -125,7 +134,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();
@@ -174,6 +183,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
@@ -185,7 +212,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);
@@ -314,20 +341,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");
@@ -721,7 +751,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);
