From 983b4ed5806f34d410d9b294179311868b8dae0d Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Sun, 3 Apr 2022 20:24:31 +0200 Subject: [PATCH] Vorlage --- .../java/de/juplo/kafka/RestProducer.java | 59 +------------------ 1 file changed, 2 insertions(+), 57 deletions(-) diff --git a/src/main/java/de/juplo/kafka/RestProducer.java b/src/main/java/de/juplo/kafka/RestProducer.java index 408cd2f..018d905 100644 --- a/src/main/java/de/juplo/kafka/RestProducer.java +++ b/src/main/java/de/juplo/kafka/RestProducer.java @@ -53,63 +53,8 @@ public class RestProducer { DeferredResult result = new DeferredResult<>(); - final long time = System.currentTimeMillis(); - - final ProducerRecord record = new ProducerRecord<>( - topic, // Topic - partition, // Partition - key, // Key - value // Value - ); - - record.headers().add("source", id.getBytes()); - if (correlationId != null) - { - record.headers().add("id", BigInteger.valueOf(correlationId).toByteArray()); - } - - producer.send(record, (metadata, e) -> - { - long now = System.currentTimeMillis(); - if (e == null) - { - // HANDLE SUCCESS - produced++; - result.setResult(new ProduceSuccess(metadata.partition(), metadata.offset())); - 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 - result.setErrorResult(new ProduceFailure(e)); - 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: Ergänzen Sie die Logik Ihres REST-Producers und + // ergänzen sie die versendten Nachrichten um die Header return result; } -- 2.20.1