Merge der überarbeiteten Compose-Konfiguration ('endless-stream-producer')

[demos/kafka/training] / src / main / java / de / juplo / kafka / EndlessProducer.java

diff --git a/src/main/java/de/juplo/kafka/EndlessProducer.java b/src/main/java/de/juplo/kafka/EndlessProducer.java
index cc3150e..df5ff01 100644 (file)
--- a/src/main/java/de/juplo/kafka/EndlessProducer.java
+++ b/src/main/java/de/juplo/kafka/EndlessProducer.java
@@ -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.concurrent.Future;
 
 
 @Slf4j
@@ -24,7 +23,6 @@ public class EndlessProducer implements Runnable
   private boolean running = false;
   private long i = 0;
   private long produced = 0;
-  private Future<?> future = null;
 
   public EndlessProducer(
       ExecutorService executor,
@@ -75,60 +73,21 @@ public class EndlessProducer implements Runnable
     }
     catch (Exception e)
     {
-
+      log.error("{} - Unexpected Exception:", id, e);
+    }
+    finally
+    {
+      synchronized (this)
+      {
+        running = false;
+        log.info("{} - Stopped - produced {} messages so far", id, produced);
+      }
     }
   }
 
   void send(String key, String value)
   {
-    final long time = System.currentTimeMillis();
-
-    final ProducerRecord<String, String> record = new ProducerRecord<>(
-        "test", // Topic
-        key,    // Key
-        value   // Value
-    );
-
-    producer.send(record, (metadata, e) ->
-    {
-      long now = System.currentTimeMillis();
-      if (e == null)
-      {
-        // HANDLE SUCCESS
-        produced++;
-        log.debug(
-            "{} - Sent key={} message={} partition={}/{} timestamp={} latency={}ms",
-            id,
-            record.key(),
-            record.value(),
-            metadata.partition(),
-            metadata.offset(),
-            metadata.timestamp(),
-            now - time
-        );
-      }
-      else
-      {
-        // HANDLE ERROR
-        log.error(
-            "{} - ERROR key={} timestamp={} latency={}ms: {}",
-            id,
-            record.key(),
-            metadata == null ? -1 : metadata.timestamp(),
-            now - time,
-            e.toString()
-        );
-      }
-    });
-
-    long now = System.currentTimeMillis();
-    log.trace(
-        "{} - Queued #{} key={} latency={}ms",
-        id,
-        value,
-        record.key(),
-        now - time
-    );
+    // TODO: Die übergebene Nachricht versenden
   }
 
   public synchronized void start()
@@ -138,7 +97,7 @@ public class EndlessProducer implements Runnable
 
     log.info("{} - Starting - produced {} messages before", id, produced);
     running = true;
-    future = executor.submit(this);
+    executor.submit(this);
   }
 
   public synchronized void stop() throws ExecutionException, InterruptedException
@@ -148,8 +107,6 @@ public class EndlessProducer implements Runnable
 
     log.info("{} - Stopping...", id);
     running = false;
-    future.get();
-    log.info("{} - Stopped - produced {} messages so far", id, produced);
   }
 
   @PreDestroy