additional reset of consumer

NucleoTeam · Sep 28, 2024 · ca21205 · ca21205
1 parent a15c985
commit ca21205
Show file tree

Hide file tree

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

diff --git a/src/main/java/com/nucleodb/library/mqs/ConsumerHandler.java b/src/main/java/com/nucleodb/library/mqs/ConsumerHandler.java
@@ -53,6 +53,7 @@ public void readFromStart() throws InterruptedException {
     queue.clear();
     startupPhaseConsume.set(true);
     leftToRead.set(0);
+    startupItems = -1;
     queueTasks.shutdownNow();
     queueTasks.awaitTermination(4, TimeUnit.SECONDS);
     queueTasks = Executors.newFixedThreadPool(60);

diff --git a/src/main/java/com/nucleodb/library/mqs/kafka/KafkaConsumerHandler.java b/src/main/java/com/nucleodb/library/mqs/kafka/KafkaConsumerHandler.java
@@ -176,10 +176,7 @@ public void run() {
                 try {
                     getConsumer().poll(Duration.ofMillis(100));
                     Thread.sleep(1000);
-                } catch (InterruptedException e) {
-//                    throw new RuntimeException(e);
-                    e.printStackTrace();
-                }
+                } catch (InterruptedException e) {}
             }
             seek(offsets);
             super.setStartupLoadCount(getDatabase().getStartupLoadCount());
@@ -189,11 +186,7 @@ public void run() {
                 try {
                     getConsumer().poll(Duration.ofMillis(100));
                     Thread.sleep(1000);
-                } catch (InterruptedException e) {
-                    e.printStackTrace();
-                    //throw new RuntimeException(e);
-
-                }
+                } catch (InterruptedException e) {}
             }
             seek(offsets);
             super.setStartupLoadCount(getConnectionHandler().getStartupLoadCount());