weiqingy commented on code in PR #944:
URL: https://github.com/apache/flink-agents/pull/944#discussion_r3745476785


##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/PythonBridgeManager.java:
##########
@@ -291,14 +294,11 @@ boolean isInitialized() {
 
     @Override
     public void close() throws Exception {
-        if (pythonActionExecutor != null) {
-            pythonActionExecutor.close();
-        }
-        if (pythonInterpreter != null) {
-            pythonInterpreter.close();
-        }
-        if (pythonEnvironmentManager != null) {
-            pythonEnvironmentManager.close();
-        }
+        IOUtils.closeAll(

Review Comment:
   Checked whether the 2-arg overload offers a way to keep the utility. It does 
not, and the reason matters. Against `flink-core-2.3.0`, three closeables, 
first throwing `OutOfMemoryError`:
   
   ```
   closeAll(a1, a2, a3)                        -> java.lang.OutOfMemoryError
                                                  closed: a1=true a2=false 
a3=false
   closeAll(asList(b1,b2,b3), Throwable.class) -> java.lang.Exception: 
java.lang.OutOfMemoryError
                                                  instanceof Error = false
                                                  closed: b1=true b2=true 
b3=true
   ExceptionUtils.rethrowException(new OutOfMemoryError("boom"))
                                               -> java.lang.OutOfMemoryError, 
instanceof Error = true
   ```
   
   `Throwable.class` does close everything, but `closeAll(Iterable, Class)` 
casts the collected throwable only when it is already an `Exception` and 
otherwise does `new Exception(t)`. `Task.preProcessException` unwraps only 
`WrappingRuntimeException` before testing `isJvmFatalError(t) || t instanceof 
OutOfMemoryError` and halting the JVM, so a wrapped OOM stops registering as 
fatal and comes back as an ordinary task failure. So `rethrowException` passing 
the `Error` through unwrapped is doing work no `closeAll` variant can.
   
   On scope, there is a fourth `closeAll` site outside your list: 
`PythonActionExecutor.close()` (`PythonActionExecutor.java:200`). It is the 
pointed one for this PR, since both `PyObject` fields are nulled before the 
aggregate runs, so an `Error` out of the async-thread-pool release leaves the 
runner-context handle unclosed and unreachable. That is the exact leak this PR 
exists to close. Should that site be in scope for whichever PR carries the 
ladder?
   



-- 
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]

Reply via email to