[FLINK-25445] No need to create local recovery dirs when disabled loc… #18306

zuston · 2022-01-09T05:30:24Z

…al-recovery in TaskExecutorLocalStateStoresManager

What is the purpose of the change

TaskExecutor will create file, check localRecoveryEnabled and load local state store for each task submission in method localStateStoreForSubtask. For batch jobs, the localRecoveryEnabled is always false, and there's no need to create these local files for task in TaskExecutor

Brief change log

When disable task local recovery, it will not create subtask state dirs.

Verifying this change

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, Kubernetes/Yarn, 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)

flinkbot · 2022-01-09T05:33:41Z

CI report:

f8c4b9c Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

flinkbot · 2022-01-09T05:35:41Z

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 7d972d4 (Sun Jan 09 05:35:41 UTC 2022)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!
This pull request references an unassigned Jira ticket. According to the code contribution guide, tickets need to be assigned before starting with the implementation work.

_{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

zuston · 2022-01-11T13:21:36Z

Could you help review? @dmvk Thanks

dmvk

Thanks for the PR @zuston! This already looks pretty good ;) I went through the usages of the LocalRecoveryDirectoryProvider and the change seems safe.

My only concern is around proper handling of @Nullable providers in LocalRecoveryConfig, PTAL

flink-runtime/src/main/java/org/apache/flink/runtime/state/LocalRecoveryConfig.java

zuston · 2022-01-11T14:02:45Z

Thanks for your review @dmvk

By the way, why we make task local-recovery disabled defaultly? Only due to the uncompatility with unaligned checkpoint?

dmvk · 2022-01-11T14:21:21Z

The behavior of the local recovery highly depends on the state backend. AFAIK heap state backend still doesn't support it. For RocksDB you can find some context in https://issues.apache.org/jira/browse/FLINK-15507

zuston · 2022-01-11T14:28:59Z

Thank very much @dmvk . I’ll take a look

zuston · 2022-01-12T11:45:17Z

Updated @dmvk Could you help review again? Thanks

zuston · 2022-01-14T13:09:41Z

@dmvk Gentle ping. If you have time, could you help review it? At weekends, i will have more time to deal with it.
Thanks

dmvk

Hi @zuston, I've added few more minor comments. I think this is mostly good to go.

My primary concern would be that we should get rid of the SavepointLocalRecoveryProvider from the state processor API, as it's no longer needed after this patch.

Thanks for working on this! 👍

dmvk · 2022-01-14T15:47:15Z

flink-runtime/src/main/java/org/apache/flink/runtime/state/LocalRecoveryConfig.java

+        return localStateDirectories == null
+                ? Optional.empty()
+                : Optional.of(localStateDirectories);


Suggested change

return localStateDirectories == null

? Optional.empty()

: Optional.of(localStateDirectories);

return Optional.ofNullable(localStateDirectories);

dmvk · 2022-01-14T16:08:58Z

flink-runtime/src/main/java/org/apache/flink/runtime/state/LocalRecoveryConfig.java

@@ -31,11 +35,11 @@
    private final boolean localRecoveryEnabled;

    /** Encapsulates the root directories and the subtask-specific path. */
-    @Nonnull private final LocalRecoveryDirectoryProvider localStateDirectories;
+    @Nullable private final LocalRecoveryDirectoryProvider localStateDirectories;

    public LocalRecoveryConfig(
            boolean localRecoveryEnabled,


I think we might want to get rid of this parameter completely. This two parameters having the same meaning thing, just leads into some weird implementations.

For example we can now easily get rid of the SavepointLocalRecoveryProvider from the state processor API.

dmvk · 2022-01-14T16:09:46Z

flink-runtime/src/main/java/org/apache/flink/runtime/state/LocalRecoveryConfig.java

@@ -44,9 +48,10 @@ public boolean isLocalRecoveryEnabled() {
        return localRecoveryEnabled;


Suggested change

return localRecoveryEnabled;

return localStateDirectories != null;

dmvk · 2022-01-14T16:11:51Z

flink-runtime/src/main/java/org/apache/flink/runtime/state/LocalRecoveryConfig.java

+
+    /**
+     * An {@link NullOfDirProviderException} is thrown when the {@link
+     * LocalRecoveryDirectoryProvider} is null.
+     */
+    @Internal
+    public static class NullOfDirProviderException extends RuntimeException {
+        public NullOfDirProviderException(String message, Throwable cause) {
+            super(message, cause);
+        }
+
+        public NullOfDirProviderException(String message) {
+            super(message);
+        }
+
+        public NullOfDirProviderException() {
+            this(
+                    "When task local recovery is enabled, the NullPointerException of LocalRecoveryDirectoryProvider should not happen.");
+        }
+    }


Suggested change

/**

* An {@link NullOfDirProviderException} is thrown when the {@link

* LocalRecoveryDirectoryProvider} is null.

*/

@Internal

public static class NullOfDirProviderException extends RuntimeException {

public NullOfDirProviderException(String message, Throwable cause) {

super(message, cause);

}

public NullOfDirProviderException(String message) {

super(message);

}

public NullOfDirProviderException() {

this(

"When task local recovery is enabled, the NullPointerException of LocalRecoveryDirectoryProvider should not happen.");

}

}

public static Supplier<IllegalStateException> localRecoveryNotEnabled() {

return () ->

new IllegalStateException(

"Getting a LocalRecoveryDirectoryProvider is only supported with the local recovery enabled. This is a bug and should be reported.");

}

zuston · 2022-01-17T10:49:49Z

Updated @dmvk. Could you help check again?
Thanks~

zuston · 2022-01-19T10:54:54Z

Gentle ping @dmvk . Thanks :)

…al-recovery in TaskExecutorLocalStateStoresManager

zuston · 2022-01-24T14:42:25Z

Rebase latest master and squash all commits. @dmvk If you have time, could you help review again? Thanks :)

