Java: Fix reactor flatMap producing out of order results (#1849)

Fixes a concurrency bug, `Flux.flatMap` will return the first available result rather than keeping the results in order, changed to concatMap
microsoft · Jul 8, 2023 · 470dae2 · 470dae2
1 parent 0e63f3b
commit 470dae2
Show file tree

Hide file tree

Showing 3 changed files with 3 additions and 3 deletions.
diff --git a/...src/main/java/com/microsoft/semantickernel/orchestration/DefaultCompletionSKFunction.java b/...src/main/java/com/microsoft/semantickernel/orchestration/DefaultCompletionSKFunction.java
@@ -126,7 +126,7 @@ public Mono<SKContext> aggregatePartitionedResultsAsync(
         // function
         BiFunction<Flux<SKContext>, String, Flux<SKContext>> executeNextChunk =
                 (contextInput, input) ->
-                        contextInput.flatMap(
+                        contextInput.concatMap(
                                 newContext -> {
                                     SKContext updated = newContext.update(input);
                                     return invokeAsync(updated, null);

diff --git a/...rc/main/java/com/microsoft/semantickernel/templateengine/DefaultPromptTemplateEngine.java b/...rc/main/java/com/microsoft/semantickernel/templateengine/DefaultPromptTemplateEngine.java
@@ -69,7 +69,7 @@ public Mono<String> renderAsync(String templateText, SKContext context) {
     /// <inheritdoc/>
     public Mono<String> renderAsync(List<Block> blocks, SKContext context) {
         return Flux.fromIterable(blocks)
-                .flatMap(
+                .concatMap(
                         block -> {
                             if (block instanceof TextRendering) {
                                 return Mono.just(

diff --git a/...microsoft/semantickernel/planner/sequentialplanner/DefaultSequentialPlannerSKContext.java b/...microsoft/semantickernel/planner/sequentialplanner/DefaultSequentialPlannerSKContext.java
@@ -202,7 +202,7 @@ Mono<SKContext> rememberFunctionsAsync(List<SKFunction<?>> availableFunctions) {
         }
 
         return Flux.fromIterable(availableFunctions)
-                .flatMap(
+                .concatMap(
                         function -> {
                             String functionName = function.toFullyQualifiedName();
                             String key = functionName;