Batching zeitnah versendeter Nachrichten ermöglicht

* Durch den Aufruf der blockierenden Methode `get()` wurde erzwungen, das
  der Versand einer Nachricht von Kafka bestätigt wurde, bevor die
  nächste Nachricht an Kafka übergeben werden konnte.
* Der blockierende Aufruf wurde jetzt in eine getrennte asynchrone
  Verarbeitung (in dem Default-Thread-Pool) verschoben.
* Dadurch können jetzt mehrere für den Versand bereitgestellte Nachrichten
  hintereinander an Kafka übergeben werden.
* _Beachte:_ Der Effekt wird erst dann wirklich deutlich, wenn das
  Throttling über `Thread.sleep(500)` deaktiviert wird!
* _Beachte:_ Da die Verarbeitung der Rückmeldung von Kafka über den
  Default-Thread-Pool erfolgt, erfolgt die Erzeugung der Log-Ausgaben
  asynchron in verschiedenen Threads. D.h., die Log-Meldungen werden
  _nicht_ in der Reihenfolge ausgegeben, in der die Nachrichten versendet
  wurden, was auf den ersten Blick sehr verwirrend ist!

diff --git a/src/main/java/de/juplo/kafka/ExampleProducer.java b/src/main/java/de/juplo/kafka/ExampleProducer.java
index ca5ba67858f7578ff8101ca4d722c997bc742378..3a5f94afeacadee3951de67061667a5eceaa566e 100644 (file)
--- a/src/main/java/de/juplo/kafka/ExampleProducer.java
+++ b/src/main/java/de/juplo/kafka/ExampleProducer.java
@@ -81,26 +81,30 @@ public class ExampleProducer
     semaphore.acquire();
     CompletableFuture<RecordMetadata> completableFuture = CompletableFuture
       .supplyAsync(() ->
+      {
+        Future<RecordMetadata> recordMetadataFuture = producer.send(record);
+        long sendRequestQueued = System.currentTimeMillis();
+        semaphore.release();
+        log.trace(
+          "{} - Queued message {}={}, latency={}ms",
+          id,
+          key,
+          value,
+          sendRequestQueued - sendRequested
+        );
+        return recordMetadataFuture;
+      }, executor)
+      .thenApplyAsync(recordMetadataFuture ->
       {
         try
         {
-          Future<RecordMetadata> result = producer.send(record);
-          long sendRequestQueued = System.currentTimeMillis();
-          semaphore.release();
-          log.trace(
-            "{} - Queued message {}={}, latency={}ms",
-            id,
-            key,
-            value,
-            sendRequestQueued - sendRequested
-          );
-          return result.get();
+          return recordMetadataFuture.get();
         }
         catch (Exception e)
         {
           throw new RuntimeException(e);
         }
-      }, executor);
+      });
 
     completableFuture.whenComplete((metadata, e) ->
     {