From: Kai Moritz <kai@juplo.de>
Date: Sat, 23 Jul 2022 09:25:35 +0000 (+0200)
Subject: Merge der überarbeiteten Compose-Konfiguration ('endless-stream-producer')
X-Git-Url: https://juplo.de/gitweb/?a=commitdiff_plain;h=refs%2Fheads%2Fendless-stream-producer-vorlage;hp=26cef68053cc7472055344b171a44f34e7592ebb;p=demos%2Fkafka%2Ftraining

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

diff --git a/src/main/java/de/juplo/kafka/EndlessProducer.java b/src/main/java/de/juplo/kafka/EndlessProducer.java
index 7a5b324..df5ff01 100644
--- a/src/main/java/de/juplo/kafka/EndlessProducer.java
+++ b/src/main/java/de/juplo/kafka/EndlessProducer.java
@@ -87,54 +87,7 @@ public class EndlessProducer implements Runnable
 
   void send(String key, String value)
   {
-    final long time = System.currentTimeMillis();
-
-    final ProducerRecord<String, String> record = new ProducerRecord<>(
-        topic,  // 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()