LoomKafkaProducer, wait for sendFromQueueThread to finish

Author: maschmid

data-plane/receiver-loom/src/main/java/dev/knative/eventing/kafka/broker/receiverloom/LoomKafkaProducer.java

@@ -42,6 +42,7 @@ public class LoomKafkaProducer<K, V> implements ReactiveKafkaProducer<K, V> {
 
     private final BlockingQueue<RecordPromise<K, V>> eventQueue;
     private final AtomicBoolean isClosed;
+    private final AtomicBoolean isFinished;
     private final ProducerTracer<?> tracer;
     private final VertxInternal vertx;
     private final Thread sendFromQueueThread;
@@ -52,6 +53,7 @@ public LoomKafkaProducer(Vertx v, Producer<K, V> producer) {
         this.producer = producer;
         this.eventQueue = new LinkedBlockingQueue<>();
         this.isClosed = new AtomicBoolean(false);
+        this.isFinished = new AtomicBoolean(false);
         this.vertx = (VertxInternal) v;
         final var ctxInt = ((ContextInternal) v.getOrCreateContext()).unwrap();
         if (ctxInt.tracer() != null) {
@@ -123,6 +125,7 @@ private void sendFromQueue() {
             }
         }
 
+        isFinished.set(true);
         logger.debug("Background thread completed.");
     }
 
@@ -136,13 +139,11 @@ public Future<Void> close() {
 
         Thread.ofVirtual().start(() -> {
             try {
-                while (!eventQueue.isEmpty()) {
-                    logger.debug("Waiting for the eventQueue to become empty");
+                while (!isFinished.get()) {
+                    logger.debug("Waiting for the sendFromQueueThread to finish");
                     Thread.sleep(2000L);
                 }
-                logger.debug("Interrupting sendFromQueueThread thread");
-                sendFromQueueThread.interrupt();
-                logger.debug("Waiting for sendFromQueueThread thread to complete");
+
                 sendFromQueueThread.join();
                 logger.debug("Closing the producer");
                 producer.close();