Re: [PR] KAFKA-17338 ConsumerConfig should prevent using partition assignors with CONSUMER group protocol [kafka]

[via GitHub]

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: jira-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org