MINOR: add docs to remind reader that impl of ConsumerPartitionAssign… (

apache#13659)

Reviewers: David Jacot <[email protected]>, Kirk True <[email protected]>

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerPartitionAssignor.java

@@ -39,11 +39,13 @@
  * as the group coordinator. The coordinator selects one member to perform the group assignment and
  * propagates the subscriptions of all members to it. Then {@link #assign(Cluster, GroupSubscription)} is called
  * to perform the assignment and the results are forwarded back to each respective members
- *
+ * <p>
  * In some cases, it is useful to forward additional metadata to the assignor in order to make
  * assignment decisions. For this, you can override {@link #subscriptionUserData(Set)} and provide custom
  * userData in the returned Subscription. For example, to have a rack-aware assignor, an implementation
  * can use this user data to forward the rackId belonging to each member.
+ * <p>
+ * The implementation can extend {@link Configurable} to get configs from consumer.
  */
 public interface ConsumerPartitionAssignor {
 

clients/src/test/java/org/apache/kafka/clients/consumer/ConsumerPartitionAssignorTest.java

@@ -18,18 +18,22 @@
 
 
 import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Configurable;
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.junit.jupiter.api.Test;
 
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.Properties;
 
 import static org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG;
 import static org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG;
 import static org.apache.kafka.clients.consumer.ConsumerPartitionAssignor.getAssignorInstances;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
@@ -108,7 +112,21 @@ public void shouldThrowKafkaExceptionOnAssignorsWithSameName() {
         ));
     }
 
-    public static class TestConsumerPartitionAssignor implements ConsumerPartitionAssignor {
+    @Test
+    public void shouldBeConfigurable() {
+        Map<String, Object> configs = Collections.singletonMap("key", "value");
+        List<ConsumerPartitionAssignor> assignors = getAssignorInstances(
+            Collections.singletonList(TestConsumerPartitionAssignor.class.getName()),
+            configs
+        );
+        assertEquals(1, assignors.size());
+        assertInstanceOf(TestConsumerPartitionAssignor.class, assignors.get(0));
+        assertEquals(configs, ((TestConsumerPartitionAssignor) assignors.get(0)).configs);
+    }
+
+
+    public static class TestConsumerPartitionAssignor implements ConsumerPartitionAssignor, Configurable {
+        private Map<String, ?> configs = null;
 
         @Override
         public GroupAssignment assign(Cluster metadata, GroupSubscription groupSubscription) {
@@ -120,6 +138,11 @@ public String name() {
             // use the RangeAssignor's name to cause naming conflict
             return new RangeAssignor().name();
         }
+
+        @Override
+        public void configure(Map<String, ?> configs) {
+            this.configs = configs;
+        }
     }
 
     private ConsumerConfig initConsumerConfigWithClassTypes(List<Object> classTypes) {