[
https://issues.apache.org/jira/browse/KAFKA-20843?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ziyun Fu updated KAFKA-20843:
-----------------------------
Description:
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.
```
was:
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.
```
> 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
> Assignee: Ziyun Fu
> Priority: Major
>
> 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)
