[FLINK-30334][runtime] Fix noMoreSplits event handling for HybridSource #21464

chucheng92 · 2022-12-08T07:35:36Z

What is the purpose of the change

SourceCoordinator#handleRequestSplitEvent hasNoMoreSplits check not consider the HybridSource situation. It will cause HybridSource do not read next child sources data and finally lead to data loss and unexpected runtime behavior.

Brief change log

Add an intercept strategy for hasNoMoreSplits checking correctly of HybridSourceSplitEnumerator.

Verifying this change

Add 2 new cases to cover the code path about the issue.

HybridSourceSplitEnumeratorTest#testInterceptNoMoreSplitEvent
SourceCoordinatorContextTest#testSignalNoMoreSplits

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): no
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? no
If yes, how is the feature documented? not applicable

flinkbot · 2022-12-08T07:42:06Z

CI report:

4ae65d4 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

chucheng92 · 2022-12-21T09:03:20Z

@zhuzhurk Hi, Can u help me to review it? thanks a lot

zhuzhurk

Thanks for opening this PR! @chucheng92
My major concern is that we should avoid changing the public interface SplitEnumeratorContext which is not necessary. Other changes looks good to me.

zhuzhurk · 2022-12-22T03:48:25Z

flink-core/src/main/java/org/apache/flink/api/connector/source/SplitEnumeratorContext.java

+     * @param subtask The index of the operator's parallel subtask that shall be signaled it will
+     *     receive splits later.
+     */
+    default void signalIntermediateNoMoreSplits(int subtask) {}


Let's avoid adding this method to SplitEnumeratorContext which is a @Public interface.
Intermediate noMoreSplits event is strictly related to HybridSource and is hard for users to understand or handle.

I think we can just directly add the method to SourceCoordinatorContext, then use instanceOf and type cast in HybridSourceSplitEnumerator to invoke this method.

@zhuzhurk Yes. SplitEnumeratorContext is a Public api. i add it with default method to avoid impact. We can not use realContext instanceOf SourceCoordinatorContext directly, HybridSourceSplitEnumerator is in flink-connector-base, SourceCoordinatorContext is in flink-runtime, how can we reference it? using reflection or add runtime dependency both are ugly.

I think we can add an interface in flink-core, e.g.

@Internal interface SupportsIntermediateNoMoreSplits { void signalIntermediateNoMoreSplits(int subtask); }

SourceCoordinatorContext should implement it. And HybridSourceSplitEnumerator should check that the realContext is an instance of SupportsIntermediateNoMoreSplits.

zhuzhurk · 2022-12-25T13:53:58Z

@chucheng92 F.I.Y. the community is preparing to release 1.16.1. I hope we can get this issue fixed in this release.
Would you try if the suggested way can work?

chucheng92 · 2022-12-26T04:24:00Z

@chucheng92 F.I.Y. the community is preparing to release 1.16.1. I hope we can get this issue fixed in this release. Would you try if the suggested way can work?

Yes, done with your suggested way. It works well. I have updated the pr. PTAL. thanks.

zhuzhurk · 2022-12-26T05:38:54Z

...re/src/test/java/org/apache/flink/api/connector/source/mocks/MockSplitEnumeratorContext.java

