lianetm commented on code in PR #16899:
URL: https://github.com/apache/kafka/pull/16899#discussion_r1857379008
##########
clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java:
##########
@@ -375,6 +379,22 @@ public class ConsumerConfig extends AbstractConfig {
private static final AtomicInteger CONSUMER_CLIENT_ID_SEQUENCE = new
AtomicInteger(1);
+ /**
+ * A list of configuration keys for CLASSIC protocol not supported.
Review Comment:
nit:
```suggestion
* A list of configuration keys not supported for CLASSIC protocol
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java:
##########
@@ -375,6 +379,22 @@ public class ConsumerConfig extends AbstractConfig {
private static final AtomicInteger CONSUMER_CLIENT_ID_SEQUENCE = new
AtomicInteger(1);
+ /**
+ * A list of configuration keys for CLASSIC protocol not supported.
+ */
+ private static final List<String> CLASSIC_PROTOCOL_UNSUPPORTED_CONFIGS =
Collections.singletonList(
+ GROUP_REMOTE_ASSIGNOR_CONFIG
+ );
+
+ /**
+ * A list of configuration keys for consumer protocol not supported.
Review Comment:
nit:
```suggestion
* A list of configuration keys not supported for CONSUMER protocol.
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java:
##########
@@ -666,7 +686,8 @@ protected Map<String, Object> postProcessParsedConfig(final
Map<String, Object>
Map<String, Object> refinedConfigs =
CommonClientConfigs.postProcessReconnectBackoffConfigs(this, parsedValues);
maybeOverrideClientId(refinedConfigs);
maybeOverrideEnableAutoCommit(refinedConfigs);
- checkGroupRemoteAssignor();
+ checkUnsupportedConfigs(GroupProtocol.CLASSIC,
CLASSIC_PROTOCOL_UNSUPPORTED_CONFIGS);
+ checkUnsupportedConfigs(GroupProtocol.CONSUMER,
CONSUMER_PROTOCOL_UNSUPPORTED_CONFIGS);
Review Comment:
could we simplify here and call `checkUnsupportedConfigs` once, with the
groupProtocol read from the config? (seems clearer to have a single call that
would encapsulate the logic to determine the `unsupportedConfigs` list to use
based on the protocol used in the config, which is only one, seems a bit
confusing how we check here for the 2 protocols). Makes sense?
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java:
##########
@@ -511,7 +509,6 @@ private Map<String, Object> composeConfigs(String groupId,
String groupProtocol,
configs.put(KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
configs.put(VALUE_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
configs.put(GROUP_PROTOCOL_CONFIG, groupProtocol);
- configs.put(PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
RangeAssignor.class.getName());
Review Comment:
should we explicitly set it as you did on
`DeleteOffsetsConsumerGroupCommandIntegrationTest`?
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ResetConsumerGroupOffsetTest.java:
##########
@@ -800,7 +798,6 @@ private Map<String, Object>
composeConsumerConfigs(ClusterInstance cluster,
configs.put(GROUP_PROTOCOL_CONFIG, groupProtocol.name);
configs.put(KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
configs.put(VALUE_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
- configs.put(PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
RangeAssignor.class.getName());
Review Comment:
should we explicitly set it as you did on
`DeleteOffsetsConsumerGroupCommandIntegrationTest`?
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
