chia7712 commented on code in PR #16899:
URL: https://github.com/apache/kafka/pull/16899#discussion_r1742455301
##########
clients/src/test/java/org/apache/kafka/clients/consumer/ConsumerConfigTest.java:
##########
@@ -237,4 +238,26 @@ public void testProtocolConfigValidation(String protocol,
boolean isValid) {
assertThrows(ConfigException.class, () -> new
ConsumerConfig(configs));
}
}
+
+ @Test
+ public void testPartitionAssigmentStrategyWithConsumerGroupProtocol() {
+ final Map<String, Object> configs = new HashMap<>();
+ configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
keyDeserializerClass);
+ configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
valueDeserializerClass);
+ configs.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
"RoundRobinAssignor");
+ configs.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG,
GroupProtocol.CONSUMER.name());
+ ConfigException exception = assertThrows(ConfigException.class, () ->
new ConsumerConfig(configs));
+
assertTrue(exception.getMessage().contains(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG
+
+ " cannot be set when " + ConsumerConfig.GROUP_PROTOCOL_CONFIG
+ "=" + GroupProtocol.CONSUMER.name()));
+ }
+
+ @Test
+ public void
testSettingDefaultValuePartitionAssigmentStrategyWithConsumerGroupProtocol() {
+ final Map<String, Object> configs = new HashMap<>();
+ configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
keyDeserializerClass);
+ configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
valueDeserializerClass);
+ configs.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
"org.apache.kafka.clients.consumer.RangeAssignor,org.apache.kafka.clients.consumer.CooperativeStickyAssignor");
Review Comment:
Could you please add following test cases?
```java
@Test
public void
testSettingDefaultValuePartitionAssigmentStrategyWithConsumerGroupProtocol() {
testSettingDefaultValuePartitionAssigmentStrategyWithConsumerGroupProtocol("org.apache.kafka.clients.consumer.RangeAssignor,org.apache.kafka.clients.consumer.CooperativeStickyAssignor");
testSettingDefaultValuePartitionAssigmentStrategyWithConsumerGroupProtocol("org.apache.kafka.clients.consumer.CooperativeStickyAssignor,
org.apache.kafka.clients.consumer.RangeAssignor");
testSettingDefaultValuePartitionAssigmentStrategyWithConsumerGroupProtocol(Arrays.asList("org.apache.kafka.clients.consumer.RangeAssignor",
CooperativeStickyAssignor.class));
testSettingDefaultValuePartitionAssigmentStrategyWithConsumerGroupProtocol(Arrays.asList(CooperativeStickyAssignor.class,
"org.apache.kafka.clients.consumer.RangeAssignor"));
testSettingDefaultValuePartitionAssigmentStrategyWithConsumerGroupProtocol(Arrays.asList(RangeAssignor.class,
"org.apache.kafka.clients.consumer.CooperativeStickyAssignor"));
testSettingDefaultValuePartitionAssigmentStrategyWithConsumerGroupProtocol(Arrays.asList("org.apache.kafka.clients.consumer.CooperativeStickyAssignor",
RangeAssignor.class));
testSettingDefaultValuePartitionAssigmentStrategyWithConsumerGroupProtocol(Arrays.asList(CooperativeStickyAssignor.class,
RangeAssignor.class));
testSettingDefaultValuePartitionAssigmentStrategyWithConsumerGroupProtocol(Arrays.asList(RangeAssignor.class,
CooperativeStickyAssignor.class));
}
private void
testSettingDefaultValuePartitionAssigmentStrategyWithConsumerGroupProtocol(Object
value) {
final Map<String, Object> configs = new HashMap<>();
configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
keyDeserializerClass);
configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
valueDeserializerClass);
configs.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
value);
configs.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG,
GroupProtocol.CONSUMER.name());
assertDoesNotThrow(() -> new ConsumerConfig(configs));
}
```
--
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]
