[FLINK-24064][connector/common] HybridSource restore from savepoint #17111

tweise · 2021-09-02T03:31:25Z

What is the purpose of the change

Restore from savepoint fails due to deserialization of underlying splits before the underlying enumerator has been restored (details in JIRA). With this change deserialization will be deferred and be explicit in the HybridSource enumerator/reader.

Verifying this change

Existing tests don't cover restore from savepoint (ITCase performs recovery from initial state). Deserialization of HybridSplit and enumerator checkpoint covered by unit test. Changes verified with internal deployment. Planning to add unit test that just deserializes HybridSourceSplit before merging.

flinkbot · 2021-09-02T03:35:05Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit 8b30569 (Thu Sep 02 03:35:04 UTC 2021)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

tweise · 2021-09-02T03:38:20Z

@AHeise @stevenzwu please take a look at the deserialization change in general. I'm planning for some more cleanup work on this PR tomorrow but would also like for this to go into the 1.14 release.

flinkbot · 2021-09-02T03:52:38Z

CI report:

5690d85 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

stevenzwu · 2021-09-02T16:40:29Z

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

@@ -92,13 +90,13 @@

    private final List<SourceListEntry> sources;
    // sources are populated per subtask at switch time
-    private final Map<Integer, Source> switchedSources;
+    private final HybridSourceSplitSerializer.SwitchedSources switchedSources;


why is SwitchedSources nested inside the HybridSourceSplitSerializer?

did I miss sth? I didn't see switchedSources used anywhere?

Yes please remove. switchedSources acted like a shared cache which is now not necessary anymore. (Not sure how I missed that in the initial review, I guess I was too focused on API.)

This should now just be a field in enumerator/reader that caches the sources.

That came in after moving away from the fixed source sequence that originally both, enumerator and serializer had access to. They still needed access to the underlying serializer and therefore to the source that provided that serializer. Now that serializers are decoupled, this hacky thing is no longer needed. I just missed that in the refactor, thanks @stevenzwu for catching it.

AHeise

Thanks for providing this fix! I left some comments, see below.

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

AHeise · 2021-09-02T18:03:50Z

...ava/org/apache/flink/connector/base/source/hybrid/HybridSourceEnumeratorStateSerializer.java

-        this.switchedSources = switchedSources;
-        this.cachedSerializers = new HashMap<>();
-    }
+    public HybridSourceEnumeratorStateSerializer() {}


Much cleaner now!

AHeise · 2021-09-02T18:06:15Z

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

