introduce interfaces fr delegating clients, use them in instrumented …

…clients (#136)

li-apache-kafka-clients/src/main/java/com/linkedin/kafka/clients/consumer/DelegatingConsumer.java

@@ -0,0 +1,12 @@
+/*
+ * Copyright 2019 LinkedIn Corp. Licensed under the BSD 2-Clause License (the "License").
 See License in the project root for license information.
+ */
+
+package com.linkedin.kafka.clients.consumer;
+
+import org.apache.kafka.clients.consumer.Consumer;
+
+
+public interface DelegatingConsumer<K, V> extends Consumer<K, V> {
+  Consumer<K, V> getDelegate();
+}

li-apache-kafka-clients/src/main/java/com/linkedin/kafka/clients/consumer/LiKafkaInstrumentedConsumerImpl.java

@@ -47,7 +47,7 @@
  * @param <K>
  * @param <V>
  */
-public class LiKafkaInstrumentedConsumerImpl<K, V> implements Consumer<K, V>, EventHandler {
+public class LiKafkaInstrumentedConsumerImpl<K, V> implements DelegatingConsumer<K, V>, EventHandler {
   private static final Logger LOG = LoggerFactory.getLogger(LiKafkaInstrumentedConsumerImpl.class);
 
   private final long initialConnectionTimeoutMs = TimeUnit.SECONDS.toMillis(30);
@@ -154,8 +154,8 @@ public void configChangeRequested(UUID commandId, Map<String, String> configDiff
     //TODO - respond to command UUID
   }
 
-  //package-private FOR TESTING
-  Consumer<K, V> getDelegate() {
+  @Override
+  public Consumer<K, V> getDelegate() {
     return delegate;
   }
 

li-apache-kafka-clients/src/main/java/com/linkedin/kafka/clients/producer/DelegatingProducer.java

@@ -0,0 +1,12 @@
+/*
+ * Copyright 2019 LinkedIn Corp. Licensed under the BSD 2-Clause License (the "License").
 See License in the project root for license information.
+ */
+
+package com.linkedin.kafka.clients.producer;
+
+import org.apache.kafka.clients.producer.Producer;
+
+
+public interface DelegatingProducer<K, V> extends Producer<K, V> {
+  Producer<K, V> getDelegate();
+}

li-apache-kafka-clients/src/main/java/com/linkedin/kafka/clients/producer/LiKafkaInstrumentedProducerImpl.java

@@ -49,7 +49,7 @@
  * @param <K>
  * @param <V>
  */
-public class LiKafkaInstrumentedProducerImpl<K, V> implements Producer<K, V>, EventHandler {
+public class LiKafkaInstrumentedProducerImpl<K, V> implements DelegatingProducer<K, V>, EventHandler {
   private static final Logger LOG = LoggerFactory.getLogger(LiKafkaInstrumentedProducerImpl.class);
   private static final String BOUNDED_FLUSH_THREAD_PREFIX = "Bounded-Flush-Thread-";
 
@@ -156,8 +156,8 @@ public void configChangeRequested(UUID commandId, Map<String, String> configDiff
     //TODO - respond to command UUID
   }
 
-  //package-private FOR TESTING
-  Producer<K, V> getDelegate() {
+  @Override
+  public Producer<K, V> getDelegate() {
     return delegate;
   }