joewitt commented on PR #11271:
URL: https://github.com/apache/nifi/pull/11271#issuecomment-5655831446
**Unsupported share-consumer configs (hand-copied deny list)**
**What was true before the PR**
`Kafka3ConnectionService` built one `consumerProperties` bag at enable time
and used it for classic `KafkaConsumer`. That bag always includes connectivity
(bootstrap, SASL/SSL, timeouts, `max.poll.records`) and **Transaction Isolation
Level** (`isolation.level`, default `read_committed`). Classic-only knobs such
as `auto.offset.reset` and `enable.auto.commit` are applied later, only on
`getConsumerService`. Users can also add Kafka consumer keys as dynamic
properties on the controller service.
Classic `KafkaConsumer` accepts those keys. Presence is not an error. NiFi
already ships one `kafka-clients` version per release (this tree: 4.3.1). Users
still run many broker versions; the default consume path is that classic
client. None of this required deleting keys before constructing the consumer.
**What is true with the PR**
Share Group reuses that same bag to construct `KafkaShareConsumer`.
KIP-932’s share client is a different type. In kafka-clients 4.3.1,
`ShareConsumerConfig` (package-private) walks a **private** `List<String>
SHARE_GROUP_UNSUPPORTED_CONFIGS` and throws `ConfigException` if **any of those
names is present**, regardless of value: `"<name> cannot be set when using a
share group."` The ten names today are:
`auto.offset.reset`, `enable.auto.commit`, `group.instance.id`,
`isolation.level`, `partition.assignment.strategy`, `interceptor.classes`,
`session.timeout.ms`, `heartbeat.interval.ms`, `group.protocol`,
`group.remote.assignor`
That check is new with share groups. It is client-side, before any broker
call. It is not a cluster rule that old brokers started enforcing.
Because the shared bag **always** contains `isolation.level`, a stock
connection service cannot build a share consumer unless that key is removed.
The other nine are not applied by `getShareConsumerService` itself; they appear
if someone set them as dynamic properties (or if a later CS property lands in
the bag). The PR copies all ten into `properties.remove(...)` and comments
“keep this list in sync with
`ShareConsumerConfig.SHARE_GROUP_UNSUPPORTED_CONFIGS`.” NiFi cannot reference
that field: the class is package-private and the list is private. The lists
match 4.3.1 **today**. They will not stay matched by compilation.
This does **not** couple the default ConsumeKafka path to a Kafka broker
version. `Group Type` still defaults to Consumer Group; mixed 3.x / 4.0 brokers
keep working as before. Share Group is opt-in and already documented as 4.1+ /
4.2+ GA brokers. The new coupling is NiFi **source** duplicating a private deny
list inside the **bundled** `kafka-clients` jar. A later NiFi kafka-clients
bump that adds a rejected key fails at processor start (`ConfigException`). A
bump that later *allows* a key we still strip means share users cannot set it
until NiFi stops removing it. `KafkaShareConsumer` is also `@Evolving`, so
share behavior can change on a client bump without a broker change.
**What I propose specifically and why**
Do not try to call Kafka’s constant from production code; it is not part of
the public API.
Keep the hardcoded `remove()`s (they are the only way to reuse the classic
bag). Pull them into a package-visible helper, e.g.
`stripShareUnsupportedConsumerConfigs(Properties)`, used by
`getShareConsumerService`.
Add a unit test that reads Kafka’s list by reflection and asserts every key
is gone after the helper runs:
```java
Class<?> clazz =
Class.forName("org.apache.kafka.clients.consumer.ShareConsumerConfig");
Field field = clazz.getDeclaredField("SHARE_GROUP_UNSUPPORTED_CONFIGS");
field.setAccessible(true);
@SuppressWarnings("unchecked")
List<String> kafkaUnsupported = (List<String>) field.get(null);
Properties properties = new Properties();
for (String key : kafkaUnsupported) {
properties.put(key, "placeholder");
}
stripShareUnsupportedConsumerConfigs(properties);
assertTrue(kafkaUnsupported.stream().noneMatch(properties::containsKey),
"Share consumer still carries keys Kafka rejects: " +
properties.keySet());
```
That test fails on the next kafka-clients bump that adds a rejected key,
instead of failing when someone enables Share Group. Over-stripping (we remove
a key Kafka later allows) is the remaining drift; that is acceptable compared
with constructor failure, and the same test makes the mismatch visible when the
Kafka list changes.
Also drop the “Kafka 4.2+” wording on this block. The reject list is defined
by the kafka-clients jar NiFi compiles against (4.3.1 here), not by the broker
version in the field.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
