pltbkd commented on code in PR #938: URL: https://github.com/apache/flink-agents/pull/938#discussion_r3765824181
########## runtime/src/main/java/org/apache/flink/agents/runtime/subagent/AsyncSubagentFuture.java: ########## @@ -0,0 +1,131 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.agents.runtime.subagent; + +import org.apache.flink.agents.api.context.RunnerContext; +import org.apache.flink.agents.api.subagent.Result; +import org.apache.flink.agents.api.subagent.SubagentFuture; +import org.apache.flink.agents.api.subagent.SubagentFutures; +import org.apache.flink.agents.runtime.subagent.BaseAsyncSubagentSetup.RunStatus; + +import javax.annotation.Nullable; + +import java.util.concurrent.CancellationException; + +/** + * The sub side of an async-job invocation: the run was already started by the durable POST of + * {@code submit}, so the handle only subscribes to it. {@link #isDone()} probes the status directly + * (not durable); {@link #await()} waits through the durable await composition; {@link #cancel()} + * propagates the cancellation through the setup's hook, and a cancelled {@code await} fails as a + * {@link CancellationException}. + * + * <p>The handle records itself in the owning base's per-task registry, so dropping it without + * collecting the outcome fails the action instead of silently losing the result. + */ +final class AsyncSubagentFuture extends SubagentFuture { + + private final BaseAsyncSubagentSetup setup; + private final RunnerContext ctx; + @Nullable private final PendingSubagentCallRegistry registry; + + private boolean consumed; + private boolean cancelled; + @Nullable private Result value; + + AsyncSubagentFuture( + BaseAsyncSubagentSetup setup, + RunnerContext ctx, + String sessionId, + String callId, + @Nullable PendingSubagentCallRegistry registry) { + super(sessionId, callId); + this.setup = setup; + this.ctx = ctx; + this.registry = registry; + if (registry != null) { + registry.trackPendingSubagentCall(identity()); + } + } + + /** + * Probes the remote status directly; not durable, so a failover replay may probe a different + * number of times than the original execution. + */ + @Override + public boolean isDone() { + if (consumed || cancelled) { + return true; + } + try { + RunStatus probe = setup.queryStatus(getSessionId(), getCallId()); + return probe.getState() == RunStatus.State.COMPLETED + || probe.getState() == RunStatus.State.FAILED; + } catch (Exception e) { Review Comment: Agreed, the swallow is not right here. The contract should be: the implementation handles the failures `queryStatus` can understand and reports them as a FAILED status, your examples fall in this category; an exception that still escapes is a system-level failure the implementation can't handle. While we generally want sub-agent exceptions kept away from the main agent, swallowing such an exception (even with a log) leaves the await waiting forever, while converting it into a failed result may diverge from what the remote side actually did. For this version, the safest choice is to throw. I plan to drop throws along the queryStatus path and the try-catch here, and ask implementors to report such failures as a RuntimeException. -- 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]
