emecii commented on code in PR #987:
URL: https://github.com/apache/flink-agents/pull/987#discussion_r3744750057
##########
runtime/src/test/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperatorTest.java:
##########
@@ -525,6 +529,65 @@ private static void replaceOperatorLtm(
ltmField.set(operator, ltm);
}
+ private static void replaceOperatorField(
+ ActionExecutionOperator<?, ?> operator, String name, Object value)
throws Exception {
+ Field field = ActionExecutionOperator.class.getDeclaredField(name);
+ field.setAccessible(true);
+ field.set(operator, value);
+ }
+
+ /**
+ * A failing component must not strand the ones behind it. This matters
most for {@code
+ * resourceCache}, which closes first and aggregates its own failures, and
for {@code
+ * pythonBridge}, which releases the embedded Python interpreter.
+ */
+ @Test
+ void closeClosesEveryComponentWhenAnEarlierCloseFails() throws Exception {
+ KeyedOneInputStreamOperatorTestHarness<Long, Long, Object> testHarness
=
+ new KeyedOneInputStreamOperatorTestHarness<>(
+ new
ActionExecutionOperatorFactory(TestAgent.getAgentPlan(false), true),
+ (KeySelector<Long, Long>) value -> value,
+ TypeInformation.of(Long.class));
+ testHarness.open();
+ ActionExecutionOperator<Long, Object> operator =
+ (ActionExecutionOperator<Long, Object>)
testHarness.getOperator();
+
+ ResourceCache resourceCache = mock(ResourceCache.class);
+ ActionTaskContextManager contextManager =
mock(ActionTaskContextManager.class);
+ PythonBridgeManager pythonBridge = mock(PythonBridgeManager.class);
+ EventRouter<Long, Object> eventRouter = mock(EventRouter.class);
+ DurableExecutionManager durableExecManager =
mock(DurableExecutionManager.class);
+ doThrow(new IllegalStateException("resource cache close failed"))
+ .when(resourceCache)
+ .close();
+
+ replaceOperatorField(operator, "resourceCache", resourceCache);
+ replaceOperatorField(operator, "contextManager", contextManager);
+ replaceOperatorField(operator, "pythonBridge", pythonBridge);
+ replaceOperatorField(operator, "eventRouter", eventRouter);
+ replaceOperatorField(operator, "durableExecManager",
durableExecManager);
+
+ try {
+ assertThatThrownBy(operator::close)
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage("resource cache close failed");
+
+ // The components behind the failing one are still released.
+ verify(contextManager).close();
+ verify(pythonBridge).close();
+ verify(eventRouter).close();
+ verify(durableExecManager).close();
Review Comment:
You are right on both counts, and I confirmed your reasoning by running it:
against the code as it stood, deleting the try/catch at `:579-583` and
replacing it with `if (firstFailure == null) super.close();` both left this
test green. The second one especially should not have — it is a partial revert
of the fix. Contract 7 was not being tested. Fixed in 9b5688d0.
Rather than drop it from the table, I think the aggregation half is
reachable — you stopped one step short. `super.close()` does bind statically,
but its *effect* does not: `AbstractStreamOperator.close()` compiles to
`stateHandler.dispose()` (verified in `flink-runtime-2.3.0` bytecode),
`stateHandler` is a `protected` field, and `StreamOperatorStateHandler` is a
public non-final class with a public non-final `dispose()`. Swapping the
inherited handler therefore makes the super call both observable and failable —
no subclass interception needed.
So:
- `closeClosesEveryComponentWhenAnEarlierCloseFails` now asserts `dispose()`
ran.
- A new `closeAggregatesSuperCloseFailureWithComponentFailure` makes
`dispose()` throw and asserts the `resourceCache` failure still reaches the
caller with the super failure attached as suppressed — which also pins the
`firstOrSuppressed` argument order, the easy thing to get backwards.
The real handler is restored in the `finally` before teardown, so the
harness still disposes it for real.
Both of your mutations now fail both tests. While in here I also applied
your `InOrder` point from the other thread to these two tests: order is
documented even more explicitly on this path (`resourceCache` before
`pythonBridge`, per the comment at `:556`), and swapping those two left the
tests green until I did. The chain now covers all five components plus
`stateHandler.dispose()` last.
--
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]
