From 2e8f4627d50c49f9ec9260728ee8514fd7b69880 Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Mon, 1 Aug 2022 22:40:27 +0200 Subject: [PATCH] Vorlage --- .../java/de/juplo/kafka/ExampleProducer.java | 54 ++++--------------- 1 file changed, 11 insertions(+), 43 deletions(-) diff --git a/src/main/java/de/juplo/kafka/ExampleProducer.java b/src/main/java/de/juplo/kafka/ExampleProducer.java index aa014a6..195e38c 100644 --- a/src/main/java/de/juplo/kafka/ExampleProducer.java +++ b/src/main/java/de/juplo/kafka/ExampleProducer.java @@ -26,11 +26,7 @@ public class ExampleProducer String clientId) { Properties props = new Properties(); - props.put("bootstrap.servers", broker); - props.put("client.id", clientId); // Nur zur Wiedererkennung - props.put("key.serializer", StringSerializer.class.getName()); - props.put("value.serializer", StringSerializer.class.getName()); - props.put("metadata.maxage.ms", 5000); + // TODO: Konfiguration für den Producer zusammenstellen this.id = clientId; this.topic = topic; @@ -66,44 +62,16 @@ public class ExampleProducer { final long time = System.currentTimeMillis(); - final ProducerRecord 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 message {}={}, partition={}:{}, timestamp={}, latency={}ms", - id, - record.key(), - record.value(), - metadata.partition(), - metadata.offset(), - metadata.timestamp(), - now - time - ); - } - else - { - // HANDLE ERROR - log.error( - "{} - ERROR for message {}={}, timestamp={}, latency={}ms: {}", - id, - record.key(), - record.value(), - metadata == null ? -1 : metadata.timestamp(), - now - time, - e.toString() - ); - } - }); + ProducerRecord record = null; + // TODO: + // + // 1. ProducerRecord erzeugen + // 2. Record mit producer.send(...) verschicken + // Dabei beachten: + // * Callback implementieren und Erfolg/Fehler logge + // * Tipp: Aktuellen Zeitstempel auch im Callback erzeugen und + // Millisekunden berechnen, die bis zum Senden verstrichen sind: + // long now = System.currentTimeMillis(); // Für: now - time long now = System.currentTimeMillis(); log.trace( -- 2.20.1