KAFKA-13864: provide the construct interceptor for KafkaProducer and KafkaConsumer

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

@@ -663,12 +663,38 @@ public KafkaConsumer(Properties properties,
     public KafkaConsumer(Map<String, Object> configs,
                          Deserializer<K> keyDeserializer,
                          Deserializer<V> valueDeserializer) {
+        this(configs, keyDeserializer, valueDeserializer, null);
+    }
+
+    /**
+     * A consumer is instantiated by providing a set of key-value pairs as configuration, and a key and a value {@link Deserializer}.
+     * <p>
+     * Valid configuration strings are documented at {@link ConsumerConfig}.
+     * <p>
+     * Note: after creating a {@code KafkaConsumer} you must always {@link #close()} it to avoid resource leaks.
+     *
+     * @param configs The consumer configs
+     * @param keyDeserializer The deserializer for key that implements {@link Deserializer}. The configure() method
+     *            won't be called in the consumer when the deserializer is passed in directly.
+     * @param valueDeserializer The deserializer for value that implements {@link Deserializer}. The configure() method
+     *            won't be called in the consumer when the deserializer is passed in directly.
+     * @param interceptors The list interceptors for consumer that implements {$link ConsumerInterceptor}.
+     */
+    public KafkaConsumer(Map<String, Object> configs,
+                         Deserializer<K> keyDeserializer,
+                         Deserializer<V> valueDeserializer,
+                         List<ConsumerInterceptor<K, V>> interceptors) {
         this(new ConsumerConfig(ConsumerConfig.appendDeserializerToConfig(configs, keyDeserializer, valueDeserializer)),
-                keyDeserializer, valueDeserializer);
+                keyDeserializer, valueDeserializer, interceptors);
     }
 
-    @SuppressWarnings("unchecked")
     KafkaConsumer(ConsumerConfig config, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer) {
+        this(config, keyDeserializer, valueDeserializer, null);
+    }
+
+    @SuppressWarnings("unchecked")
+    KafkaConsumer(ConsumerConfig config, Deserializer<K> keyDeserializer,
+                  Deserializer<V> valueDeserializer, List<ConsumerInterceptor<K, V>> interceptors) {
         try {
             GroupRebalanceConfig groupRebalanceConfig = new GroupRebalanceConfig(config,
                     GroupRebalanceConfig.ProtocolType.CONSUMER);
@@ -705,6 +731,9 @@ public KafkaConsumer(Map<String, Object> configs,
                     ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG,
                     ConsumerInterceptor.class,
                     Collections.singletonMap(ConsumerConfig.CLIENT_ID_CONFIG, clientId));
+            if (interceptors != null && !interceptors.isEmpty()) {
+                interceptorList.addAll(interceptors);
+            }
             this.interceptors = new ConsumerInterceptors<>(interceptorList);
             if (keyDeserializer == null) {
                 this.keyDeserializer = config.getConfiguredInstance(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, Deserializer.class);

clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java

@@ -290,6 +290,27 @@ public KafkaProducer(Map<String, Object> configs, Serializer<K> keySerializer, S
                 keySerializer, valueSerializer, null, null, null, Time.SYSTEM);
     }
 
+    /**
+     * A producer is instantiated by providing a set of key-value pairs as configuration, a key and a value {@link Serializer}.
+     * Valid configuration strings are documented <a href="http://kafka.apache.org/documentation.html#producerconfigs">here</a>.
+     * Values can be either strings or Objects of the appropriate type (for example a numeric configuration would accept
+     * either the string "42" or the integer 42).
+     * <p>
+     * Note: after creating a {@code KafkaProducer} you must always {@link #close()} it to avoid resource leaks.
+     * @param configs   The producer configs
+     * @param keySerializer  The serializer for key that implements {@link Serializer}. The configure() method won't be
+     *                       called in the producer when the serializer is passed in directly.
+     * @param valueSerializer  The serializer for value that implements {@link Serializer}. The configure() method won't
+     *                         be called in the producer when the serializer is passed in directly.
+     * @param interceptors The list interceptors for producer that implements {$link ProducerInterceptor}.
+     */
+    public KafkaProducer(Map<String, Object> configs, Serializer<K> keySerializer, Serializer<V> valueSerializer,
+                         List<ProducerInterceptor<K, V>> interceptors) {
+        this(new ProducerConfig(ProducerConfig.appendSerializerToConfig(configs, keySerializer, valueSerializer)),
+                keySerializer, valueSerializer, null, null,
+                new ProducerInterceptors<>(interceptors), Time.SYSTEM);
+    }
+
     /**
      * A producer is instantiated by providing a set of key-value pairs as configuration. Valid configuration strings
      * are documented <a href="http://kafka.apache.org/documentation.html#producerconfigs">here</a>.

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

@@ -333,6 +333,17 @@ public void testInvalidSocketReceiveBufferSize() {
             () -> new KafkaConsumer<>(config, new ByteArrayDeserializer(), new ByteArrayDeserializer()));
     }
 
+    @Test
+    public void testConstructConsumerWithInterceptor() {
+        Map<String, Object> config = new HashMap<>();
+        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
+        config.put(ConsumerConfig.RECEIVE_BUFFER_CONFIG, -2);
+        List<ConsumerInterceptor<String, String>> interceptors = new ArrayList<>();
+        interceptors.add(new MockConsumerInterceptor());
+        assertThrows(KafkaException.class,
+                () -> new KafkaConsumer<>(config, new StringDeserializer(), new StringDeserializer(), interceptors));
+    }
+
     @Test
     public void shouldIgnoreGroupInstanceIdForEmptyGroupId() {
         Map<String, Object> config = new HashMap<>();

clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java

@@ -522,6 +522,31 @@ public void testSerializerClose() {
         assertEquals(oldCloseCount + 2, MockSerializer.CLOSE_COUNT.get());
     }
 
+    @Test
+    public void testConstructKafkaProducerWithInterceptor() {
+        try {
+            Map<String, Object> configs = new HashMap<>();
+            configs.put(ProducerConfig.CLIENT_ID_CONFIG, "testConstructKafkaProducerWithInterceptor");
+            configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
+
+            List<ProducerInterceptor<String, String>> interceptors = new ArrayList<>();
+            interceptors.add(new MockProducerInterceptor());
+            KafkaProducer<String, String> producer = new KafkaProducer<>(
+                    configs, new StringSerializer(), new StringSerializer(), interceptors);
+            assertEquals(1, MockProducerInterceptor.INIT_COUNT.get());
+            assertEquals(0, MockProducerInterceptor.CLOSE_COUNT.get());
+
+            assertNull(MockProducerInterceptor.CLUSTER_META.get());
+
+            producer.close();
+            assertEquals(1, MockProducerInterceptor.INIT_COUNT.get());
+            assertEquals(1, MockProducerInterceptor.CLOSE_COUNT.get());
+        } finally {
+            // cleanup since we are using mutable static variables in MockProducerInterceptor
+            MockProducerInterceptor.resetCounters();
+        }
+    }
+
     @Test
     public void testInterceptorConstructClose() {
         try {