Skip to content

Commit da07836

Browse files
authored
KAFKA-20843 Allow custom consumer group assignors to override built-in assignor names (#23473)
#22982 added a check that rejects a custom assignor reusing a built-in name (e.g. uniform, range). This is a behavior change: before 4.4 such a configuration was accepted and intentionally used to replace a built-in assignor. Given the concerns raised in review and the upcoming 4.4 RC, this patch removes that check to restore the previous behavior. The duplicate name check is kept, so two configured assignors resolving to the same name still fail at startup. The streams group check is unchanged since group.streams.assignors is new in 4.4. Whether the strict rule should return can be revisited for 4.5. Reviewers: Chia-Ping Tsai <chia7712@gmail.com>, David Jacot <david.jacot@gmail.com>
1 parent 3258362 commit da07836

3 files changed

Lines changed: 15 additions & 27 deletions

File tree

‎docs/getting-started/upgrade.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,7 @@ type: docs
5656
* Controllers can now be unregistered from the cluster metadata. A new `kafka-cluster.sh unregister-controller` command and a `--unregister` flag on `kafka-metadata-quorum.sh remove-controller` are provided, backed by the new `Admin#unregisterController` API and the new `UnregisterController` RPC. This introduces the error code `CONTROLLER_ID_NOT_REGISTERED` (136) and requires metadata version `4.4-IV2` (`IBP_4_4_IV2`). For further details, please refer to [KIP-1312](https://cwiki.apache.org/confluence/spaces/KAFKA/pages/406623954/KIP-1312+Support+unregistering+controllers).
5757
* A new client configuration, `bootstrap.resolve.timeout.ms`, controls how DNS resolution of `bootstrap.servers` is performed. The default value of `0` preserves the historical synchronous behavior. Setting a positive value opts into asynchronous bootstrap resolution: the client constructor returns immediately, `NetworkClient` retries DNS resolution up to the configured timeout, and failures are surfaced later as `BootstrapResolutionExceptions` from subsequent API calls. This failure is unrecoverable: once raised, the exception is re-thrown by every API call until the client is closed and re-created after fixing the underlying DNS or `bootstrap.servers` configuration. Both asynchronous bootstrap resolution (enabled by setting `bootstrap.resolve.timeout.ms` to a positive value) and `BootstrapResolutionException` are evolving features and may undergo compatibility-breaking changes in a minor release. For further details, please refer to [KIP-909](https://cwiki.apache.org/confluence/x/MJtbDg). Note that Kafka Streams and Kafka Connect do not support asynchronous bootstrap resolution yet: for every client they create internally, `bootstrap.resolve.timeout.ms` is forced to `0` and a user-supplied positive value is ignored with a warning.
5858
* For tiered-storage topics, time-based local retention (`local.retention.ms`) now uses a segment's last modified time as the age anchor when the segment's largest record timestamp is in the future. Previously such a segment was never removed by time-based retention (its future timestamp kept it perpetually within the retention window) and was only reclaimed once `local.retention.bytes` was breached, even though it had already been uploaded to remote storage. For further details, please refer to [KAFKA-20609](https://issues.apache.org/jira/browse/KAFKA-20609).
59-
* `group.consumer.assignors` no longer allows a custom assignor to reuse the name of a built-in assignor (e.g. `uniform`, `range`), even when it is configured on its own to intentionally replace the built-in's behavior. Such a configuration used to be accepted and now fails broker startup with a `ConfigException`. For further details, please refer to [KAFKA-20843](https://issues.apache.org/jira/browse/KAFKA-20843).
59+
* `group.consumer.assignors` now fails broker startup with a `ConfigException` when two configured assignors resolve to the same name, for example a built-in listed both by its name and by its class name, or a custom assignor reusing the name of another configured assignor. Such a configuration used to pass startup validation and fail later during group coordinator loading. For further details, please refer to [KAFKA-20843](https://issues.apache.org/jira/browse/KAFKA-20843).
6060

6161
## Upgrading to 4.3.1
6262

‎group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java‎

Lines changed: 2 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -883,15 +883,9 @@ public static GroupCoordinatorConfig fromProps(
883883
protected List<ConsumerGroupPartitionAssignor> consumerGroupAssignors(
884884
AbstractConfig config
885885
) {
886-
Map<String, ConsumerGroupPartitionAssignor> builtInAssignors = CONSUMER_GROUP_BUILTIN_ASSIGNORS
886+
Map<String, ConsumerGroupPartitionAssignor> defaultAssignors = CONSUMER_GROUP_BUILTIN_ASSIGNORS
887887
.stream()
888888
.collect(Collectors.toMap(ConsumerGroupPartitionAssignor::name, Function.identity()));
889-
// A built-in may be configured either by its name or by its class name, so it is recognised
890-
// by class rather than by how it was resolved below.
891-
Set<Class<? extends ConsumerGroupPartitionAssignor>> builtInAssignorClasses = CONSUMER_GROUP_BUILTIN_ASSIGNORS
892-
.stream()
893-
.map(ConsumerGroupPartitionAssignor::getClass)
894-
.collect(Collectors.toSet());
895889

896890
List<ConsumerGroupPartitionAssignor> assignors = new ArrayList<>();
897891
Set<String> assignorNames = new HashSet<>();
@@ -900,7 +894,7 @@ protected List<ConsumerGroupPartitionAssignor> consumerGroupAssignors(
900894
// `configuredAssignor` is either the name of a built-in assignor,
901895
// or a fully qualified class name of a custom assignor
902896
for (String configuredAssignor : config.getList(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG)) {
903-
ConsumerGroupPartitionAssignor assignor = builtInAssignors.get(configuredAssignor);
897+
ConsumerGroupPartitionAssignor assignor = defaultAssignors.get(configuredAssignor);
904898
if (assignor == null) {
905899
try {
906900
assignor = Utils.newInstance(configuredAssignor, ConsumerGroupPartitionAssignor.class);
@@ -919,12 +913,6 @@ protected List<ConsumerGroupPartitionAssignor> consumerGroupAssignors(
919913

920914
assignors.add(assignor);
921915

922-
if (!builtInAssignorClasses.contains(assignor.getClass()) && builtInAssignors.containsKey(assignor.name())) {
923-
throw new ConfigException(CONSUMER_GROUP_ASSIGNORS_CONFIG, configuredAssignor,
924-
"Assignor name '" + assignor.name() + "' is reserved by a built-in assignor. " +
925-
"A custom assignor must not reuse the name of a built-in assignor");
926-
}
927-
928916
if (!assignorNames.add(assignor.name())) {
929917
throw new ConfigException(CONSUMER_GROUP_ASSIGNORS_CONFIG, configuredAssignor,
930918
"Assignor name '" + assignor.name() + "' is already registered by another configured assignor. " +

‎group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfigTest.java‎

Lines changed: 12 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -102,7 +102,7 @@ public GroupAssignment assign(
102102
public static class UniformNamedAssignor implements ConsumerGroupPartitionAssignor {
103103
@Override
104104
public String name() {
105-
// Collides with the built-in "uniform" assignor.
105+
// Overrides the built-in "uniform" assignor.
106106
return "uniform";
107107
}
108108

@@ -207,28 +207,28 @@ public void testConsumerGroupAssignorsWithDuplicateNamesFails() {
207207
}
208208

209209
@Test
210-
public void testConsumerGroupAssignorsWithReservedBuiltinNameFails() {
211-
// A custom assignor must not take the name of a built-in, whether or not the built-in is
212-
// itself configured: a member selecting that name would otherwise silently get the custom one.
210+
public void testConsumerGroupAssignorsOverridingBuiltinName() {
211+
// A custom assignor may reuse the name of a built-in; members selecting that name get the custom one.
213212
Map<String, Object> configs = new HashMap<>();
214213
configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, UniformNamedAssignor.class.getName());
215-
assertEquals("Invalid value " + UniformNamedAssignor.class.getName() +
216-
" for configuration group.consumer.assignors: Assignor name 'uniform' is reserved by a " +
217-
"built-in assignor. A custom assignor must not reuse the name of a built-in assignor",
218-
assertThrows(ConfigException.class, () -> createConfig(configs)).getMessage());
214+
GroupCoordinatorConfig config = createConfig(configs);
215+
List<ConsumerGroupPartitionAssignor> assignors = config.consumerGroupAssignors();
216+
assertEquals(1, assignors.size());
217+
assertInstanceOf(UniformNamedAssignor.class, assignors.get(0));
218+
assertEquals("uniform", assignors.get(0).name());
219219

220+
// Configuring the built-in alongside the custom assignor overriding its name still fails the duplicate check.
220221
configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG,
221222
List.of("uniform", UniformNamedAssignor.class.getName()));
222223
assertEquals("Invalid value " + UniformNamedAssignor.class.getName() +
223-
" for configuration group.consumer.assignors: Assignor name 'uniform' is reserved by a " +
224-
"built-in assignor. A custom assignor must not reuse the name of a built-in assignor",
224+
" for configuration group.consumer.assignors: Assignor name 'uniform' is already " +
225+
"registered by another configured assignor. Assignor names, whether built-in or custom, must be unique",
225226
assertThrows(ConfigException.class, () -> createConfig(configs)).getMessage());
226227
}
227228

228229
@Test
229230
public void testConsumerGroupAssignorsBuiltinByClassName() {
230-
// A built-in may also be configured by its class name, so the reserved-name check must
231-
// recognise it by class rather than by name.
231+
// A built-in may also be configured by its class name and resolves to the built-in itself.
232232
Map<String, Object> configs = new HashMap<>();
233233
configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG,
234234
List.of(UniformAssignor.class.getName(), RangeAssignor.class.getName()));

0 commit comments

Comments
 (0)