check for null before closing delegates of instrumented clients (#163)

linkedin · Jan 12, 2020 · ab76279 · ab76279
1 parent 1f981de
commit ab76279
Show file tree

Hide file tree

Showing 2 changed files with 26 additions and 11 deletions.
diff --git a/...ts/src/main/java/com/linkedin/kafka/clients/consumer/LiKafkaInstrumentedConsumerImpl.java b/...ts/src/main/java/com/linkedin/kafka/clients/consumer/LiKafkaInstrumentedConsumerImpl.java
@@ -707,9 +707,12 @@ public void close() {
         return;
       }
       try {
-        delegate.close();
+        Consumer<K, V> delegate = this.delegate;
+        if (delegate != null) {
+          delegate.close();
+        }
       } finally {
-        delegate = null;
+        this.delegate = null;
         closeMdsClient();
       }
     }
@@ -723,9 +726,12 @@ public void close(long timeout, TimeUnit unit) {
         return;
       }
       try {
-        delegate.close(timeout, unit);
+        Consumer<K, V> delegate = this.delegate;
+        if (delegate != null) {
+          delegate.close(timeout, unit);
+        }
       } finally {
-        delegate = null;
+        this.delegate = null;
         closeMdsClient();
       }
     }
@@ -738,9 +744,12 @@ public void close(Duration timeout) {
         return;
       }
       try {
-        delegate.close(timeout);
+        Consumer<K, V> delegate = this.delegate;
+        if (delegate != null) {
+          delegate.close(timeout);
+        }
       } finally {
-        delegate = null;
+        this.delegate = null;
         closeMdsClient();
       }
     }

diff --git a/...ts/src/main/java/com/linkedin/kafka/clients/producer/LiKafkaInstrumentedProducerImpl.java b/...ts/src/main/java/com/linkedin/kafka/clients/producer/LiKafkaInstrumentedProducerImpl.java
@@ -407,9 +407,12 @@ public void close() {
       return;
     }
     try {
-      delegate.close();
+      Producer<K, V> delegate = this.delegate;
+      if (delegate != null) {
+        delegate.close();
+      }
     } finally {
-      delegate = null;
+      this.delegate = null;
       closeMdsClient();
     }
   }
@@ -426,10 +429,13 @@ public void close(Duration timeout) {
       return;
     }
     try {
-      //TODO - fix back after bumping up kafka
-      delegate.close(timeout.toMillis(), TimeUnit.MILLISECONDS);
+      Producer<K, V> delegate = this.delegate;
+      if (delegate != null) {
+        //TODO - fix back after bumping up kafka
+        delegate.close(timeout.toMillis(), TimeUnit.MILLISECONDS);
+      }
     } finally {
-      delegate = null;
+      this.delegate = null;
       closeMdsClient();
     }
   }