WIP

[demos/kafka/training] / src / main / java / de / juplo / kafka / SimpleProducer.java

diff --git a/src/main/java/de/juplo/kafka/SimpleProducer.java b/src/main/java/de/juplo/kafka/SimpleProducer.java
index 06d28ff..2b5b441 100644 (file)
--- a/src/main/java/de/juplo/kafka/SimpleProducer.java
+++ b/src/main/java/de/juplo/kafka/SimpleProducer.java
@@ -7,6 +7,7 @@ import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.serialization.StringSerializer;
 
 import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
 
 
 @Slf4j
@@ -56,8 +57,9 @@ public class SimpleProducer
     }
   }
 
-  void send(String key, String value)
+  CompletableFuture<Long> send(String key, String value)
   {
+    final CompletableFuture<Long> result = new CompletableFuture<>();
     final long time = System.currentTimeMillis();
 
     final ProducerRecord<String, String> record = new ProducerRecord<>(
@@ -83,6 +85,7 @@ public class SimpleProducer
             metadata.timestamp(),
             now - time
         );
+        result.complete(metadata.offset());
       }
       else
       {
@@ -95,6 +98,7 @@ public class SimpleProducer
             now - time,
             e.toString()
         );
+        result.completeExceptionally(e);
       }
     });
 
@@ -106,6 +110,8 @@ public class SimpleProducer
         record.key(),
         now - time
     );
+
+    return result;
   }