Vorlage

diff --git a/src/main/java/de/juplo/kafka/SimpleConsumer.java b/src/main/java/de/juplo/kafka/SimpleConsumer.java
index 913ffd5..a5554c1 100644 (file)
--- a/src/main/java/de/juplo/kafka/SimpleConsumer.java
+++ b/src/main/java/de/juplo/kafka/SimpleConsumer.java
@@ -27,19 +27,7 @@ public class SimpleConsumer
 
   public SimpleConsumer()
   {
-    // tag::create[]
-    Properties props = new Properties();
-    props.put("bootstrap.servers", ":9092");
-    props.put("group.id", "my-consumer"); // << Used for Offset-Management
-    // end::create[]
-    props.put("auto.offset.reset", "earliest");
-    // tag::create[]
-    props.put("key.deserializer", StringDeserializer.class.getName());
-    props.put("value.deserializer", StringDeserializer.class.getName());
-
-    Consumer<String, String> consumer = new KafkaConsumer<>(props);
-    // end::create[]
-    this.consumer = consumer;
+    this.consumer = null; // TODO: Eine Instanz von KafkaConsumer erzeugen
   }
 
 
@@ -52,31 +40,12 @@ public class SimpleConsumer
       log.info("{} - Subscribing to topic test", id);
       consumer.subscribe(Arrays.asList("test"));
 
-      // tag::loop[]
       while (true)
       {
-        ConsumerRecords<String, String> records =
-            consumer.poll(Duration.ofSeconds(1));
-
-        // Do something with the data...
-        // end::loop[]
-        log.info("{} - Received {} messages", id, records.count());
-        for (ConsumerRecord<String, String> record : records)
-        {
-          consumed++;
-          log.info(
-              "{} - {}: {}/{} - {}={}",
-              id,
-              record.offset(),
-              record.topic(),
-              record.partition(),
-              record.key(),
-              record.value()
-          );
-        }
-        // tag::loop[]
+        // TODO:
+        // Über consumer.poll() Nachrichten abrufen und diese
+        // über log.info() ausgeben.
       }
-      // end::loop[]
     }
     catch(WakeupException e)
     {

diff --git a/src/main/java/de/juplo/kafka/SimpleProducer.java b/src/main/java/de/juplo/kafka/SimpleProducer.java
index c0a7334..1ebc851 100644 (file)
--- a/src/main/java/de/juplo/kafka/SimpleProducer.java
+++ b/src/main/java/de/juplo/kafka/SimpleProducer.java
@@ -20,18 +20,12 @@ public class SimpleProducer
 
   public SimpleProducer(String clientId, String topic)
   {
-    // tag::create[]
     Properties props = new Properties();
-    props.put("bootstrap.servers", "localhost:9092");
-    props.put("key.serializer", StringSerializer.class.getName());
-    props.put("value.serializer", StringSerializer.class.getName());
-
-    Producer<String, String> producer = new KafkaProducer<>(props);
-    // end::create[]
+    // TODO: Konfiguration für Producer zusammenstellen
 
     this.id = clientId;
     this.topic = topic;
-    this.producer = producer;
+    this.producer = null; // TODO: Eine Instanz von KafkaProducer erzeugen
   }
 
   public void run()
@@ -59,43 +53,12 @@ public class SimpleProducer
   {
     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()
-        );
-      }
-    });
+    final ProducerRecord<String, String> record = null; // TODO:
+    // Instanz von ProducerRecord mit key & value aus den Parametern
+    // der Methode erzeugen und diesen an das Topic aus dem Attribut
+    // this.topic versenden.
+    // Dabei den Callback für die Ausgabe von Erfolg/Fehlern nutzen.
+    // Für ganz Schnelle: Versanddauer im Callback messen und ausgeben.
 
     long now = System.currentTimeMillis();
     log.trace(