Repository navigation
[runtime] Add comprehensive tests for ActionExecutionOperator manager classes - #659
Merged
Merged
Conversation
github-actions
Bot
added
doc-not-needed
Adds three narrow seams to production code so that follow-up commits covering DurableExecutionManager, EventRouter, and ActionTaskContextManager have observable targets for their contract assertions. No behaviour change for production callers. * DurableExecutionManager.getCheckpointIdToSeqNums() accessor — exposes the in-heap snapshot map used by snapshotState and notifyCheckpointComplete. Follows the existing getActionStateStore() pattern. * EventRouter.addEventListener(EventListener) accessor — current eventListeners list has no public registration path; required by listener-notification tests. * EventRouter constructor refactor — existing 2-arg constructor delegates to a new 3-arg @VisibleForTesting constructor that accepts a pre-built EventLogger. Allows tests to inject a mock(EventLogger.class) and avoid file I/O. createEventLogger is converted from instance method to static (required by Java for use in a chained this(...) call). Part of apache#646.
Closes the snapshot/recovery + pruning coverage gap deferred from PR apache#546 (issue apache#646, row 2 of the deferred-coverage table). * snapshotCapturesKeySequenceNumbers — mocks KeyedStateBackend.applyToAllKeys via Mockito doAnswer and verifies the in-heap checkpointIdToSeqNums map is populated per-key. * notifyPrunesNotifiedCheckpointAndPreservesOthers — populates two checkpoints in the map ({1000L: {k1, k2}} and {1001L: {k3}}), notifies completion of 1000L, and asserts: k1 and k2 are pruned, k3 survives (selectivity holds), the 1000-entry is removed, the 1001-entry remains. Combines multi-key (loop body) and multi-checkpoint (isolation) concerns into one strictly stronger assertion than two separate tests. * handleRecoveryCallsRebuildState — spy on InMemoryActionStateStore; mock OperatorStateBackend.getUnionListState to return a marker list; verify rebuildState(markers) is invoked with the expected payload. The handleRecovery test verifies the call contract; mutation observability for the rebuildState side effect is covered by operator-level integration tests against real stores (InMemoryActionStateStore.rebuildState is a no-op). Uses the @VisibleForTesting getCheckpointIdToSeqNums() accessor added in the previous commit. No production-code changes. Part of apache#646.
Closes the listener-notification + watermark-draining coverage gap deferred from PR apache#546 (issue apache#646, row 3 of the deferred-coverage table). * notifyEventProcessedNotifiesAllListeners — multi-listener fan-out via the @VisibleForTesting addEventListener seam. * notifyEventProcessedIncrementsMetric — spy on BuiltInMetrics verifies markEventProcessed() is invoked once per event. * notifyEventProcessedAppendsAndFlushesLogger — Mockito InOrder verifies append() is called before flush(); logger injected via the 3-arg @VisibleForTesting constructor as a mock(EventLogger). * processEligibleWatermarksDrainsInArrivalOrder — drives SegmentedQueue across two segments and removes keys in completion order that differs from arrival; asserts ordered emit via containsExactly(WM1, WM2). * processEligibleWatermarksBlocksWhileSegmentHasKeys — confirms no watermark is emitted while a key in its segment is still in flight, then verifies emit after the key is removed. Extracts a private static makeMetrics() helper for the 5 tests that need BuiltInMetrics. Uses the addEventListener and 3-arg constructor seams added in commit 1. No production-code changes. Part of apache#646.
Closes the memory-context lifecycle coverage gap deferred from PR apache#546 (issue apache#646, row 4 of the deferred-coverage table). Maps the issue body's four phases (create → transfer → complete → cleanup) to concrete code points. * createAndSetRunnerContextBuildsFreshMemoryContextOnFirstCall — first call wires a JavaRunnerContextImpl and creates a fresh MemoryContext. * createAndSetRunnerContextReusesExistingMemoryContext — second call reuses the map entry that transferContexts populated. * transferContextsCopiesMemoryAndContinuationToNewTask — transfer populates the to-task and leaves the from-task's manager-map entries untouched. Asserts the remove-then-null contract for the memory context. Documents the shared-responsibility cleanup model: transferContexts is a pure copy; the operator drives from-task cleanup at ActionExecutionOperator.java:366-369. * transferContextsRoutesDurableContextThroughManager — spy on DurableExecutionManager verifies the durable-context branch routes through putDurableContext(toTask, ...). * closeIsIdempotent — close() can be called twice without exception, matching ContinuationActionExecutor.shutdownNow() semantics. Extracts a private static invokeCreateAndSetRunnerContext helper (4 use sites) to factor mocked-collaborator boilerplate. No production-code changes; uses existing public/package-private API. Part of apache#646.
Collaborator
Author
|
Hey @wenjin272, this PR follows up on the review comment about adding more tests for #546. Could you please review it when you get a chance? Thanks! |
Collaborator
Author
|
The 1 remaining failing check is unrelated to this PR — it's an
The same flake hit this PR earlier (run |
8 of 10 tasks
weiqingy
added a commit
to weiqingy/flink-agents
that referenced
this pull request
May 15, 2026
…ntries Issue apache#665. When Flink aborts a checkpoint, it calls notifyCheckpointAborted instead of notifyCheckpointComplete. The DurableExecutionManager only handled the complete path, so the per-checkpoint sequence-number entry recorded by snapshotLastCompletedSequenceNumbers was never released for aborted checkpoints. Under sustained abort pressure (timeouts, alignment failures, backend pressure), checkpointIdToSeqNums grew unboundedly. Changes: - Add DurableExecutionManager.notifyCheckpointAborted(long): removes the entry from checkpointIdToSeqNums, guarded by the same actionStateStore != null check as notifyCheckpointComplete. Does NOT prune durable action state — the aborted checkpoint's writes were never committed, so the prior committed checkpoint's recovery state is still load-bearing and must not be pruned. - Add ActionExecutionOperator.notifyCheckpointAborted(long): thin override that delegates to the manager and then calls super, mirroring the existing notifyCheckpointComplete override. - Extend the symmetric-guard invariant javadoc on snapshotLastCompletedSequenceNumbers and notifyCheckpointComplete to name both release paths (complete OR abort). The actionStateStore != null guard now lives on three methods; the cross-linked javadoc makes that explicit and cites issues apache#645 and apache#665. - Three new DurableExecutionManagerTest cases (using the existing getCheckpointIdToSeqNums() @VisibleForTesting accessor introduced in apache#659): * notifyAbortedRemovesEntryWithoutPruning — entry released, durable state untouched (verified against a real InMemoryActionStateStore so wrongful pruning would be observable). * completedAndAbortedInterleavedKeepsInFlightEntries — three in-flight checkpoints, one completes (state pruned), one aborts (state preserved), one remains. * noStoreModeNotifyCheckpointAbortedIsNoOp — symmetric null-store no-op coverage matching the existing notifyCheckpointComplete null-store case.
weiqingy
added a commit
to weiqingy/flink-agents
that referenced
this pull request
May 15, 2026
…ntries Issue apache#665. When Flink aborts a checkpoint, it calls notifyCheckpointAborted instead of notifyCheckpointComplete. The DurableExecutionManager only handled the complete path, so the per-checkpoint sequence-number entry recorded by snapshotLastCompletedSequenceNumbers was never released for aborted checkpoints. Under sustained abort pressure (timeouts, alignment failures, backend pressure), checkpointIdToSeqNums grew unboundedly. Changes: - Add DurableExecutionManager.notifyCheckpointAborted(long): removes the entry from checkpointIdToSeqNums, guarded by the same actionStateStore != null check as notifyCheckpointComplete. Does NOT prune durable action state — the aborted checkpoint's writes were never committed, so the prior committed checkpoint's recovery state is still load-bearing and must not be pruned. - Add ActionExecutionOperator.notifyCheckpointAborted(long): thin override that delegates to the manager and then calls super, mirroring the existing notifyCheckpointComplete override. - Extend the symmetric-guard invariant javadoc on snapshotLastCompletedSequenceNumbers and notifyCheckpointComplete to name both release paths (complete OR abort). The actionStateStore != null guard now lives on three methods; the cross-linked javadoc makes that explicit and cites issues apache#645 and apache#665. - Three new DurableExecutionManagerTest cases (using the existing getCheckpointIdToSeqNums() @VisibleForTesting accessor introduced in apache#659): * notifyAbortedRemovesEntryWithoutPruning — entry released, durable state untouched (verified against a real InMemoryActionStateStore so wrongful pruning would be observable). * completedAndAbortedInterleavedKeepsInFlightEntries — three in-flight checkpoints, one completes (state pruned), one aborts (state preserved), one remains. * noStoreModeNotifyCheckpointAbortedIsNoOp — symmetric null-store no-op coverage matching the existing notifyCheckpointComplete null-store case.
wenjin272
pushed a commit
that referenced
this pull request
May 16, 2026
…ntries (#667) Issue #665. When Flink aborts a checkpoint, it calls notifyCheckpointAborted instead of notifyCheckpointComplete. The DurableExecutionManager only handled the complete path, so the per-checkpoint sequence-number entry recorded by snapshotLastCompletedSequenceNumbers was never released for aborted checkpoints. Under sustained abort pressure (timeouts, alignment failures, backend pressure), checkpointIdToSeqNums grew unboundedly. Changes: - Add DurableExecutionManager.notifyCheckpointAborted(long): removes the entry from checkpointIdToSeqNums, guarded by the same actionStateStore != null check as notifyCheckpointComplete. Does NOT prune durable action state — the aborted checkpoint's writes were never committed, so the prior committed checkpoint's recovery state is still load-bearing and must not be pruned. - Add ActionExecutionOperator.notifyCheckpointAborted(long): thin override that delegates to the manager and then calls super, mirroring the existing notifyCheckpointComplete override. - Extend the symmetric-guard invariant javadoc on snapshotLastCompletedSequenceNumbers and notifyCheckpointComplete to name both release paths (complete OR abort). The actionStateStore != null guard now lives on three methods; the cross-linked javadoc makes that explicit and cites issues #645 and #665. - Three new DurableExecutionManagerTest cases (using the existing getCheckpointIdToSeqNums() @VisibleForTesting accessor introduced in #659): * notifyAbortedRemovesEntryWithoutPruning — entry released, durable state untouched (verified against a real InMemoryActionStateStore so wrongful pruning would be observable). * completedAndAbortedInterleavedKeepsInFlightEntries — three in-flight checkpoints, one completes (state pruned), one aborts (state preserved), one remains. * noStoreModeNotifyCheckpointAbortedIsNoOp — symmetric null-store no-op coverage matching the existing notifyCheckpointComplete null-store case.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Linked issue: #646
Purpose of change
Closes the heavier test coverage explicitly deferred at PR #546's review for the manager classes that decomposed
ActionExecutionOperator. PR #546 added 4 contract test files (7 tests, 295 lines) using mocks and stubs; heavier coverage — snapshot/recovery cycle, listener notification, watermark draining order, memory-context lifecycle — was deferred and tracked in #646. This PR closes that gap.Scope (matches issue #646's deferred-coverage table):
DurableExecutionManagernotifyCheckpointCompletepruning behaviour (3 new tests)EventRouterActionTaskContextManagerOperatorStateManageris dropped from this PR — its observable contracts are already covered byActionExecutionOperatorTest.java:176–231(rescale + recovery) and:254–306(durable state recovery). Adding a manager-scoped harness test would duplicate without adding coverage.PythonBridgeManageris out of scope per the issue body's deferred-coverage table.Commit structure (4 commits)
DurableExecutionManager.getActionStateStore()precedent:DurableExecutionManager.getCheckpointIdToSeqNums()accessorEventRouter.addEventListener(EventListener)accessorEventRouter3-arg@VisibleForTestingconstructor forEventLoggerinjection (2-arg constructor delegates; production signature unchanged)Tests
runtimemodule: 262 tests, 0 failures, 0 errors, 0 skipped.### FORMAT CHECK PASSED ###.Verification commands:
Final diff: 5 files, +464 / -2 (2 production-code files for the seams; 3 test files for the new tests).
Test design notes (each test catches a distinct mutation)
DurableExecutionManagerTest.notifyPrunesNotifiedCheckpointAndPreservesOthersis a single consolidated test that exercises BOTH the multi-key loop body (two keys under one checkpoint, both must be pruned) AND multi-checkpoint isolation (a third key under a different checkpoint must survive). Strictly stronger than two separate tests.DurableExecutionManagerTest.handleRecoveryCallsRebuildStateverifies the call contract; mutation observability for therebuildStateside effect is covered by operator-level integration tests against real stores (InMemoryActionStateStore.rebuildStateis a no-op).ActionTaskContextManagerTest.transferContextsCopiesMemoryAndContinuationToNewTaskcovers the Java path; the Python path requires Pemja and is covered byActionExecutionOperatorTest's Python flows.EventRouterTest.processEligibleWatermarksDrainsInArrivalOrderdrivesSegmentedQueueacross two segments and removes keys in completion order that differs from arrival; ordered assertioncontainsExactly(WM1, WM2)catches both wrong-order draining and skipped watermarks.API
No public API changes. Three new
@VisibleForTestingpackage-private members in production code (one accessor onDurableExecutionManager, one accessor + one constructor overload onEventRouter). The existingEventRouter(AgentPlan, boolean)constructor signature is unchanged; it delegates to the new 3-arg form.createEventLoggeris converted from instance tostatic(required by Java for chainedthis(...)constructors; the original body uses only its parameter).Documentation
doc-neededdoc-not-neededdoc-included