-                    return source.getSplitSerializer();
-                }));
+    /** Sources that participated in switching with cached serializers. */
+    public static class SwitchedSources implements Serializable {


I don't see why this is a nested class here.

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

AHeise · 2021-09-02T18:08:33Z

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

@@ -92,13 +90,13 @@

    private final List<SourceListEntry> sources;
    // sources are populated per subtask at switch time
-    private final Map<Integer, Source> switchedSources;
+    private final HybridSourceSplitSerializer.SwitchedSources switchedSources;


Yes please remove. switchedSources acted like a shared cache which is now not necessary anymore. (Not sure how I missed that in the initial review, I guess I was too focused on API.)

This should now just be a field in enumerator/reader that caches the sources.

stevenzwu · 2021-09-02T19:20:09Z

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

+        return wrappedStateBytes;
+    }
+
+    public int wrappedStateSerializerVersion() {


nit: wrappedStateSerializerVersion -> getWrappedStateSerializerVersion just to be consistent of Flink style

stevenzwu · 2021-09-02T19:26:29Z

...ava/org/apache/flink/connector/base/source/hybrid/HybridSourceEnumeratorStateSerializer.java

-            out.writeInt(enumStateBytes.length);
-            out.write(enumStateBytes);
+            out.writeInt(enumState.wrappedStateSerializerVersion());
+            out.writeInt(enumState.getWrappedState().length);


integer would limit the state size to 2 GB. not sure if we need to worry about it or not. It can happen if the historical storage (like HDFS or Iceberg) have many files/splits for the booststrap scan.

Each Iceberg split contains data files, delete files (for upsert), schema string. Each data file also contains stats for every column. if the table is wide (many columns), each split may go over 10 KB

This discussion is probably outside the scope of this PR

Please note that the limit would not be the integer to represent the size but rather the byte[] array that cannot go beyond that. We do not have 64 bit array https://www.nayuki.io/page/large-arrays-proposal-for-java

yes. understood it is the choice of bye[], which is then a limitation of SimpleVersionedSerializer's API

I also wonder if we would hit other issues with such large state serialized in the coordinator? Can IcebergSource limit the number of splits it keeps in the checkpoint and only add more once some have been processed?

@tweise I have thought about adding the optimization (of limiting the number of splits) for streaming read in the future. for bounded job, we can't. on the other hand, bounded job may not need to have checkpoint enabled

stevenzwu · 2021-09-02T20:51:19Z

@tweise do we need a MiniCluster unit test for the savepoint trigger and restore?

stevenzwu · 2021-09-03T03:51:01Z

...nector-base/src/main/java/org/apache/flink/connector/base/source/hybrid/SwitchedSources.java

+    }
+
+    public SimpleVersionedSerializer<SourceSplit> serializerOf(int sourceIndex) {
+        return cachedSerializers.computeIfAbsent(


why do we need to cache the SplitSerializer? seems unnecessary to me

To not create a new serializer instance per split, but rather once per coordinator/operator (matching how it works for the top level source).

I see. Originally I was imagining the singleton pattern from file source. Then this caching is not necessary.

@Override public SimpleVersionedSerializer<FileSourceSplit> getSplitSerializer() { return FileSourceSplitSerializer.INSTANCE; }

I guess it depends on the implementation. Some source impls may construct a new object in this method and hence this caching might be beneficial.

tweise · 2021-09-03T05:29:56Z

@tweise do we need a MiniCluster unit test for the savepoint trigger and restore?

I'm going to look into adding that to HybridSourceITCase. Probably outside of this PR because I want to backport this to release branches and not risk adding potential test instability.

SteNicholas · 2021-09-03T07:44:58Z

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

@@ -21,18 +21,25 @@
 /** The state of hybrid source enumerator. */
 public class HybridSourceEnumeratorState {
    private final int currentSourceIndex;
-    private final Object wrappedState;
+    private byte[] wrappedStateBytes;
+    private final int wrappedStateSerializerVersion;


Suggested change

private final int wrappedStateSerializerVersion;

private final int serializerVersion;

Considered that as well and prefer the verbose name to make clear that this is the serializer version for the underlying state vs that of HybridSourceEnumeratorState.

SteNicholas · 2021-09-03T07:46:18Z

...ctor-base/src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSourceSplit.java

 import java.util.List;
 import java.util.Objects;

 /** Source split that wraps the actual split type. */
 public class HybridSourceSplit implements SourceSplit {

-    private final SourceSplit wrappedSplit;
+    private final byte[] wrappedSplitBytes;
+    private final int wrappedSplitSerializerVersion;


Suggested change

private final int wrappedSplitSerializerVersion;

private final int serializerVersion;

Considered that as well and prefer the verbose name to make clear that this is the serializer version for the underlying state vs that of HybridSourceSplit.

SteNicholas · 2021-09-03T07:47:40Z

...ctor-base/src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSourceSplit.java

@@ -57,38 +69,64 @@ public boolean equals(Object o) {
            return false;
        }
        HybridSourceSplit that = (HybridSourceSplit) o;
-        return sourceIndex == that.sourceIndex && wrappedSplit.equals(that.wrappedSplit);
+        return sourceIndex == that.sourceIndex
+                && Arrays.equals(wrappedSplitBytes, that.wrappedSplitBytes);


Don't need the splitId equal?

Not needed because splitId is already part of wrappedSplitBytes.

SteNicholas · 2021-09-03T07:48:05Z

...ctor-base/src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSourceSplit.java

    }

    @Override
    public String toString() {
        return "HybridSourceSplit{"
                + "realSplit="
-                + wrappedSplit
+                + wrappedSplitBytes
                + ", sourceIndex="
                + sourceIndex


Add the splitId field.

Done. I also removed wrappedSplitBytes because it doesn't provide meaningful information.

stevenzwu

LGTM after @SteNicholas 's comments are addressed

tweise · 2021-09-03T23:24:25Z

@stevenzwu @AHeise @SteNicholas thanks for the review!

tweise requested a review from AHeise September 2, 2021 03:31

rmetzger added the review=description? label Sep 2, 2021

rmetzger added the component=Connectors/Common label Sep 2, 2021

stevenzwu reviewed Sep 2, 2021

View reviewed changes

AHeise reviewed Sep 2, 2021

View reviewed changes

stevenzwu reviewed Sep 2, 2021

View reviewed changes

tweise force-pushed the hybridsource-savepoint branch from fd39ae6 to f4b4b59 Compare September 3, 2021 00:08

stevenzwu reviewed Sep 3, 2021

View reviewed changes

SteNicholas reviewed Sep 3, 2021

View reviewed changes

stevenzwu approved these changes Sep 3, 2021

View reviewed changes

[FLINK-24064][connector/common] HybridSource restore from savepoint

5690d85

tweise force-pushed the hybridsource-savepoint branch from 74087ed to 5690d85 Compare September 3, 2021 18:53

tweise merged commit 2984d87 into apache:master Sep 3, 2021

tweise deleted the hybridsource-savepoint branch September 3, 2021 23:23

This was referenced Sep 3, 2021

[release-1.14][FLINK-24064][connector/common] HybridSource restore from savepoint #17143

Merged

[release-1.13][FLINK-24064][connector/common] HybridSource restore from savepoint #17144

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-24064][connector/common] HybridSource restore from savepoint #17111

[FLINK-24064][connector/common] HybridSource restore from savepoint #17111

tweise commented Sep 2, 2021

flinkbot commented Sep 2, 2021

tweise commented Sep 2, 2021

flinkbot commented Sep 2, 2021 •

edited

Loading

stevenzwu Sep 2, 2021

stevenzwu Sep 2, 2021

AHeise Sep 2, 2021

tweise Sep 2, 2021 •

edited

Loading

AHeise left a comment

AHeise Sep 2, 2021

AHeise Sep 2, 2021 •

edited

Loading

AHeise Sep 2, 2021

stevenzwu Sep 2, 2021

stevenzwu Sep 2, 2021

stevenzwu Sep 2, 2021

stevenzwu Sep 2, 2021

AHeise Sep 2, 2021

stevenzwu Sep 2, 2021

tweise Sep 3, 2021

stevenzwu Sep 3, 2021

stevenzwu commented Sep 2, 2021

stevenzwu Sep 3, 2021

tweise Sep 3, 2021

stevenzwu Sep 3, 2021

tweise commented Sep 3, 2021

SteNicholas Sep 3, 2021

tweise Sep 3, 2021

SteNicholas Sep 3, 2021

tweise Sep 3, 2021

SteNicholas Sep 3, 2021

tweise Sep 3, 2021

SteNicholas Sep 3, 2021

tweise Sep 3, 2021

stevenzwu left a comment

tweise commented Sep 3, 2021

	private final int wrappedStateSerializerVersion;
	private final int serializerVersion;

	private final int wrappedSplitSerializerVersion;
	private final int serializerVersion;

[FLINK-24064][connector/common] HybridSource restore from savepoint #17111

[FLINK-24064][connector/common] HybridSource restore from savepoint #17111

Conversation

tweise commented Sep 2, 2021

What is the purpose of the change

Verifying this change

flinkbot commented Sep 2, 2021

Automated Checks

Review Progress

tweise commented Sep 2, 2021

flinkbot commented Sep 2, 2021 • edited Loading

CI report:

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tweise Sep 2, 2021 • edited Loading

Choose a reason for hiding this comment

AHeise left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

AHeise Sep 2, 2021 • 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

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

stevenzwu commented Sep 2, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tweise commented Sep 3, 2021

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

stevenzwu left a comment

Choose a reason for hiding this comment

tweise commented Sep 3, 2021

flinkbot commented Sep 2, 2021 •

edited

Loading

tweise Sep 2, 2021 •

edited

Loading

AHeise Sep 2, 2021 •

edited

Loading