[FLINK-37133][table] Support Submitting Refresh Job of Materialized Table to Yarn/K8s #25988

hackergin · 2025-01-15T07:43:00Z

What is the purpose of the change

Support submitting refresh task of materialized table to yarn/k8s

Brief change log

Support submitting refresh task of materialized table to yarn/k8s

Verifying this change

Some e2e test case is added to verify this feature.

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? (yes)
If yes, how is the feature documented? (not documented)

flinkbot · 2025-01-15T09:40:50Z

CI report:

b34eeb3 Azure: CANCELED

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

flink-end-to-end-tests/test-scripts/test_kubernetes_materialized_table.sh

.../java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java

lsyldliu

@hackergin Thanks for your contribution, I left some comments.

Due to the e2e test only cover one case, so I think we need to test yarn and k8s manually. We should cover the following cases:

yarn-application: continuous mode and full mode, create & suspend & resume & drop action
yarn-session: continuous mode and full mode, create & suspend & resume & drop action
k8s-application: continuous mode and full mode, create & suspend & resume & drop action
k8s-session: continuous mode and full mode, create & suspend & resume & drop action

...link-table-common/src/main/java/org/apache/flink/table/refresh/ContinuousRefreshHandler.java

...utils/src/main/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalog.java

...ateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationExecutor.java

...l-gateway/src/main/java/org/apache/flink/table/gateway/service/application/ScriptRunner.java

.../java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java

flink-end-to-end-tests/test-scripts/test_kubernetes_materialized_table.sh

lsyldliu · 2025-01-16T06:49:13Z

flink-end-to-end-tests/test-scripts/test_kubernetes_materialized_table.sh

+# 2. suspend & resume materialized table in continuous mode
+execute_statement $session_handle "alter materialized table my_materialized_table_in_continuous_mode suspend"
+
+kubectl delete deployment $APPLICATION_CLUSTER_ID


Why do we need to manually delete clusters？

In my test, I found that after "stop with Savepoint", the deployment does not end. I need to spend some time to verify it again. In theory, it should exit automatically.

After my testing, it seems that the failure of deployment to exit normally is a bug in SqlDriver when splitting sql. When there is no semicolon at the end of the sql statement, it will enter an infinite loop and thus infinitely submit new jobs. Since SqlDriver needs time to fix this bug, currently we can add a semicolon by default when generate statement of the refresh job to avoid this problem.

After a discuss with @fsk119 , We decide to directly fix this is this pr, the commit is: de3675e

After a discuss with @fsk119 , We decide to directly fix this is this pr, the commit is: de3675e

BTW, the CI is failed.

It seems there are some issues with executing DML statements, so I will switch to executing a simple DDL statement instead.

lsyldliu · 2025-01-16T07:02:48Z

@hackergin Thanks for your contribution, I left some comments.

Due to the e2e test only cover one case, so I think we need to test yarn and k8s manually. We should cover the following cases:

yarn-application: continuous mode and full mode, create & suspend & resume & drop action

yarn-session: continuous mode and full mode, create & suspend & resume & drop action

k8s-application: continuous mode and full mode, create & suspend & resume & drop action

k8s-session: continuous mode and full mode, create & suspend & resume & drop action

We can make these test cases work as one part of crossing-team test, but we should try our best to start crossing-team test as soon as possible.

davidradl · 2025-01-16T14:04:19Z

Reviewed by Chi on 16/01/2025 Go back to the submitter with review comments.

.../java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java

lsyldliu · 2025-01-17T09:29:44Z

.../java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java