+
+    @Override
+    public void signalIntermediateNoMoreSplits(int subtask) {
+        subtaskHasNoMoreSplits[subtask] = false;


This method should be empty. The values should be false initially but not set to be false in this method.

zhuzhurk · 2022-12-26T05:42:36Z

...src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSourceSplitEnumerator.java

@@ -157,7 +158,17 @@ private void sendSwitchSourceEvent(int subtaskId, int sourceIndex) {
                LOG.debug("Restoring splits to subtask={} {}", subtaskId, splits);
                context.assignSplits(
                        new SplitsAssignment<>(Collections.singletonMap(subtaskId, splits)));
-                context.signalNoMoreSplits(subtaskId);
+                if (context instanceof SupportsIntermediateNoMoreSplits) {


It's better to have a method to be reused, e.g.

private static void signalNoMoreSplits( SplitEnumeratorContext<HybridSourceSplit> context, subtaskId, int sourceIndex, int sourceSize);

zhuzhurk · 2022-12-26T05:44:06Z

...time/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContext.java

+        // It's an intermediate noMoreSplit event, notify subtask to deal with this event.
+        callInCoordinatorThread(
+                () -> {
+                    subtaskHasNoMoreSplits[subtask] = false;


Why is it needed to be set to false?

In my understanding, it should be false initially and set to true when on signalNoMoreSplits(int). When failover happens, the value will be reset to false. Seems there is no need for signalIntermediateNoMoreSplits(int) to do the reset.

yes, no need to set false. i delete it. thanks.

chucheng92 · 2022-12-26T06:59:34Z

@zhuzhurk thanks for reviewing, i have fixed the problem you commented. PTAL.

zhuzhurk

Thanks for addressing all the comments! @chucheng92
LGTM.

chucheng92 · 2022-12-26T09:20:18Z

@flinkbot run azure

zhuzhurk · 2022-12-26T09:30:05Z

...test/java/org/apache/flink/connector/base/source/hybrid/HybridSourceSplitEnumeratorTest.java

+        // test add splits back, then SUBTASK0 restore splitFromSource0 split
+        // reset splits assignment & previous subtaskHasNoMoreSplits flag.
+        context.getSplitsAssignmentSequence().clear();
+        Whitebox.setInternalState(context, "subtaskHasNoMoreSplits", new boolean[] {false, false});


It's better to add a method resetNoMoreSplits(int) in MockSplitEnumeratorContext.
Instead of using reflection to do this.

@zhuzhurk thanks for your suggestion.

…hasNoMoreSplits check not consider the hybridsource situation

chucheng92 · 2022-12-26T10:30:07Z

@flinkbot run azure

zhuzhurk · 2022-12-27T01:23:58Z

Merging.

This closes #21464.

This closes apache#21464.

chucheng92 changed the title ~~[FLINK-30334] Reset SourceCoordinator#handleRequestSplitEvent without…~~ [FLINK-30334] Reset SourceCoordinator#handleRequestSplitEvent without violent hasNoMoreSplits check Dec 8, 2022

chucheng92 changed the title ~~[FLINK-30334] Reset SourceCoordinator#handleRequestSplitEvent without violent hasNoMoreSplits check~~ [FLINK-30334] Reset SourceCoordinator#handleRequestSplitEvent because wrong hasNoMoreSplits check Dec 8, 2022

chucheng92 changed the title ~~[FLINK-30334] Reset SourceCoordinator#handleRequestSplitEvent because wrong hasNoMoreSplits check~~ [FLINK-30334] Reset SourceCoordinator#handleRequestSplitEvent check logic because wrong hasNoMoreSplits check Dec 8, 2022

chucheng92 force-pushed the FLINK-30334 branch from 93544b9 to 964073c Compare December 8, 2022 08:02

flinkbot added the component=Runtime/Coordination label Dec 8, 2022

chucheng92 force-pushed the FLINK-30334 branch 3 times, most recently from d463f68 to 16f81a4 Compare December 21, 2022 09:02

zhuzhurk self-assigned this Dec 22, 2022

zhuzhurk reviewed Dec 23, 2022

View reviewed changes

chucheng92 force-pushed the FLINK-30334 branch 2 times, most recently from 6cdf6d6 to ae9e955 Compare December 26, 2022 04:18

chucheng92 force-pushed the FLINK-30334 branch 2 times, most recently from 3d0b9bd to 2f4ba4a Compare December 26, 2022 05:41

zhuzhurk reviewed Dec 26, 2022

View reviewed changes

chucheng92 force-pushed the FLINK-30334 branch 2 times, most recently from 39fda04 to de3b039 Compare December 26, 2022 06:57

zhuzhurk approved these changes Dec 26, 2022

View reviewed changes

chucheng92 force-pushed the FLINK-30334 branch from de3b039 to 2e80782 Compare December 26, 2022 09:19

zhuzhurk reviewed Dec 26, 2022

View reviewed changes

[FLINK-30334][runtime] Fix SourceCoordinator#handleRequestSplitEvent …

4ae65d4

…hasNoMoreSplits check not consider the hybridsource situation

chucheng92 force-pushed the FLINK-30334 branch from 2e80782 to 4ae65d4 Compare December 26, 2022 10:26

zhuzhurk changed the title ~~[FLINK-30334][runtime] Fix SourceCoordinator#handleRequestSplitEvent hasNoMoreSplits check not consider the hybridsource situation~~ [FLINK-30334][runtime] Fix noMoreSplits event handling for HybridSource Dec 27, 2022

zhuzhurk closed this in 6e4e6c6 Dec 27, 2022

zhuzhurk pushed a commit that referenced this pull request Dec 27, 2022

[FLINK-30334][runtime] Fix noMoreSplits event handling for HybridSource

2b5e4ac

This closes #21464.

chucheng92 added a commit to chucheng92/flink that referenced this pull request Feb 3, 2023

[FLINK-30334][runtime] Fix noMoreSplits event handling for HybridSource

9397212

This closes apache#21464.

sergeitsar pushed a commit to fentik/flink that referenced this pull request Feb 8, 2023

[FLINK-30334][runtime] Fix noMoreSplits event handling for HybridSource

bb60933

This closes apache#21464.

sergeitsar pushed a commit to fentik/flink that referenced this pull request Feb 8, 2023

[FLINK-30334][runtime] Fix noMoreSplits event handling for HybridSource

44b39e6

This closes apache#21464.

akkinenivijay pushed a commit to krisnaru/flink that referenced this pull request Feb 11, 2023

[FLINK-30334][runtime] Fix noMoreSplits event handling for HybridSource

a733387

This closes apache#21464.

chucheng92 deleted the FLINK-30334 branch June 14, 2023 03:56

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-30334][runtime] Fix noMoreSplits event handling for HybridSource #21464

[FLINK-30334][runtime] Fix noMoreSplits event handling for HybridSource #21464

chucheng92 commented Dec 8, 2022 •

edited

Loading

flinkbot commented Dec 8, 2022 •

edited

Loading

chucheng92 commented Dec 21, 2022

zhuzhurk left a comment

zhuzhurk Dec 22, 2022

chucheng92 Dec 23, 2022 •

edited

Loading

zhuzhurk Dec 23, 2022 •

edited

Loading

zhuzhurk commented Dec 25, 2022

chucheng92 commented Dec 26, 2022 •

edited

Loading

zhuzhurk Dec 26, 2022

zhuzhurk Dec 26, 2022

zhuzhurk Dec 26, 2022

zhuzhurk Dec 26, 2022

chucheng92 Dec 26, 2022

chucheng92 commented Dec 26, 2022

zhuzhurk left a comment

chucheng92 commented Dec 26, 2022

zhuzhurk Dec 26, 2022

chucheng92 Dec 26, 2022

chucheng92 commented Dec 26, 2022

zhuzhurk commented Dec 27, 2022

[FLINK-30334][runtime] Fix noMoreSplits event handling for HybridSource #21464

[FLINK-30334][runtime] Fix noMoreSplits event handling for HybridSource #21464

Conversation

chucheng92 commented Dec 8, 2022 • edited Loading

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented Dec 8, 2022 • edited Loading

CI report:

chucheng92 commented Dec 21, 2022

zhuzhurk left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chucheng92 Dec 23, 2022 • edited Loading

Choose a reason for hiding this comment

zhuzhurk Dec 23, 2022 • edited Loading

Choose a reason for hiding this comment

zhuzhurk commented Dec 25, 2022

chucheng92 commented Dec 26, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chucheng92 commented Dec 26, 2022

zhuzhurk left a comment

Choose a reason for hiding this comment

chucheng92 commented Dec 26, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chucheng92 commented Dec 26, 2022

zhuzhurk commented Dec 27, 2022

chucheng92 commented Dec 8, 2022 •

edited

Loading

flinkbot commented Dec 8, 2022 •

edited

Loading

chucheng92 Dec 23, 2022 •

edited

Loading

zhuzhurk Dec 23, 2022 •

edited

Loading

chucheng92 commented Dec 26, 2022 •

edited

Loading