Ziyun Fu created KAFKA-20843:
--------------------------------
Summary: Custom consumer group assignors may silently use
duplicate or built-in assignor names
Key: KAFKA-20843
URL: https://issues.apache.org/jira/browse/KAFKA-20843
Project: Kafka
Issue Type: Bug
Reporter: Ziyun Fu
Server-side consumer group assignors are configured through
`group.consumer.assignors` as a list of either built-in assignor names or
fully-qualified class names.
Each entry is resolved against the built-in name map first:
* `uniform` → `UniformAssignor`
* `range` → `RangeAssignor`
An entry is only treated as a class name if that lookup misses
(`GroupCoordinatorConfig#consumerGroupAssignors`,
`GroupCoordinatorConfig.java:861-891`).
Assignors are then addressed *by name* everywhere downstream:
* `GroupMetadataManager` builds a `name -> assignor` map with
`Collectors.toMap(ConsumerGroupPartitionAssignor::name, Function.identity())`
(`GroupMetadataManager.java:573-576`).
* `GroupCoordinatorService` validates
`ConsumerGroupHeartbeatRequest.serverAssignor` against the set of configured
assignor names (`GroupCoordinatorService.java:508`).
* Target assignment resolves the group’s preferred assignor with
`consumerGroupAssignors.get(preferredServerAssignor)`
(`GroupMetadataManager.java:4217`), where the name comes from the members’
persisted assignor preference.
### 1. Two assignors configured with the same `name()`
`verifyConsumerGroupConfigs()` only validates timeouts and intervals, and the
`ConfigDef.ValidList.anyNonDuplicateValues` validator only de-duplicates the
raw config entries. As a result, distinct entries that produce the same
`name()` pass configuration parsing and the broker starts normally.
The failure surfaces much later: the `Collectors.toMap` call above throws
`IllegalStateException: Duplicate key <name>` when a `GroupMetadataManager` is
constructed. That happens per coordinator shard while a `__consumer_offsets`
partition is loaded (`GroupCoordinatorShard.java:267`, reached from the
transition to `LOADING` in `CoordinatorRuntime.java:533-545`).
So the broker comes up healthy, but then fails to load group coordinator shards
with an error message that does not point at the misconfiguration.
This also covers configuring one built-in assignor twice under two spellings,
for example:
`group.consumer.assignors=uniform,org.apache.kafka.coordinator.group.assignor.UniformAssignor`
The two entries differ as strings, so the duplicate check in `ConfigDef` does
not catch them.
Because a name has to unambiguously identify one assignor, these configurations
should be rejected but currently are not.
### 2. A custom assignor whose `name()` equals a built-in assignor name
A custom assignor returning `"uniform"` or `"range"` from `name()` is accepted,
even when the built-in assignor it collides with is not configured at all.
This silently re-points the built-in name to the custom class:
* every client that sets `group.remote.assignor=uniform`
* every existing group whose persisted preferred assignor is `uniform`
would then be served by the custom implementation with no warning.
Conversely, because config entries are resolved against the built-in name map
first, the short name `uniform` can never select the custom class. So the same
name means two different things depending on how it is spelled in the config.
The client-side assignors for the classic protocol already guard against this:
`ConsumerPartitionAssignor#getAssignorInstances` throws a `KafkaException` when
two configured assignors report the same `name()`
(`ConsumerPartitionAssignor.java:439-444`).
The server side has no equivalent check.
```
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