+            Configuration executionConfig,
+            OperationExecutor operationExecutor,
+            OperationHandle operationHandle) {
+        if (isApplicationMode(operationExecutor.getSessionContext().getSessionConf())) {


For minicluster, is it not remote?

.../java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java

...link-table-common/src/main/java/org/apache/flink/table/refresh/ContinuousRefreshHandler.java

lsyldliu · 2025-01-19T15:33:21Z

...link-table-common/src/main/java/org/apache/flink/table/refresh/ContinuousRefreshHandler.java

    public Optional<String> getRestorePath() {
        return Optional.ofNullable(restorePath);
    }

    @Override
    public String asSummaryString() {
        return String.format(
-                "{\njobId=%s,\n executionTarget=%s%s\n}",
+                "{\n jobId=%s,\n executionTarget=%s,\n clusterId=%s%s\n}",


Suggested change

"{\n jobId=%s,\n executionTarget=%s,\n clusterId=%s%s\n}",

"{\n executionTarget=%s,\n clusterId=%s,\n jobId=%s%s\n}",

lsyldliu · 2025-01-19T15:36:36Z

.../java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java

+        }
+    }
+
+    private static @Nullable String getClusterIdKeyName(String targetName) {


I think we can use Optional<String> as the return type, it would more better.

…hout trailing semicolon

...utils/src/main/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalog.java

lsyldliu · 2025-01-20T01:52:17Z

...gateway/src/main/java/org/apache/flink/table/gateway/service/application/ScriptExecutor.java

@@ -194,6 +194,7 @@ public boolean hasNext() {
                    return true;
                }
            }
+            position = script.length();


Because I'm not familiar with related code logic, I'm curious why adding this line of code can fix the issue.

The position keeps track of the current location in the SQL text being parsed. Previously, the logic only updated the position when a semicolon was encountered. However, if there is no semicolon at the end of the text—meaning we’ve finished processing all the text without updating the position—the next time we attempt to split, it will start from the previous position again.

lsyldliu · 2025-01-20T01:53:19Z

flink-table/flink-sql-gateway/src/test/resources/application/no_semicolon.q

+  'connector' = 'datagen'
+); -- This is another comment
+
+DESCRIBE src


Why change the select statement to describe statement?

The SELECT statement still requires a MiniCluster for execution. Therefore, I referred to other tests in this unit test and only verified DDL-related operations without relying on the MiniCluster.

.../java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java

…ion when constructing table paths

…duler when closing without open

…able to Yarn/K8s

hackergin · 2025-01-20T04:02:39Z

@flinkbot run azure

hackergin · 2025-01-20T06:28:31Z

test_ci connect failed because : https://issues.apache.org/jira/browse/FLINK-36290
test_ci table failed because : https://issues.apache.org/jira/browse/FLINK-36167

hackergin · 2025-01-20T06:50:06Z

@flinkbot run azure

hackergin · 2025-01-20T09:13:40Z

@flinkbot run azure

hackergin · 2025-01-20T09:14:45Z

https://github.com/hackergin/flink/actions/runs/12860642915 github ci has passed

hackergin · 2025-01-20T09:33:09Z

@flinkbot run azure

lsyldliu

LGTM

lsyldliu · 2025-01-20T09:58:50Z

The related tests have been verified manually, and merged it.

hackergin force-pushed the support-session-mode-mt branch 4 times, most recently from 4e75a35 to efec0ab Compare January 15, 2025 09:10

hackergin force-pushed the support-session-mode-mt branch 4 times, most recently from d04dd76 to 1a18003 Compare January 15, 2025 13:42

davidradl reviewed Jan 15, 2025

View reviewed changes

flink-end-to-end-tests/test-scripts/test_kubernetes_materialized_table.sh Show resolved Hide resolved

davidradl reviewed Jan 15, 2025

View reviewed changes

.../java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java Outdated Show resolved Hide resolved

davidradl reviewed Jan 15, 2025

View reviewed changes

.../java/org/apache/flink/table/gateway/service/materializedtable/MaterializedTableManager.java Show resolved Hide resolved

lsyldliu reviewed Jan 16, 2025

View reviewed changes

lsyldliu reviewed Jan 17, 2025

View reviewed changes

hackergin changed the title ~~[FLINK-37133][table] Support Submitting Refresh Tasks of Materialized Table to Yarn/K8s~~ [FLINK-37133][table] Support Submitting Refresh Job of Materialized Table to Yarn/K8s Jan 18, 2025

hackergin force-pushed the support-session-mode-mt branch 3 times, most recently from 191165a to bf1ae6d Compare January 19, 2025 13:11

lsyldliu reviewed Jan 19, 2025

View reviewed changes

[FLINK-37133][table] Fix ScriptExecutor to properly split scripts wit…

9f1cd6a

…hout trailing semicolon

hackergin force-pushed the support-session-mode-mt branch from bf1ae6d to 095a54f Compare January 20, 2025 01:36

lsyldliu reviewed Jan 20, 2025

View reviewed changes

hackergin added 2 commits January 20, 2025 10:05

[FLINK-37133][table] Fix TestFileSystemCatalog losing scheme informat…

74dffd2

…ion when constructing table paths

[FLINK-37133][table] Fix NullPointerException in EmbeddedWorkflowSche…

fed9274

…duler when closing without open

hackergin force-pushed the support-session-mode-mt branch 2 times, most recently from 5840cd8 to 0bd4f8b Compare January 20, 2025 02:39

[FLINK-37133][table] Support Submitting Refresh Job of Materialized T…

b34eeb3

…able to Yarn/K8s

hackergin force-pushed the support-session-mode-mt branch from 0bd4f8b to b34eeb3 Compare January 20, 2025 03:49

lsyldliu approved these changes Jan 20, 2025

View reviewed changes

lsyldliu merged commit 4a2535c into apache:master Jan 20, 2025

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-37133][table] Support Submitting Refresh Job of Materialized Table to Yarn/K8s #25988

