weiqingy commented on code in PR #944:
URL: https://github.com/apache/flink-agents/pull/944#discussion_r3725959384
##########
runtime/src/test/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperatorTest.java:
##########
@@ -75,6 +81,40 @@ void resetReconcilableFixtures() {
TestAgent.resetMixedRecoveryFixture();
}
+ @Test
+ void closeAttemptsAllResourcesAndSuppressesLaterFailures() throws
Exception {
+ AgentPlan plan = new AgentPlan(new HashMap<>(), new HashMap<>());
+ ActionStateStore actionStateStore = mock(ActionStateStore.class);
+ ActionExecutionOperator<Object, Object> operator =
+ new ActionExecutionOperator<>(plan, true, null, null,
actionStateStore);
+ ResourceCache resourceCache = mock(ResourceCache.class);
+ ActionTaskContextManager contextManager =
mock(ActionTaskContextManager.class);
+ PythonBridgeManager pythonBridge = mock(PythonBridgeManager.class);
+ RuntimeException resourceFailure = new RuntimeException("resource
cache close failed");
+ RuntimeException contextFailure = new RuntimeException("context
manager close failed");
+ RuntimeException bridgeFailure = new RuntimeException("python bridge
close failed");
+ RuntimeException durableFailure = new RuntimeException("durable
manager close failed");
+
+ doThrow(resourceFailure).when(resourceCache).close();
+ doThrow(contextFailure).when(contextManager).close();
+ doThrow(bridgeFailure).when(pythonBridge).close();
+ doThrow(durableFailure).when(actionStateStore).close();
+ setPrivateField(operator, "resourceCache", resourceCache);
+ setPrivateField(operator, "contextManager", contextManager);
+ setPrivateField(operator, "pythonBridge", pythonBridge);
+
+ assertThatThrownBy(operator::close)
+ .isSameAs(resourceFailure)
+ .hasSuppressedException(contextFailure)
+ .hasSuppressedException(bridgeFailure)
+ .hasSuppressedException(durableFailure);
+ InOrder closeOrder = inOrder(resourceCache, contextManager,
pythonBridge, actionStateStore);
Review Comment:
This exercises six closeables and verifies four. `eventRouter` and
`super::close` are never injected or asserted.
I tried deleting each from the `closeAll(...)` list, and all 407
runtime-module tests stay green either way. Dropping `super::close` would
silently stop disposing Flink operator state, which is the costlier of the two.
`eventRouter` looks cheap to cover: it is `private final transient`, and the
`setPrivateField` helper this test already uses works on non-static final
instance fields, so a throwing mock could join the `InOrder`. Any appetite for
that?
`super.close()` I don't have a cheap way to assert, so I'd call that half a
known gap rather than something to chase here.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/CloseableUtils.java:
##########
@@ -0,0 +1,32 @@
+/*
+ * 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.operator;
+
+import org.apache.flink.util.LambdaUtil;
+
+import java.util.Arrays;
+
+final class CloseableUtils {
+
+ private CloseableUtils() {}
+
+ static void closeAll(AutoCloseable... closeables) throws Exception {
Review Comment:
nit: `org.apache.flink.util.IOUtils.closeAll(AutoCloseable...) throws
Exception` already exists with the same signature, and ships in both flink-core
1.20.0 and 2.3.0, so this isn't a cross-version gap. It has the semantics this
helper relies on: null-guards, skips null elements, aggregates through
`ExceptionUtils.firstOrSuppressed`, rethrows. The repo already uses it at
`runtime/src/main/java/.../feedback/Checkpoints.java:22,63`.
Moving the helper out of `OperatorUtils` makes sense, since that class does
have a per-version variant under `dist/flink-1.20/`. Is there a reason to
prefer a local helper over the Flink one?
If it did become `IOUtils.closeAll`, one knock-on: `CloseableUtils` is
package-private in `operator`, so `PythonActionExecutor.close()`
(`PythonActionExecutor.java:200-214`) still hand-rolls the same
`firstOrSuppressed` accumulation and can't reach it. The public utility would
let that collapse too.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/python/utils/PythonActionExecutor.java:
##########
@@ -188,17 +189,32 @@ public boolean callPythonAwaitable(String
pythonAwaitableRef) {
}
public void close() throws Exception {
- if (interpreter != null) {
- if (pythonAsyncThreadPool != null) {
- interpreter.invoke(CLOSE_ASYNC_THREAD_POOL,
pythonAsyncThreadPool);
- }
+ PyObject asyncThreadPool = pythonAsyncThreadPool;
+ PyObject runnerContext = pythonRunnerContext;
+ pythonAsyncThreadPool = null;
+ pythonRunnerContext = null;
+
+ Exception exception = null;
+ try {
+ closePythonObject(CLOSE_ASYNC_THREAD_POOL, asyncThreadPool);
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e, exception);
+ }
+ try {
+ closePythonObject(CLOSE_FLINK_RUNNER_CONTEXT, runnerContext);
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e, exception);
+ }
+
+ if (exception != null) {
+ throw exception;
Review Comment:
Both frames check out. This is about a third instance rather than those two.
`ActionTaskContextManager.close()`
(`runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionTaskContextManager.java:319-330`)
still has the round-1 shape:
```java
if (runnerContext != null) {
try { runnerContext.close(); } finally { runnerContext = null; }
}
if (continuationActionExecutor != null) {
continuationActionExecutor.close();
}
```
It is argument 2 of the operator's own new `closeAll(...)` list, so it sits
inside the chain you just hardened. The throw path is this PR's own failure
case: `RunnerContextImpl.close()` (`context/RunnerContextImpl.java:340-344`)
calls `this.ltm.close()`, which reaches `Mem0LongTermMemory.close()` and a
Pemja `invoke`.
The skipped close is not a no-op everywhere. The JDK 21 variant
(`runtime/src/main/java21/.../async/ContinuationActionExecutor.java:151-153`)
is `asyncExecutor.shutdownNow()`, while the JDK 11 one (`:59`) is empty. So on
JDK 21 a failing LTM close leaks those threads.
Narrow exposure: JDK 21 only, and only when a close is already failing.
Worth giving this frame the same treatment while you are in here?
One wrinkle if you do. `ContinuationActionExecutor.close()` does not declare
`throws`, so it would need `implements AutoCloseable` or a lambda wrapper to go
into the varargs list.
I checked the rest of the chain for the same shape and it is clean, so this
looks like the last one rather than the first of many.
--
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]
