Adamyuanyuan commented on PR #10208:
URL: https://github.com/apache/seatunnel/pull/10208#issuecomment-3857552556
> ### Issue 1: Missing complete test for checkpoint recovery scenarios
> **Location**:
`seatunnel-translation/seatunnel-translation-flink/seatunnel-translation-flink-common/src/test/java/org/apache/seatunnel/translation/flink/source/FlinkSourceEnumeratorTest.java`
>
> ```java
> @Test
> void testResignalNoMoreSplitsAfterReaderReregister() {
> SourceSplitEnumerator<DummySplit, Serializable> sourceSplitEnumerator =
> Mockito.mock(SourceSplitEnumerator.class);
> SplitEnumeratorContext<SplitWrapper<DummySplit>> enumeratorContext =
> Mockito.mock(SplitEnumeratorContext.class);
> Mockito.when(enumeratorContext.currentParallelism()).thenReturn(2);
>
> Set<Integer> noMoreSplitsSignaledReaders =
ConcurrentHashMap.newKeySet();
> noMoreSplitsSignaledReaders.add(0); // Manually added
>
> FlinkSourceEnumerator<DummySplit, Serializable> enumerator =
> new FlinkSourceEnumerator<>(
> sourceSplitEnumerator, enumeratorContext,
noMoreSplitsSignaledReaders);
>
> enumerator.addReader(0);
>
> Mockito.verify(enumeratorContext).signalNoMoreSplits(0);
> }
> ```
>
> **Related Context**:
>
> * Test class: `FlinkSourceEnumeratorTest.java:58`
> * Modified class: `FlinkSourceEnumerator.java:98`
> * Tested SeaTunnel Engine: `SourceSplitEnumeratorTaskTest.java:109`
>
> **Problem Description**:
>
> The current test case is too simple, manually adding subtask 0 to the
`noMoreSplitsSignaledReaders` collection, which does not realistically simulate
the failure recovery scenario.
>
> In a real failure recovery scenario:
>
> 1. `FlinkSource.restoreEnumerator()` is called
> 2. A **new** `FlinkSourceEnumerator` instance is created, where
`noMoreSplitsSignaledReaders` is an **empty collection**
> 3. Reader re-registers
> 4. `run()` is re-executed (because `isRun = false`)
> 5. `run()` calls `context.signalNoMoreSplits()` in it, adding to the
collection
> 6. Re-sending is triggered only if the reader registers again
>
> The current test does not cover this complete flow.
>
> **Potential Risks**:
>
> * **Risk 1**: The test cannot verify real failure recovery scenarios and
may miss boundary condition issues
> * **Risk 2**: If someone modifies `restoreEnumerator` logic in the future
(e.g., changing to a non-empty collection), the test cannot capture regression
issues
> * **Risk 3**: The test gives developers a false impression that failure
recovery scenarios have been fully tested
>
> **Impact Scope**:
>
> * **Direct Impact**: `FlinkSourceEnumeratorTest` test coverage is
incomplete
> * **Indirect Impact**: May not detect failure recovery logic bugs in time
> * **Affected Area**: Flink Translation layer
>
> **Severity**: Medium
>
> **Improvement Suggestions**:
>
> ```java
> @Test
> void testResignalNoMoreSplitsAfterReaderReregister() {
> SourceSplitEnumerator<DummySplit, Serializable> sourceSplitEnumerator =
> Mockito.mock(SourceSplitEnumerator.class);
> SplitEnumeratorContext<SplitWrapper<DummySplit>> enumeratorContext =
> Mockito.mock(SplitEnumeratorContext.class);
> Mockito.when(enumeratorContext.currentParallelism()).thenReturn(2);
>
> Set<Integer> noMoreSplitsSignaledReaders =
ConcurrentHashMap.newKeySet();
> FlinkSourceEnumerator<DummySplit, Serializable> enumerator =
> new FlinkSourceEnumerator<>(
> sourceSplitEnumerator, enumeratorContext,
noMoreSplitsSignaledReaders);
>
> // Simulate reader registration and run() execution
> enumerator.addReader(0);
> enumerator.addReader(1);
>
> // Reset mock, clear previous calls
> Mockito.reset(enumeratorContext);
>
> // Simulate reader re-registration (fault recovery scenario)
> enumerator.addReader(0);
>
> // Verify signal was re-sent
> Mockito.verify(enumeratorContext).signalNoMoreSplits(0);
> }
>
> @Test
> void testNoResignalWhenRunNotCompleted() {
> // Test re-registration before run() completes, should not re-send
signal
> SourceSplitEnumerator<DummySplit, Serializable> sourceSplitEnumerator =
> Mockito.mock(SourceSplitEnumerator.class);
> SplitEnumeratorContext<SplitWrapper<DummySplit>> enumeratorContext =
> Mockito.mock(SplitEnumeratorContext.class);
> Mockito.when(enumeratorContext.currentParallelism()).thenReturn(2);
>
> Set<Integer> noMoreSplitsSignaledReaders =
ConcurrentHashMap.newKeySet();
> FlinkSourceEnumerator<DummySplit, Serializable> enumerator =
> new FlinkSourceEnumerator<>(
> sourceSplitEnumerator, enumeratorContext,
noMoreSplitsSignaledReaders);
>
> // Register only one reader, run() will not execute
> enumerator.addReader(0);
>
> // Reset mock
> Mockito.reset(enumeratorContext);
>
> // Reader re-registration
> enumerator.addReader(0);
>
> // Verify no signal was re-sent
> Mockito.verify(enumeratorContext,
Mockito.never()).signalNoMoreSplits(Mockito.anyInt());
> }
> ```
>
> **Reason**: These test cases more realistically simulate failure recovery
scenarios and cover key boundary conditions (different behavior before and
after run() completes).
>
> ### Issue 3: SeaTunnel Engine layer `hasNoMoreSplitsSignaled` method
access modifier should be private
> **Location**:
`seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/context/SeaTunnelSplitEnumeratorContext.java:116-118`
>
> ```java
> public boolean hasNoMoreSplitsSignaled(int subtaskIndex) {
> return noMoreSplitsSignaledReaders.contains(subtaskIndex);
> }
> ```
>
> **Related Context**:
>
> * Caller: `SourceSplitEnumeratorTask.java:230` (only used internally)
> * Same pattern: Flink layer does not expose this method (directly accesses
the collection in `FlinkSourceEnumerator`)
>
> **Problem Description**:
>
> The `hasNoMoreSplitsSignaled` method is declared as `public`, but it is
only called in the `SourceSplitEnumeratorTask` class with no external callers.
>
> Setting it to `public` may lead to:
>
> 1. External code incorrectly relying on this internal implementation detail
> 2. Increased API surface area, increasing maintenance costs
> 3. If this logic needs to be modified in the future, it will affect all
callers
>
> **Potential Risks**:
>
> * **Risk 1**: External code may depend on this method, making future
refactoring difficult
> * **Risk 2**: Violates the principle of least privilege, increasing
unnecessary coupling
>
> **Impact Scope**:
>
> * **Direct Impact**: API design of `SeaTunnelSplitEnumeratorContext`
> * **Indirect Impact**: Potential external callers (if any)
> * **Affected Area**: SeaTunnel Engine
>
> **Severity**: Low (code works properly, but design is not rigorous enough)
>
> **Improvement Suggestions**:
>
> ```java
> // Change access modifier to package-private (default)
> boolean hasNoMoreSplitsSignaled(int subtaskIndex) {
> return noMoreSplitsSignaledReaders.contains(subtaskIndex);
> }
>
> // Or change to private and provide a package-visible accessor
> private boolean hasNoMoreSplitsSignaled(int subtaskIndex) {
> return noMoreSplitsSignaledReaders.contains(subtaskIndex);
> }
> ```
>
> Since `SourceSplitEnumeratorTask` and `SeaTunnelSplitEnumeratorContext`
are in the same package, package-private is sufficient.
>
> **Reason**: Reducing unnecessary API exposure reduces coupling and
improves code maintainability and evolvability.
>
> ### Issue 4: Missing CHANGELOG update
> **Location**: Project root directory `CHANGELOG.md` (or other changelog
files)
>
> **Problem Description**:
>
> This is a bug fix and should update the CHANGELOG to explain what problem
was fixed and the impact scope.
>
> **Potential Risks**:
>
> * **Risk 1**: Users are unaware of this fix and may still be affected by
this bug
> * **Risk 2**: Lack of reference information when upgrading, unable to
assess impact
>
> **Impact Scope**:
>
> * **Direct Impact**: Transparency of version releases
> * **Indirect Impact**: Users' upgrade decisions
> * **Affected Area**: All users
>
> **Severity**: Low (does not affect code functionality, but affects project
management)
>
> **Improvement Suggestions**:
>
> When merging, add an entry like this to the CHANGELOG:
>
> ```
> ## [Version X.Y.Z] - 2024-XX-XX
>
> ### Bug Fixes
> * **[Translation]** Fixed an issue where bounded source jobs could hang
after TaskManager failover
> because `NoMoreSplitsEvent` was not re-sent to recovered readers. The
framework now tracks
> which subtasks have received the terminal signal and re-sends it when
readers re-register.
> (Affected connectors: JDBC, Iceberg, Hive, File and all other bounded
sources)
> ```
>
> **Reason**: Keeping the CHANGELOG updated helps users understand version
changes, assess upgrade impacts, and quickly locate issues.
>
> ### Issue 5: Missing integration tests or E2E tests for critical paths
> **Location**: Test coverage
>
> **Related Context**:
>
> * Unit tests: `FlinkSourceEnumeratorTest.java`,
`SourceSplitEnumeratorTaskTest.java`
> * E2E tests: `seatunnel-e2e/` directory
>
> **Problem Description**:
>
> The PR only contains unit tests and lacks integration tests or E2E tests
to verify real failure recovery scenarios.
>
> While unit tests are useful, they:
>
> 1. Use Mockito mocking and cannot verify real Flink behavior
> 2. Test scenarios are too simplified to cover complex failure recovery
processes
> 3. Cannot verify interactions with other components (such as checkpoint,
state backend)
>
> **Potential Risks**:
>
> * **Risk 1**: Unit tests pass, but there may be issues in actual scenarios
> * **Risk 2**: Cannot capture Flink framework layer compatibility issues
> * **Risk 3**: Regression risk (future code modifications may break this
functionality)
>
> **Impact Scope**:
>
> * **Direct Impact**: Completeness of test coverage
> * **Indirect Impact**: Code quality and stability
> * **Affected Area**: All users using the Flink engine
>
> **Severity**: Medium (unit tests exist but are insufficient)
>
> **Improvement Suggestions**:
>
> It is recommended to add an integration test using a real Flink
MiniCluster to test failure recovery scenarios:
>
> ```java
> // Pseudo-code example
> @Test
> void testFailoverWithBoundedSource() throws Exception {
> // 1. Start a Flink MiniCluster
> // 2. Submit a job with a bounded data source (e.g., JDBC)
> // 3. Wait for the job to run for a while
> // 4. Simulate TaskManager failure (kill process or trigger exception)
> // 5. Wait for Flink to recover the job
> // 6. Verify the job can complete normally (instead of getting stuck)
>
> // Or use Flink's fault injection mechanism
> // 1. Create a Source that automatically fails
> // 2. Configure Flink for fault recovery
> // 3. Verify the job eventually completes
> }
> ```
>
> However, considering the complexity and execution time of integration
tests, you can also consider:
>
> 1. Add failure recovery scenarios to existing E2E tests (if any)
> 2. Perform manual testing before release
> 3. Document this fix and encourage users to verify in their test
environment
>
> **Reason**: Integration tests can provide higher confidence to ensure the
fix works effectively in real scenarios.
Thank you very much for your detailed and comprehensive code review! All the
issues you raised are very valuable. Here are my responses:
### Regarding Test Improvements (Issue 1 & Issue 5)
The test scenarios you suggested are indeed more comprehensive and rigorous.
Regarding the improvement suggestions:
- **Unit Test Optimization**: The test cases you proposed (simulating reader
re-registration and run() incomplete scenarios) are very valuable. I will
consider adding these boundary condition tests in future optimizations.
- **E2E/Integration Tests**: I understand that adding integration tests for
failure recovery scenarios would provide higher confidence. However, as you
mentioned, such tests are quite complex to implement and would significantly
increase E2E test execution time and maintenance costs. Considering the scope
of changes in the current PR and the cost-benefit ratio of testing, I'll
maintain the existing unit test coverage for now.
Since we have an upcoming release, I plan to add these additional test
improvements in a follow-up PR after the release.
### Regarding API Design (Issue 3)
Your suggestion about the access modifier of the `hasNoMoreSplitsSignaled`
method is very reasonable. Changing it to package-private does align with the
principle of least privilege and reduces unnecessary API exposure. I will
consider this optimization in a follow-up PR.
### Regarding CHANGELOG (Issue 4)
CHANGELOG updates are typically done during the official release process, at
which time the details and impact scope of this fix will be documented. Thank
you for the reminder!
---
Thank you again for investing your time in such a thorough code review.
These suggestions are very helpful for improving code quality. The current PR
focuses primarily on fixing the core issue, and I will gradually address some
of the optimization suggestions in future iterations.
--
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]