[FLINK-37133][table] Support Submitting Refresh Job of Materialized Table to Yarn/K8s #25988

hackergin commented Jan 15, 2025

flinkbot commented Jan 15, 2025 •

edited

Loading

lsyldliu left a comment

lsyldliu Jan 16, 2025

hackergin Jan 17, 2025

hackergin Jan 19, 2025 •

edited

Loading

hackergin Jan 19, 2025 •

edited

Loading

lsyldliu Jan 20, 2025

hackergin Jan 20, 2025 •

edited

Loading

lsyldliu commented Jan 16, 2025

davidradl commented Jan 16, 2025

lsyldliu Jan 17, 2025

lsyldliu Jan 19, 2025

lsyldliu Jan 19, 2025

lsyldliu Jan 20, 2025

hackergin Jan 20, 2025

lsyldliu Jan 20, 2025

hackergin Jan 20, 2025

hackergin commented Jan 20, 2025

hackergin commented Jan 20, 2025

hackergin commented Jan 20, 2025

hackergin commented Jan 20, 2025

hackergin commented Jan 20, 2025

hackergin commented Jan 20, 2025

lsyldliu left a comment

lsyldliu commented Jan 20, 2025

	"{\n jobId=%s,\n executionTarget=%s,\n clusterId=%s%s\n}",
	"{\n executionTarget=%s,\n clusterId=%s,\n jobId=%s%s\n}",

[FLINK-37133][table] Support Submitting Refresh Job of Materialized Table to Yarn/K8s #25988

[FLINK-37133][table] Support Submitting Refresh Job of Materialized Table to Yarn/K8s #25988

Conversation

hackergin commented Jan 15, 2025

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 15, 2025 • edited Loading

CI report:

lsyldliu left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hackergin Jan 19, 2025 • edited Loading

Choose a reason for hiding this comment

hackergin Jan 19, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hackergin Jan 20, 2025 • edited Loading

Choose a reason for hiding this comment

lsyldliu commented Jan 16, 2025

davidradl commented Jan 16, 2025

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

hackergin commented Jan 20, 2025

hackergin commented Jan 20, 2025

hackergin commented Jan 20, 2025

hackergin commented Jan 20, 2025

hackergin commented Jan 20, 2025

hackergin commented Jan 20, 2025

lsyldliu left a comment

Choose a reason for hiding this comment

lsyldliu commented Jan 20, 2025

flinkbot commented Jan 15, 2025 •

edited

Loading

hackergin Jan 19, 2025 •

edited

Loading

hackergin Jan 19, 2025 •

edited

Loading

hackergin Jan 20, 2025 •

edited

Loading