[FLINK-25329][runtime] Support memory execution graph store in session cluster #18360

FangYongs · 2022-01-14T05:15:32Z

What is the purpose of the change

This PR aims to support memory execution graph store in session cluster

Brief change log

- Improve MemoryExecutionGraphInfoStore to support expiration time and maximum capacity
- Support memory and file execution graph store in SessionClusterEntrypoint

Verifying this change

This change added tests and can be verified as follows:

- Add MemoryExecutionGraphInfoStoreTest that test the methods in MemoryExecutionGraphInfoStore

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

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

Documentation

Does this pull request introduce a new feature? (yes / no) no
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

flinkbot · 2022-01-14T05:19:58Z

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 ff1d101 (Fri Jan 14 05:19:58 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

flinkbot · 2022-01-14T10:08:30Z

CI report:

c6b8f50 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

FangYongs · 2022-01-16T06:45:55Z

@flinkbot run azure

FangYongs · 2022-01-17T02:30:03Z

@rmetzger Can you please help to review this PR when you're free, thanks

KarmaGYZ · 2022-01-19T06:10:47Z

Thanks for the PR, @zjureel . I think it is a valid improvement. Regarding the implementation, I think we can simply introduce a config option "jobstore.flush-to-disk" to FileExecutionGraphInfoStore instead of introducing a new one, for deduplication.

FangYongs · 2022-01-24T07:53:08Z

Thanks @KarmaGYZ I have updated the codes and add a config option jobstore.flush-to-disk in FileExecutionGraphInfoStore instead of a memory store

KarmaGYZ

Thanks for the PR @zjureel . It LGTM except for some minor comments. Please address them and squash your commits.

...k-runtime/src/main/java/org/apache/flink/runtime/dispatcher/FileExecutionGraphInfoStore.java

...ntime/src/test/java/org/apache/flink/runtime/dispatcher/FileExecutionGraphInfoStoreTest.java

FangYongs · 2022-01-25T07:40:25Z

Thanks @KarmaGYZ , I have updated the doc and comments in FileExecutionGraphInfoStore and FileExecutionGraphInfoStoreTest

KarmaGYZ · 2022-01-25T08:55:02Z

@zjureel I think we need also ignore the maximumCacheSizeBytes in this case. However, there might be a lot of "if...else" in FileExecutionGraphInfoStore. Maybe we should return to a separate MemoryExecutionGraphInfoStore. Sorry for the inconvenience.

FangYongs · 2022-01-30T09:30:35Z

Thanks @KarmaGYZ , it doesn't matter and I have used MemoryExecutionGraphInfoStore, you can review the codes again when you're free, thanks :)

KarmaGYZ · 2022-02-07T07:44:05Z

flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java

+                                                    "'Memory': the memory job store keeps the archived execution graphs in memory and "
+                                                            + " it may cause FullGC or OOM when there are too many graphs"))


Suggested change

"'Memory': the memory job store keeps the archived execution graphs in memory and "

+ " it may cause FullGC or OOM when there are too many graphs"))

"'Memory': the memory job store keeps the archived execution graphs in memory. You "

+ "may need to limit the %s to mitigate FullGC or OOM when there are too many graphs",

code(JOB_STORE_MAX_CAPACITY.key())))

KarmaGYZ · 2022-02-07T07:52:45Z

