From 944066848f9e6f90f80ec9bd5f07df07774f4ad1 Mon Sep 17 00:00:00 2001
From: Nathaniel Davidson <nathaniel.davidson@gmail.com>
Date: Thu, 26 Sep 2024 01:34:58 -0700
Subject: [PATCH] fix interupt

---
 build.gradle                                           |  2 +-
 .../java/com/nucleodb/library/mqs/ConsumerHandler.java |  4 +++-
 .../library/mqs/kafka/KafkaConsumerHandler.java        | 10 ++++++----
 3 files changed, 10 insertions(+), 6 deletions(-)

diff --git a/build.gradle b/build.gradle
index ad1046f..8413fe3 100644
--- 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
index dab0c2a..3131659 100644
--- 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
index 8b632c6..011c3a5 100644
--- 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;