Synchronisation bei Start/Stop repariert und Ausgabe für Fehler ergänzt
authorKai Moritz <kai@juplo.de>
Fri, 25 Mar 2022 13:06:53 +0000 (14:06 +0100)
committerKai Moritz <kai@juplo.de>
Fri, 25 Mar 2022 14:03:14 +0000 (15:03 +0100)
src/main/java/de/juplo/kafka/EndlessProducer.java

index cc3150e..33a3815 100644 (file)
@@ -9,7 +9,6 @@ import javax.annotation.PreDestroy;
 import java.util.Properties;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.Properties;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
 
 
 @Slf4j
 
 
 @Slf4j
@@ -24,7 +23,6 @@ public class EndlessProducer implements Runnable
   private boolean running = false;
   private long i = 0;
   private long produced = 0;
   private boolean running = false;
   private long i = 0;
   private long produced = 0;
-  private Future<?> future = null;
 
   public EndlessProducer(
       ExecutorService executor,
 
   public EndlessProducer(
       ExecutorService executor,
@@ -75,7 +73,15 @@ public class EndlessProducer implements Runnable
     }
     catch (Exception e)
     {
     }
     catch (Exception e)
     {
-
+      log.error("{} - Unexpected Exception:", id, e);
+    }
+    finally
+    {
+      synchronized (this)
+      {
+        running = false;
+        log.info("{} - Stopped - produced {} messages so far", id, produced);
+      }
     }
   }
 
     }
   }
 
@@ -138,7 +144,7 @@ public class EndlessProducer implements Runnable
 
     log.info("{} - Starting - produced {} messages before", id, produced);
     running = true;
 
     log.info("{} - Starting - produced {} messages before", id, produced);
     running = true;
-    future = executor.submit(this);
+    executor.submit(this);
   }
 
   public synchronized void stop() throws ExecutionException, InterruptedException
   }
 
   public synchronized void stop() throws ExecutionException, InterruptedException
@@ -148,8 +154,6 @@ public class EndlessProducer implements Runnable
 
     log.info("{} - Stopping...", id);
     running = false;
 
     log.info("{} - Stopping...", id);
     running = false;
-    future.get();
-    log.info("{} - Stopped - produced {} messages so far", id, produced);
   }
 
   @PreDestroy
   }
 
   @PreDestroy