Clean useless InterruptedException warn in ingestion task log (#15519)

* Clean useless InterruptedException warn in ingestion task log * test coverage for the code change, manually close the scheduler thread to trigger Interrupt signal --------- Co-authored-by: Qiong Chen <[email protected]>
apache · Dec 15, 2023 · c9be1cb · c9be1cb
1 parent 9deeb28
commit c9be1cb
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 0 deletions.
diff --git a/...ions-contrib/kafka-emitter/src/main/java/org/apache/druid/emitter/kafka/KafkaEmitter.java b/...ions-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/...-contrib/kafka-emitter/src/test/java/org/apache/druid/emitter/kafka/KafkaEmitterTest.java b/...-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());