weiqingy commented on code in PR #926:
URL: https://github.com/apache/flink-agents/pull/926#discussion_r3793394857
##########
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:
Checked this at head, including the timeout path. Closed from my side.
##########
python/flink_agents/api/runner_context.py:
##########
@@ -24,12 +25,49 @@
from flink_agents.api.metric_group import MetricGroup
from flink_agents.api.resource import Resource, ResourceType
-__all__ = ["AsyncExecutionResult", "RunnerContext"]
+__all__ = ["AsyncExecutionResult", "DurableCall", "Outcome", "RunnerContext"]
if TYPE_CHECKING:
from flink_agents.api.memory_object import MemoryObject
+@dataclass(frozen=True)
+class DurableCall:
+ """A deterministic durable call entry for batch execution."""
+
+ id: str
Review Comment:
Thanks for opening #1016, that covers it.
--
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]
