fix interupt

NucleoTeam · Sep 26, 2024 · 9440668 · 9440668
1 parent 5582ea9
commit 9440668
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 6 deletions.
diff --git a/build.gradle b/build.gradle
@@ -7,7 +7,7 @@ plugins {
 }
 
 group = 'com.nucleodb'
-version = '1.18.10'
+version = '1.18.11'
 
 repositories {
     mavenCentral()

diff --git a/src/main/java/com/nucleodb/library/mqs/ConsumerHandler.java b/src/main/java/com/nucleodb/library/mqs/ConsumerHandler.java
@@ -52,7 +52,9 @@ public void readFromStart(){
     Thread queueThread;
     while((queueThread = queueThreads.poll())!=null) {
       try {
-        queueThread.interrupt();
+        synchronized (queueThread) {
+          queueThread.interrupt();
+        }
       }catch (Exception e){}
     }
   }

diff --git a/src/main/java/com/nucleodb/library/mqs/kafka/KafkaConsumerHandler.java b/src/main/java/com/nucleodb/library/mqs/kafka/KafkaConsumerHandler.java
@@ -113,10 +113,12 @@ public void start(int queueHandlers) {
 
     @Override
     public void readFromStart() {
-        kafkaConsumingThread.interrupt();
-        try {
-            kafkaConsumingThread.wait();
-        } catch (InterruptedException e) {
+        synchronized (kafkaConsumingThread) {
+            kafkaConsumingThread.interrupt();
+            try {
+                kafkaConsumingThread.wait();
+            } catch (InterruptedException e) {
+            }
         }
         boolean connectionType = this.getConnectionHandler() != null;
         boolean databaseType = this.getDatabase() != null;