diff --git a/extensions-contrib/kafka-emitter/src/main/java/org/apache/druid/emitter/kafka/KafkaEmitter.java b/extensions-contrib/kafka-emitter/src/main/java/org/apache/druid/emitter/kafka/KafkaEmitter.java
index 4a58464f7308..c776e3f2f8b2 100644
--- a/extensions-contrib/kafka-emitter/src/main/java/org/apache/druid/emitter/kafka/KafkaEmitter.java
+++ b/extensions-contrib/kafka-emitter/src/main/java/org/apache/druid/emitter/kafka/KafkaEmitter.java
@@ -181,6 +181,10 @@ private void sendToKafka(final String topic, MemoryBoundLinkedBlockingQueue<Stri
       }
     }
     catch (Throwable e) {
+      if (e instanceof InterruptedException && e.getMessage() == null) {
+        log.info("Normal exit.");
+        return;
+      }
       log.warn(e, "Exception while getting record from queue or producer send, Events would not be emitted anymore.");
     }
   }
diff --git a/extensions-contrib/kafka-emitter/src/test/java/org/apache/druid/emitter/kafka/KafkaEmitterTest.java b/extensions-contrib/kafka-emitter/src/test/java/org/apache/druid/emitter/kafka/KafkaEmitterTest.java
index 9e6846a5d8b8..f39e8e826063 100644
--- a/extensions-contrib/kafka-emitter/src/test/java/org/apache/druid/emitter/kafka/KafkaEmitterTest.java
+++ b/extensions-contrib/kafka-emitter/src/test/java/org/apache/druid/emitter/kafka/KafkaEmitterTest.java
@@ -140,6 +140,8 @@ protected Producer<String, String> setKafkaProducer()
     }
     countDownSentEvents.await();
 
+    kafkaEmitter.close();
+
     Assert.assertEquals(0, kafkaEmitter.getMetricLostCount());
     Assert.assertEquals(0, kafkaEmitter.getAlertLostCount());
     Assert.assertEquals(0, kafkaEmitter.getSegmentMetadataLostCount());