From c9be1cb4e84f15724a3d881fccc80e3b2432c0d5 Mon Sep 17 00:00:00 2001
From: sensor <fectrain@outlook.com>
Date: Fri, 15 Dec 2023 11:18:53 +0800
Subject: [PATCH] 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 <qiong.chen@shopee.com>
---
 .../java/org/apache/druid/emitter/kafka/KafkaEmitter.java     | 4 ++++
 .../java/org/apache/druid/emitter/kafka/KafkaEmitterTest.java | 2 ++
 2 files changed, 6 insertions(+)

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());