From 05d8944e74a968a29554aa5438b94fdcbe2aa2aa Mon Sep 17 00:00:00 2001
From: Yoann Vernageau <6807151+yvrng@users.noreply.github.com>
Date: Wed, 4 Sep 2024 17:51:54 +0200
Subject: [PATCH] fix: handle processing of an empty list or file

---
 src/main/java/io/kestra/plugin/kafka/Produce.java | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/src/main/java/io/kestra/plugin/kafka/Produce.java b/src/main/java/io/kestra/plugin/kafka/Produce.java
index 57fe91d..2d2c6a5 100644
--- a/src/main/java/io/kestra/plugin/kafka/Produce.java
+++ b/src/main/java/io/kestra/plugin/kafka/Produce.java
@@ -205,7 +205,7 @@ public Output run(RunContext runContext) throws Exception {
 
                         count = resultFlowable
                             .reduce(Integer::sum)
-                            .block();
+                            .blockOptional().orElse(0);
                     }
                 } else {
                     flowable = Flux.fromArray(((List<Object>) this.from).toArray());
@@ -213,7 +213,7 @@ public Output run(RunContext runContext) throws Exception {
 
                     count = resultFlowable
                         .reduce(Integer::sum)
-                        .block();
+                        .blockOptional().orElse(0);
                 }
             } else {
                 producer.send(this.producerRecord(runContext, producer, (Map<String, Object>) this.from));