da-daken commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3792071175
##########
runtime/src/main/java21/org/apache/flink/agents/runtime/async/ContinuationActionExecutor.java:
##########
@@ -148,6 +162,142 @@ public <T> T executeAsync(ContinuationContext context,
Supplier<T> supplier) thr
return (T) context.getAsyncResultRef().get();
}
+ /**
+ * Executes all suppliers as one async batch and returns one {@link
Outcome} per supplier.
+ * Supplier failures are captured in their own outcome so one failed
supplier does not abort the
+ * whole batch.
+ *
+ * @param context the continuation context for this action
+ * @param suppliers the suppliers to execute
+ * @param timeout the timeout for the whole batch; null or non-positive
means no timeout
+ * @param <T> the result type
+ * @return outcomes in supplier order
+ */
+ @SuppressWarnings("unchecked")
+ public <T> List<Outcome<T>> executeAllAsync(
+ ContinuationContext context,
+ List<Callable<T>> suppliers,
+ Duration timeout,
+ int maxParallelism)
+ throws Exception {
+ context.clearAsyncState();
+ if (suppliers.isEmpty()) {
+ return List.of();
+ }
+
+ final int batchSize = suppliers.size();
+ CompletableFuture<Outcome<T>>[] slots = new
CompletableFuture[batchSize];
+ boolean[] counted = new boolean[batchSize];
+ int completed = 0;
+ int nextToSubmit = 0;
+ int parallelismLimit = Math.min(Math.max(maxParallelism, 1),
batchSize);
+
+ long deadlineNanos = getDeadlineNanos(timeout);
+ CompletableFuture<Void> batchBarrier = new CompletableFuture<>();
Review Comment:
Good catch. There was a deadlock here, which is exactly why the CI below
timed out. I've already fixed it in a new commit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
