Merge pull request #578 from folio-org/b17.4-CIRCSTORE-639

CIRCSTORE-654 Flush and close Kafka producer (backport to Sunflower)

Author: roman-barannyk

NEWS.md

@@ -1,3 +1,6 @@
+## 17.4.4 2026-06-09
+* Flush and close Kafka producer (CIRCSTORE-654)
+
 ## 17.4.3 2026-02-05
 * Fix request status not updating upon check-in (CIRCSTORE-634)
 * Fix missing request position (CIRCSTORE-635)

pom.xml

@@ -2,7 +2,7 @@
   <modelVersion>4.0.0</modelVersion>
   <artifactId>mod-circulation-storage</artifactId>
   <groupId>org.folio</groupId>
-  <version>17.4.4-SNAPSHOT</version>
+  <version>17.4.5-SNAPSHOT</version>
   <licenses>
     <license>
       <name>Apache License 2.0</name>

src/main/java/org/folio/service/event/DomainEventPublisher.java

@@ -1,7 +1,7 @@
 package org.folio.service.event;
 
+import static io.vertx.core.Future.succeededFuture;
 import static org.apache.logging.log4j.LogManager.getLogger;
-import static org.folio.service.event.EntityChangedEventPublisherFactory.requestEventPublisher;
 
 import java.util.Map;
 
@@ -11,7 +11,6 @@
 import org.folio.kafka.SimpleKafkaProducerManager;
 import org.folio.kafka.services.KafkaEnvironmentProperties;
 import org.folio.kafka.services.KafkaProducerRecordBuilder;
-import org.folio.rest.jaxrs.model.Request;
 import org.folio.rest.tools.utils.TenantTool;
 
 import io.vertx.core.Context;
@@ -42,27 +41,51 @@ public Future<Void> publish(K key, DomainEvent<T> event, Map<String, String> oka
     log.info("publish:: key = {}, eventId = {}, type = {}, topic = {}", key, event.getId(),
       event.getType(), kafkaTopic);
 
-    KafkaProducerRecord<K, String> producerRecord =
-      new KafkaProducerRecordBuilder<K, DomainEvent<T>>(TenantTool.tenantId(okapiHeaders))
-        .key(key)
-        .value(event)
-        .topic(kafkaTopic)
-        .propagateOkapiHeaders(okapiHeaders)
-        .build();
+    KafkaProducerRecord<K, String> producerRecord = buildProducerRecord(key, event, okapiHeaders);
     log.info("publish:: kafkaRecord = [{}]", producerRecord);
 
-    KafkaProducer<K, String> producer = getOrCreateProducer();
-    log.info("publish:: Producer created, sending the record...");
+    KafkaProducer<K, String> producer = null;
+    try {
+      producer = getOrCreateProducer();
+      log.info("publish:: Producer created, sending the record...");
+      send(producer, key, producerRecord);
+    } catch (Exception e) {
+      log.error("publish:: Failed to initiate send for domain event with key [{}], kafka record [{}]",
+        key, producerRecord, e);
+      if (producer != null) {
+        log.info("publish:: Producer is not null, trying to close. Event key: {}.", key);
+        producer.close();
+      }
+      failureHandler.handle(e, producerRecord);
+    }
+
+    return succeededFuture();
+  }
+
+  private KafkaProducerRecord<K, String> buildProducerRecord(K key, DomainEvent<T> event,
+    Map<String, String> okapiHeaders) {
+
+    return new KafkaProducerRecordBuilder<K, DomainEvent<T>>(TenantTool.tenantId(okapiHeaders))
+      .key(key)
+      .value(event)
+      .topic(kafkaTopic)
+      .propagateOkapiHeaders(okapiHeaders)
+      .build();
+  }
+
+  private void send(KafkaProducer<K, String> producer, K key,
+    KafkaProducerRecord<K, String> producerRecord) {
 
-    return producer.send(producerRecord)
-      .onSuccess(r -> log.info("publish:: Succeeded sending domain event with key [{}], " +
+    producer.send(producerRecord)
+      .onSuccess(r -> log.info("send:: Succeeded sending domain event with key [{}], " +
         "kafka record [{}]", key, producerRecord))
-      .<Void>mapEmpty()
       .onFailure(cause -> {
-        log.error("publish:: Unable to send domain event with key [{}], kafka record [{}]",
+        log.error("send:: Unable to send domain event with key [{}], kafka record [{}]",
           key, producerRecord, cause);
         failureHandler.handle(cause, producerRecord);
-      });
+      })
+      .eventually(() -> producer.flush())
+      .eventually(() -> producer.close());
   }
 
   private KafkaProducer<K, String> getOrCreateProducer() {