[FLINK-12729][state-processor-api] Add state reader for consuming non-partitioned operator state #8615

sjwiesman · 2019-06-04T13:36:35Z

What is the purpose of the change

This is the initial PR for FLIP-43 adding the functionality to read non-partitioned operator state from a state snapshot.

Brief changelog

The first two commits expose internal api's for use outside of their package but mark them for internal use only.
The third commit adds the operator state reader functionality

Verifying this change

This change added tests and can be verified as follows:

Unit and IT tests

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

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

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
(Full documentation coming in the keyed state reader PR)

flinkbot · 2019-06-04T13:40:15Z

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.

Review Progress

✅ 1. The [description] looks good.
- Approved by @tzulitai [PMC]
✅ 2. There is [consensus] that the contribution should go into to Flink.
- Approved by @tzulitai [PMC]
❗ 3. Needs [attention] from.
- Needs attention by @tzulitai [PMC]
❓ 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

sjwiesman · 2019-06-04T13:42:06Z

@flinkbot attention @tzulitai

tzulitai · 2019-06-05T02:10:34Z

@flinkbot approve description
@flinkbot approve consensus

tzulitai

Thanks for the great work @sjwiesman!
The comments I have are mostly minor, with only 2 more important comments about classloaders and whether or not the work of reading savepoint metadata can be done earlier and done only once.

One thing to probably think about (maybe can be done as a follow-up PR):
Is the naming of Savepoint and ExistingSavepoint still sensible here, now that we renamed this series of work as "State Processor API"?

Otherwise, I have tried this manually, and it works for both types of backends, haven't discovered any problems so far. +1 to merge this once the comments are addressed.

...libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/Savepoint.java

tzulitai · 2019-06-24T02:49:14Z

...s/flink-state-processing-api/src/main/java/org/apache/flink/state/api/ExistingSavepoint.java

