Merge pull request #439 from kestra-io/fix/fix-npe-with-result-flowab…

…le-block fix(): changed block() to blockOptional()
kestra-io · Sep 5, 2024 · def5c47 · def5c47
2 parents 80e7fcd + c02565d
commit def5c47
Showing 1 changed file with 2 additions and 2 deletions.
diff --git a/src/main/java/io/kestra/plugin/gcp/pubsub/Publish.java b/src/main/java/io/kestra/plugin/gcp/pubsub/Publish.java
@@ -91,7 +91,7 @@ public Publish.Output run(RunContext runContext) throws Exception {
                 flowable = FileSerde.readAll(inputStream, Message.class);
                 resultFlowable = this.buildFlowable(flowable, publisher, runContext);
 
-                count = resultFlowable.reduce(Integer::sum).block();
+                count = resultFlowable.reduce(Integer::sum).blockOptional().orElse(0);
             }
 
         } else if (this.from instanceof List) {
@@ -101,7 +101,7 @@ public Publish.Output run(RunContext runContext) throws Exception {
 
             resultFlowable = this.buildFlowable(flowable, publisher, runContext);
 
-            count = resultFlowable.reduce(Integer::sum).block();
+            count = resultFlowable.reduce(Integer::sum).blockOptional().orElse(0);
         } else {
             var msg = JacksonMapper.toMap(this.from, Message.class);
             publisher.publish(msg.to(runContext, this.serdeType));