Close the pulsar and Kafka topic handlers safely.

Signed-off-by: Keshava Munegowda <keshava.gowda@gmail.com>
kmgowda · Feb 1, 2020 · e192f42 · e192f42
1 parent 925c47e
commit e192f42
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 1 deletion.
diff --git a/driver-kafka/src/main/java/io/sbk/Kafka/Kafka.java b/driver-kafka/src/main/java/io/sbk/Kafka/Kafka.java
@@ -113,7 +113,9 @@ public void openStorage(final Parameters params) throws  IOException {
 
     @Override
     public void closeStorage(final Parameters params) throws IOException {
-        topicHandler.close();
+        if (topicHandler != null) {
+            topicHandler.close();
+        }
     }
 
     @Override

diff --git a/driver-pulsar/src/main/java/io/sbk/Pulsar/Pulsar.java b/driver-pulsar/src/main/java/io/sbk/Pulsar/Pulsar.java
@@ -115,7 +115,9 @@ public void openStorage(final Parameters params) throws  IOException {
 
     @Override
     public void closeStorage(final Parameters params) throws IOException {
+        if (topicHandler != null) {
             topicHandler.close();
+        }
     }
 
     @Override