+	ExistingSavepoint(ExecutionEnvironment env, String path, StateBackend stateBackend) {
+		this.env = env;
+		this.existingSavepoint = path;
+		this.stateBackend = stateBackend;


nit: missing Preconditions.checkNotNull checks

tzulitai · 2019-06-24T02:51:50Z

...s/flink-state-processing-api/src/main/java/org/apache/flink/state/api/ExistingSavepoint.java

+		String uid,
+		String name,
+		TypeInformation<T> typeInfo,
+		TypeSerializer<T> serializer) {


As a follow-up, we can probably think about a variant where the user simply passes in a TypeSerializer and no TypeInformation. In this case, can we just wrap the given serializer into a "dummy" type info?
Not entirely sure what methods of the TypeInformation will be used in the batch processing API.

tzulitai · 2019-06-24T02:52:39Z

...processing-api/src/main/java/org/apache/flink/state/api/input/BroadcastStateInputFormat.java

+	 */
+	public BroadcastStateInputFormat(String savepointPath, String uid, MapStateDescriptor<K, V> descriptor) {
+		super(savepointPath, uid, true);
+		this.descriptor = descriptor;


nit: missing Preconditions.checkNotNull() check.

Please check all other constructors of this problem, I think the issue occurs in multiple constructors.

tzulitai · 2019-06-24T02:58:00Z

...-processing-api/src/main/java/org/apache/flink/state/api/input/OperatorStateInputFormat.java

+ *
+ * @param <OT> The type of the input.
+ */
+abstract class OperatorStateInputFormat<OT> extends SavepointInputFormat<OT, OperatorStateInputSplit> {


@Internal annotation

...-processing-api/src/main/java/org/apache/flink/state/api/input/OperatorStateInputFormat.java

tzulitai · 2019-06-24T03:06:01Z

...tate-processing-api/src/main/java/org/apache/flink/state/api/input/SavepointInputFormat.java

+	 * @throws IOException If the savepoint path is invalid or the uid does not exist
+	 */
+	OperatorState getOperatorState() throws IOException {
+		final Savepoint savepoint = SavepointLoader.loadSavepoint(savepointPath);


I've been thinking whether or not this operation can be done more earlier when loading a ExistingSavepoint, and maintained by the ExistingSavepoint class. Not entirely sure though, as that would require access to the DFS on the client side; not sure how feasible that is in practice.

Not too much of a deal, as this shouldn't be a heavy workload.

tzulitai · 2019-06-24T03:09:33Z

...k-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointLoader.java

+			.resolveCheckpointPointer(savepointPath);
+
+		try (DataInputStream stream = new DataInputStream(location.getMetadataHandle().openInputStream())) {
+			return Checkpoints.loadCheckpointMetadata(stream, NullClassLoader.INSTANCE);


In the future, once we add type information of state into the savepoint metadata file, I don't think this NullClassLoader is correct anymore. We would potentially need the actual user classloader, since the metadata would contain user classes (e.g. user implemented TypeSerializerSnapshots)

There was one place where I couldn't get access to the user classloader but it wasn't being used so I decided to go this route. However, if we do the reading on the client the current thread class loader should be the user class loader so that makes it easier.

...k-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/NullClassLoader.java

…tes as public

…as public

…-partitioned operator state

…ing non-partitioned operator state

sjwiesman · 2019-06-24T15:22:19Z

Thanks for the review @tzulitai !

I've addressed all the comments including moving the savepoint loader to run once on the client.

…ing non-partitioned operator state

…n OnDiskSavepointMetadata This closes apache#8615.

sjwiesman changed the title ~~Flink 12729~~ [FLINK-12729][state-processor-api] Add state reader for consuming non-partitioned operator state Jun 4, 2019

rmetzger added the review=description? label Jun 4, 2019

sjwiesman force-pushed the FLINK-12729 branch from 275063c to bc852dd Compare June 4, 2019 13:49

rmetzger added component=API/DataStream component=Runtime/StateBackends labels Jun 4, 2019

sjwiesman force-pushed the FLINK-12729 branch from bc852dd to 755348a Compare June 4, 2019 16:40

sjwiesman mentioned this pull request Jun 4, 2019

[FLINK-12732][state-processor-api] Add savepoint reader for consuming partitioned operator state #8618

Closed

tzulitai self-requested a review June 5, 2019 02:10

rmetzger added review=architecture? and removed review=description? labels Jun 5, 2019

tzulitai reviewed Jun 24, 2019

View reviewed changes

sjwiesman added 3 commits June 24, 2019 09:26

[hotfix] Expose StateAssignmentOperation#reDistributePartitionableSta…

52ec4bb

…tes as public

[hotfix] Expose AbstractFsCheckpointStorage#resolveCheckpointPointer …

ac0329a

…as public

[FLINK-12729][state-processor-api] Add state reader for consuming non…

d17d4cf

…-partitioned operator state

sjwiesman force-pushed the FLINK-12729 branch from 755348a to eb57cf9 Compare June 24, 2019 14:49

fixup! [FLINK-12729][state-processor-api] Add state reader for consum…

11e262a

…ing non-partitioned operator state

sjwiesman force-pushed the FLINK-12729 branch from eb57cf9 to 11e262a Compare June 24, 2019 14:52

fixup! [FLINK-12729][state-processor-api] Add state reader for consum…

585907f

…ing non-partitioned operator state

sjwiesman force-pushed the FLINK-12729 branch from db509bc to 585907f Compare June 24, 2019 14:59

fixup! [FLINK-12729][state-processor-api] Add state reader for consum…

dceecbb

…ing non-partitioned operator state

sjwiesman force-pushed the FLINK-12729 branch from 7aedfe4 to dceecbb Compare June 25, 2019 02:47

tzulitai added a commit to tzulitai/flink that referenced this pull request Jun 25, 2019

[FLINK-12729] [state-processing-api] Build operator state index withi…

9c63353

…n OnDiskSavepointMetadata This closes apache#8615.

asfgit closed this in 88d2e3c Jun 26, 2019

zentol pushed a commit to zentol/flink that referenced this pull request Jun 26, 2019

[FLINK-12729] [state-processing-api] Build operator state index withi…

a3c5ab3

…n OnDiskSavepointMetadata This closes apache#8615.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-12729][state-processor-api] Add state reader for consuming non-partitioned operator state #8615

[FLINK-12729][state-processor-api] Add state reader for consuming non-partitioned operator state #8615

sjwiesman commented Jun 4, 2019

flinkbot commented Jun 4, 2019 •

edited

Loading

sjwiesman commented Jun 4, 2019

tzulitai commented Jun 5, 2019

tzulitai left a comment

tzulitai Jun 24, 2019

tzulitai Jun 24, 2019

tzulitai Jun 24, 2019

tzulitai Jun 24, 2019

tzulitai Jun 24, 2019

tzulitai Jun 24, 2019

tzulitai Jun 24, 2019

sjwiesman Jun 24, 2019

sjwiesman commented Jun 24, 2019

[FLINK-12729][state-processor-api] Add state reader for consuming non-partitioned operator state #8615

[FLINK-12729][state-processor-api] Add state reader for consuming non-partitioned operator state #8615

Conversation

sjwiesman commented Jun 4, 2019

What is the purpose of the change

Brief changelog

Verifying this change

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

Documentation

flinkbot commented Jun 4, 2019 • edited Loading

Review Progress

sjwiesman commented Jun 4, 2019

tzulitai commented Jun 5, 2019

tzulitai left a 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

sjwiesman commented Jun 24, 2019

flinkbot commented Jun 4, 2019 •

edited

Loading