diff --git a/li-apache-kafka-clients/src/main/java/com/linkedin/kafka/clients/consumer/DelegatingConsumer.java b/li-apache-kafka-clients/src/main/java/com/linkedin/kafka/clients/consumer/DelegatingConsumer.java
new file mode 100644
index 0000000..c93aeb5
--- /dev/null
+++ b/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();
+}
diff --git a/li-apache-kafka-clients/src/main/java/com/linkedin/kafka/clients/consumer/LiKafkaInstrumentedConsumerImpl.java b/li-apache-kafka-clients/src/main/java/com/linkedin/kafka/clients/consumer/LiKafkaInstrumentedConsumerImpl.java
index e83d38a..5bedd4d 100644
--- a/li-apache-kafka-clients/src/main/java/com/linkedin/kafka/clients/consumer/LiKafkaInstrumentedConsumerImpl.java
+++ b/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;
   }
 
diff --git a/li-apache-kafka-clients/src/main/java/com/linkedin/kafka/clients/producer/DelegatingProducer.java b/li-apache-kafka-clients/src/main/java/com/linkedin/kafka/clients/producer/DelegatingProducer.java
new file mode 100644
index 0000000..8d838d4
--- /dev/null
+++ b/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();
+}
diff --git a/li-apache-kafka-clients/src/main/java/com/linkedin/kafka/clients/producer/LiKafkaInstrumentedProducerImpl.java b/li-apache-kafka-clients/src/main/java/com/linkedin/kafka/clients/producer/LiKafkaInstrumentedProducerImpl.java
index 2e1d3cb..da8aad0 100644
--- a/li-apache-kafka-clients/src/main/java/com/linkedin/kafka/clients/producer/LiKafkaInstrumentedProducerImpl.java
+++ b/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;
   }