You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2023/05/29 06:16:15 UTC
[incubator-paimon] branch master updated: [flink] Use pipeline.name to action job name (#1244)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 11af346d2 [flink] Use pipeline.name to action job name (#1244)
11af346d2 is described below
commit 11af346d2268f2a51691896945e275e9d122dfda
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Mon May 29 14:16:11 2023 +0800
[flink] Use pipeline.name to action job name (#1244)
---
.../main/java/org/apache/paimon/flink/action/ActionBase.java | 11 +++++++++++
.../java/org/apache/paimon/flink/action/CompactAction.java | 2 +-
.../flink/action/cdc/mysql/MySqlSyncDatabaseAction.java | 2 +-
.../paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java | 2 +-
4 files changed, 14 insertions(+), 3 deletions(-)
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/ActionBase.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/ActionBase.java
index c12c280e4..c07480e1b 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/ActionBase.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/ActionBase.java
@@ -21,9 +21,14 @@ package org.apache.paimon.flink.action;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.flink.FlinkCatalog;
import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.flink.utils.StreamExecutionEnvironmentUtils;
import org.apache.paimon.options.CatalogOptions;
import org.apache.paimon.options.Options;
+import org.apache.flink.configuration.PipelineOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
import java.util.Map;
import java.util.UUID;
@@ -45,4 +50,10 @@ public abstract class ActionBase implements Action {
catalog = FlinkCatalogFactory.createPaimonCatalog(catalogOptions);
flinkCatalog = FlinkCatalogFactory.createCatalog(catalogName, catalog);
}
+
+ protected void execute(StreamExecutionEnvironment env, String defaultName) throws Exception {
+ ReadableConfig conf = StreamExecutionEnvironmentUtils.getConfiguration(env);
+ String name = conf.getOptional(PipelineOptions.NAME).orElse(defaultName);
+ env.execute(name);
+ }
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java
index 5b903cd01..4c4303ffc 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java
@@ -170,6 +170,6 @@ public class CompactAction extends TableActionBase {
public void run() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
build(env);
- env.execute("Compact job");
+ execute(env, "Compact job");
}
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
index f5bafa65d..ec9cc8f3c 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
@@ -433,6 +433,6 @@ public class MySqlSyncDatabaseAction extends ActionBase {
public void run() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
build(env);
- env.execute(String.format("MySQL-Paimon Database Sync: %s", database));
+ execute(env, String.format("MySQL-Paimon Database Sync: %s", database));
}
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java
index e13611574..33ae27b0e 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java
@@ -384,6 +384,6 @@ public class MySqlSyncTableAction extends ActionBase {
public void run() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
build(env);
- env.execute(String.format("MySQL-Paimon Table Sync: %s.%s", database, table));
+ execute(env, String.format("MySQL-Paimon Table Sync: %s.%s", database, table));
}
}