...runtime/src/main/java/org/apache/flink/runtime/dispatcher/MemoryExecutionGraphInfoStore.java

 */
 public class MemoryExecutionGraphInfoStore implements ExecutionGraphInfoStore {

-    private final Map<JobID, ExecutionGraphInfo> serializableExecutionGraphInfos = new HashMap<>(4);
+    private final Map<JobID, ExecutionGraphInfo> serializableExecutionGraphInfos =
+            new ConcurrentHashMap<>(4);


Suggested change

new ConcurrentHashMap<>(4);

new ConcurrentHashMap<>();

KarmaGYZ · 2022-02-07T07:54:32Z

...runtime/src/main/java/org/apache/flink/runtime/dispatcher/MemoryExecutionGraphInfoStore.java

+    /** Capacity of the memory store, 0 means unlimited. */
+    private final int maximumCapacity;
+
+    private final ScheduledExecutor scheduledExecutor;


Suggested change

private final ScheduledExecutor scheduledExecutor;

@Nullable private final ScheduledExecutor scheduledExecutor;

KarmaGYZ · 2022-02-07T07:56:54Z

...runtime/src/main/java/org/apache/flink/runtime/dispatcher/MemoryExecutionGraphInfoStore.java

+    private final int maximumCapacity;
+
+    private final ScheduledExecutor scheduledExecutor;
+    private ScheduledFuture<?> cleanupFuture;


Suggested change

private ScheduledFuture<?> cleanupFuture;

@Nullable private ScheduledFuture<?> cleanupFuture;

KarmaGYZ · 2022-02-07T08:17:20Z

...runtime/src/main/java/org/apache/flink/runtime/dispatcher/MemoryExecutionGraphInfoStore.java

+            new ConcurrentHashMap<>(4);
+
+    /** Job id with creation timestamp. */
+    private final Queue<JobGraphTimestamp> jobGraphQueue = new ConcurrentLinkedQueue<>();


I think we can leverage the guava cache to handle the cleanup, expiration and concurrent access. Please refer to how we use it in FileExecutionGraphInfoStore. We can maintain a Cache<JobID, ExecutionGraphInfo> serializableExecutionGraphInfos.

I address it myself in https://github.com/KarmaGYZ/flink/tree/FLINK_25329_support_memory_store_in_session . You can take a look and fix the test issue.

KarmaGYZ · 2022-02-07T09:07:11Z

...k-runtime/src/test/java/org/apache/flink/runtime/dispatcher/ExecutionGraphInfoStoreTest.java

+ * Base test case of execution graph store for {@link FileExecutionGraphInfoStore} and {@link
+ * MemoryExecutionGraphInfoStore}.
+ */
+public abstract class ExecutionGraphInfoStoreTest extends TestLogger {


Maybe we can make it a utility class.

FangYongs · 2022-02-08T06:50:12Z

Thanks @KarmaGYZ , use cache in MemoryExecutionGraphInfoStore instead of queue and hashmap sounds good to me. I have update the code in MemoryExecutionGraphInfoStore and the test case, thanks

KarmaGYZ

Thanks for the update. The PR is now LGTM. Please address the left minor comments and squash the commits. I'll merge it once CI passed.

KarmaGYZ · 2022-02-08T07:10:25Z

...runtime/src/main/java/org/apache/flink/runtime/dispatcher/MemoryExecutionGraphInfoStore.java


    @Override
    public int size() {
-        return serializableExecutionGraphInfos.size();
+        return (int) serializableExecutionGraphInfos.size();


Suggested change

return (int) serializableExecutionGraphInfos.size();

return Math.toIntExact(serializableExecutionGraphInfos.size());

KarmaGYZ · 2022-02-08T07:16:45Z

...time/src/test/java/org/apache/flink/runtime/dispatcher/ExecutionGraphInfoStoreTestUtils.java

+ * Test utils class for {@link FileExecutionGraphInfoStore} and {@link
+ * MemoryExecutionGraphInfoStore}.
+ */
+public class ExecutionGraphInfoStoreTestUtils {


I like the idea of extracting the shared utilities for memory and file store tests. I think this change should belong to a separate hotfix commit.

KarmaGYZ · 2022-02-08T07:27:33Z

...ime/src/test/java/org/apache/flink/runtime/dispatcher/MemoryExecutionGraphInfoStoreTest.java

+                    executionGraphInfoStore.get(executionGraphInfo.getJobId()),
+                    Matchers.nullValue());
+
+            // check that the persisted file has been deleted


Suggested change

// check that the persisted file has been deleted

// check that the store is empty

KarmaGYZ · 2022-02-08T07:29:36Z

...ime/src/test/java/org/apache/flink/runtime/dispatcher/MemoryExecutionGraphInfoStoreTest.java

+
+    /** Tests that all persisted files are cleaned up after closing the store. */
+    @Test
+    public void testCloseCleansUp() throws IOException {


Better to use try clause here in case there is IOException from Store#put.

...time/src/test/java/org/apache/flink/runtime/dispatcher/ExecutionGraphInfoStoreTestUtils.java

…onGraphInfoStoreTestUtils

…mory graph store in session cluster This closes apache#18360.

FangYongs force-pushed the FLINK_25329_support_memory_store_in_session branch from ff1d101 to b5d9b58 Compare January 14, 2022 05:22

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

FangYongs force-pushed the FLINK_25329_support_memory_store_in_session branch 2 times, most recently from 93ceba0 to b10b4c3 Compare January 24, 2022 07:50

KarmaGYZ reviewed Jan 24, 2022

View reviewed changes

FangYongs force-pushed the FLINK_25329_support_memory_store_in_session branch from 65220a8 to 56fee29 Compare January 25, 2022 07:41

FangYongs force-pushed the FLINK_25329_support_memory_store_in_session branch 3 times, most recently from 6915706 to 5d6e58a Compare January 30, 2022 05:01

KarmaGYZ requested changes Feb 7, 2022

View reviewed changes

KarmaGYZ reviewed Feb 7, 2022

View reviewed changes

KarmaGYZ approved these changes Feb 8, 2022

View reviewed changes

FangYongs force-pushed the FLINK_25329_support_memory_store_in_session branch from 72130f1 to 8c7f47b Compare February 8, 2022 08:52

KarmaGYZ reviewed Feb 8, 2022

View reviewed changes

...time/src/test/java/org/apache/flink/runtime/dispatcher/ExecutionGraphInfoStoreTestUtils.java Show resolved Hide resolved

FangYongs force-pushed the FLINK_25329_support_memory_store_in_session branch from 8c7f47b to 8e4a673 Compare February 8, 2022 12:08

FangYongs added 2 commits February 9, 2022 10:10

[hotfix][runtime] Extract common inner classes and methods to Executi…

ca7f8ed

…onGraphInfoStoreTestUtils

[FLINK-25329][runtime] Use cache in memory graph store and support me…

c6b8f50

…mory graph store in session cluster This closes apache#18360.

KarmaGYZ force-pushed the FLINK_25329_support_memory_store_in_session branch from 8e4a673 to c6b8f50 Compare February 9, 2022 02:16

KarmaGYZ closed this in 7cb00d1 Feb 9, 2022

MrWhiteSike pushed a commit to MrWhiteSike/flink that referenced this pull request Mar 3, 2022

[FLINK-25329][runtime] Use cache in memory graph store and support me…

ffddf69

…mory graph store in session cluster This closes apache#18360.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-25329][runtime] Support memory execution graph store in session cluster #18360

[FLINK-25329][runtime] Support memory execution graph store in session cluster #18360

FangYongs commented Jan 14, 2022

flinkbot commented Jan 14, 2022

flinkbot commented Jan 14, 2022 •

edited

FangYongs commented Jan 16, 2022

FangYongs commented Jan 17, 2022

KarmaGYZ commented Jan 19, 2022

FangYongs commented Jan 24, 2022

KarmaGYZ left a comment

FangYongs commented Jan 25, 2022

KarmaGYZ commented Jan 25, 2022

FangYongs commented Jan 30, 2022

KarmaGYZ Feb 7, 2022

KarmaGYZ Feb 7, 2022

KarmaGYZ Feb 7, 2022

KarmaGYZ Feb 7, 2022

KarmaGYZ Feb 7, 2022

KarmaGYZ Feb 7, 2022

KarmaGYZ Feb 7, 2022

FangYongs commented Feb 8, 2022

KarmaGYZ left a comment

KarmaGYZ Feb 8, 2022

KarmaGYZ Feb 8, 2022

KarmaGYZ Feb 8, 2022

KarmaGYZ Feb 8, 2022

		"'Memory': the memory job store keeps the archived execution graphs in memory and "
		+ " it may cause FullGC or OOM when there are too many graphs"))

-                                                    "'Memory': the memory job store keeps the archived execution graphs in memory and "
-                                                            + " it may cause FullGC or OOM when there are too many graphs"))
+                                                    "'Memory': the memory job store keeps the archived execution graphs in memory. You "
+                                                            + "may need to limit the %s to mitigate FullGC or OOM when there are too many graphs",
+                                                            code(JOB_STORE_MAX_CAPACITY.key())))

	private final ScheduledExecutor scheduledExecutor;
	@Nullable private final ScheduledExecutor scheduledExecutor;

	private ScheduledFuture<?> cleanupFuture;
	@Nullable private ScheduledFuture<?> cleanupFuture;

	return (int) serializableExecutionGraphInfos.size();
	return Math.toIntExact(serializableExecutionGraphInfos.size());

	// check that the persisted file has been deleted
	// check that the store is empty

[FLINK-25329][runtime] Support memory execution graph store in session cluster #18360

[FLINK-25329][runtime] Support memory execution graph store in session cluster #18360

Conversation

FangYongs commented Jan 14, 2022

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 14, 2022

Automated Checks

Review Progress

flinkbot commented Jan 14, 2022 • edited

CI report:

FangYongs commented Jan 16, 2022

FangYongs commented Jan 17, 2022

KarmaGYZ commented Jan 19, 2022

FangYongs commented Jan 24, 2022

KarmaGYZ left a comment

Choose a reason for hiding this comment

FangYongs commented Jan 25, 2022

KarmaGYZ commented Jan 25, 2022

FangYongs commented Jan 30, 2022

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

FangYongs commented Feb 8, 2022

KarmaGYZ 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

flinkbot commented Jan 14, 2022 •

edited