zuston · 2022-01-25T10:27:23Z

@flinkbot run azure

zuston · 2022-02-01T13:22:34Z

@dmvk Gentle ping. Could u help review it ? Thanks

dmvk

LGTM 👍 Thanks for the contribution

rmetzger added the component=Runtime/Coordination label Jan 9, 2022

dmvk requested changes Jan 11, 2022

View reviewed changes

flink-runtime/src/main/java/org/apache/flink/runtime/state/LocalRecoveryConfig.java Outdated Show resolved Hide resolved

flink-runtime/src/main/java/org/apache/flink/runtime/state/LocalRecoveryConfig.java Outdated Show resolved Hide resolved

dmvk requested changes Jan 14, 2022

View reviewed changes

[FLINK-25445] No need to create local recovery dirs when disabled loc…

f8c4b9c

…al-recovery in TaskExecutorLocalStateStoresManager

zuston force-pushed the FLINK-25445 branch from 7c716c0 to f8c4b9c Compare January 24, 2022 14:40

dmvk approved these changes Feb 1, 2022

View reviewed changes

XComp merged commit 9f61e12 into apache:master Feb 1, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-25445] No need to create local recovery dirs when disabled loc… #18306

[FLINK-25445] No need to create local recovery dirs when disabled loc… #18306

zuston commented Jan 9, 2022 •

edited

flinkbot commented Jan 9, 2022 •

edited

flinkbot commented Jan 9, 2022

zuston commented Jan 11, 2022

dmvk left a comment

zuston commented Jan 11, 2022 •

edited

dmvk commented Jan 11, 2022

zuston commented Jan 11, 2022

zuston commented Jan 12, 2022

zuston commented Jan 14, 2022

dmvk left a comment

dmvk Jan 14, 2022

dmvk Jan 14, 2022

dmvk Jan 14, 2022

dmvk Jan 14, 2022

zuston Jan 17, 2022

zuston commented Jan 17, 2022

zuston commented Jan 19, 2022

zuston commented Jan 24, 2022

zuston commented Jan 25, 2022

zuston commented Feb 1, 2022

dmvk left a comment

		@@ -44,9 +48,10 @@ public boolean isLocalRecoveryEnabled() {
		return localRecoveryEnabled;

	return localRecoveryEnabled;
	return localStateDirectories != null;

[FLINK-25445] No need to create local recovery dirs when disabled loc… #18306

[FLINK-25445] No need to create local recovery dirs when disabled loc… #18306

Conversation

zuston commented Jan 9, 2022 • edited

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 Jan 9, 2022 • edited

CI report:

flinkbot commented Jan 9, 2022

Automated Checks

Review Progress

zuston commented Jan 11, 2022

dmvk left a comment

Choose a reason for hiding this comment

zuston commented Jan 11, 2022 • edited

dmvk commented Jan 11, 2022

zuston commented Jan 11, 2022

zuston commented Jan 12, 2022

zuston commented Jan 14, 2022

dmvk left a comment

Choose a reason for hiding this comment

dmvk Jan 14, 2022

Choose a reason for hiding this comment

dmvk Jan 14, 2022

Choose a reason for hiding this comment

dmvk Jan 14, 2022

Choose a reason for hiding this comment

dmvk Jan 14, 2022

Choose a reason for hiding this comment

zuston Jan 17, 2022

Choose a reason for hiding this comment

zuston commented Jan 17, 2022

zuston commented Jan 19, 2022

zuston commented Jan 24, 2022

zuston commented Jan 25, 2022

zuston commented Feb 1, 2022

dmvk left a comment

Choose a reason for hiding this comment

zuston commented Jan 9, 2022 •

edited

flinkbot commented Jan 9, 2022 •

edited

zuston commented Jan 11, 2022 •

edited