From 95a4a71fd0a67163459e3bd56852b5e9e06d3ad9 Mon Sep 17 00:00:00 2001
From: Kai Moritz <kai@juplo.de>
Date: Fri, 1 Apr 2022 15:08:14 +0200
Subject: [PATCH] Vorlage

---
 .../java/de/juplo/kafka/SimpleConsumer.java   | 62 ++------------
 .../java/de/juplo/kafka/SimpleProducer.java   | 82 +------------------
 2 files changed, 7 insertions(+), 137 deletions(-)

diff --git a/src/main/java/de/juplo/kafka/SimpleConsumer.java b/src/main/java/de/juplo/kafka/SimpleConsumer.java
index e4d9697..a4b1196 100644
--- a/src/main/java/de/juplo/kafka/SimpleConsumer.java
+++ b/src/main/java/de/juplo/kafka/SimpleConsumer.java
@@ -18,7 +18,6 @@ import java.util.concurrent.locks.ReentrantLock;
 @Slf4j
 public class SimpleConsumer
 {
-  private long consumed = 0;
   private KafkaConsumer<String, String> consumer;
   private Lock lock = new ReentrantLock();
   private Condition stopped = lock.newCondition();
@@ -26,79 +25,28 @@ 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-Commits
-    // end::create[]
-    props.put("auto.offset.reset", "earliest");
-    // tag::create[]
-    props.put("key.deserializer", StringDeserializer.class.getName());
-    props.put("value.deserializer", StringDeserializer.class.getName());
-
-    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
-    // end::create[]
-    this.consumer = consumer;
+    this.consumer = null; // TODO: Instanziierung
   }
 
 
   public void run()
   {
-    String id = "C";
-
     try
     {
-      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[]
-      }
-      // end::loop[]
-    }
-    catch(WakeupException e)
-    {
-      log.info("{} - RIIING!", id);
-    }
-    catch(Exception e)
-    {
-      log.error("{} - Unexpected error: {}", id, e.toString());
+      // TODO: Nachrichten konsumieren
     }
+    // TODO: Exception-Handling
     finally
     {
       this.lock.lock();
       try
       {
-        log.info("{} - Closing the KafkaConsumer", id);
-        consumer.close();
-        log.info("C - DONE!");
+        // TODO: Clean-Up
         stopped.signal();
       }
       finally
       {
         this.lock.unlock();
-        log.info("{}: Consumed {} messages in total, exiting!", id, consumed);
       }
     }
   }
@@ -113,7 +61,7 @@ public class SimpleConsumer
       instance.lock.lock();
       try
       {
-        instance.consumer.wakeup();
+        // TODO: Consumer beenden
         instance.stopped.await();
       }
       catch (InterruptedException e)
diff --git a/src/main/java/de/juplo/kafka/SimpleProducer.java b/src/main/java/de/juplo/kafka/SimpleProducer.java
index 43a7227..c5ea5ef 100644
--- a/src/main/java/de/juplo/kafka/SimpleProducer.java
+++ b/src/main/java/de/juplo/kafka/SimpleProducer.java
@@ -17,96 +17,18 @@ public class SimpleProducer
   private final String topic;
   private final KafkaProducer<String, String> producer;
 
-  private long produced = 0;
-
   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());
-
-    KafkaProducer<String, String> producer = new KafkaProducer<>(props);
-    // end::create[]
-
     this.id = clientId;
     this.topic = topic;
-    this.producer = producer;
+    this.producer = null; // TODO: Instanziierung
   }
 
   public void run()
   {
-    long i = 0;
-
-    try
-    {
-      for (; i < 100 ; i++)
-      {
-        send(Long.toString(i%10), Long.toString(i));
-      }
-
-      log.info("{} - Done", id);
-    }
-    finally
-    {
-      log.info("{}: Closing the KafkaProducer", id);
-      producer.close();
-      log.info("{}: Produced {} messages in total, exiting!", id, produced);
-    }
+    // TODO: Nachrichten versenden
   }
 
-  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
-    );
-  }
 
 
   public static void main(String[] args) throws Exception
-- 
2.20.1