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/10/23 07:20:08 UTC
[incubator-paimon] branch master updated: [hotfix] Flink procedures should pass the execution environment to action code (#2165)
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 08a574b5e [hotfix] Flink procedures should pass the execution environment to action code (#2165)
08a574b5e is described below
commit 08a574b5eb1a4bb6940b4a4d6dc8da0d83cdb406
Author: yuzelin <33...@users.noreply.github.com>
AuthorDate: Mon Oct 23 15:20:02 2023 +0800
[hotfix] Flink procedures should pass the execution environment to action code (#2165)
---
.../cdc/MessageQueueSyncTableActionBase.java | 6 ++----
.../action/cdc/kafka/KafkaSyncDatabaseAction.java | 6 ++----
.../cdc/mongodb/MongoDBSyncDatabaseAction.java | 6 ++----
.../action/cdc/mongodb/MongoDBSyncTableAction.java | 6 ++----
.../action/cdc/mysql/MySqlSyncDatabaseAction.java | 6 ++----
.../action/cdc/mysql/MySqlSyncTableAction.java | 6 ++----
.../flink/action/cdc/CdcActionITCaseBase.java | 4 ++--
.../cdc/mysql/MySqlSyncDatabaseActionITCase.java | 4 ++--
.../org/apache/paimon/flink/action/Action.java | 5 +----
.../org/apache/paimon/flink/action/ActionBase.java | 22 +++++++++++++++++-----
.../apache/paimon/flink/action/CompactAction.java | 5 ++---
.../paimon/flink/action/CompactDatabaseAction.java | 13 ++++++-------
.../paimon/flink/action/SortCompactAction.java | 5 ++---
.../paimon/flink/procedure/MergeIntoProcedure.java | 1 +
.../paimon/flink/procedure/ProcedureBase.java | 7 ++++---
.../ChangelogWithKeyFileStoreTableITCase.java | 6 +++---
.../paimon/flink/action/CompactActionITCase.java | 3 ++-
.../flink/action/CompactDatabaseActionITCase.java | 20 ++++++++++++++------
18 files changed, 68 insertions(+), 63 deletions(-)
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSyncTableActionBase.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSyncTableActionBase.java
index 42fdff6a2..7c132cae1 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSyncTableActionBase.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSyncTableActionBase.java
@@ -35,7 +35,6 @@ import org.apache.paimon.table.FileStoreTable;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.connector.source.Source;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -149,7 +148,7 @@ public abstract class MessageQueueSyncTableActionBase extends ActionBase {
protected abstract String sourceName();
@Override
- public void build(StreamExecutionEnvironment env) throws Exception {
+ public void build() throws Exception {
Source<String, ?, ?> source = buildSource();
catalog.createDatabase(database, true);
@@ -273,8 +272,7 @@ public abstract class MessageQueueSyncTableActionBase extends ActionBase {
@Override
public void run() throws Exception {
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- build(env);
+ build();
env.execute(jobName());
}
}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseAction.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseAction.java
index fc9fa30bf..5cfe5a528 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseAction.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseAction.java
@@ -35,7 +35,6 @@ import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecordSchemaBuilder;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import javax.annotation.Nullable;
@@ -141,7 +140,7 @@ public class KafkaSyncDatabaseAction extends ActionBase {
}
@Override
- public void build(StreamExecutionEnvironment env) throws Exception {
+ public void build() throws Exception {
boolean caseSensitive = catalog.caseSensitive();
if (!caseSensitive) {
@@ -212,8 +211,7 @@ public class KafkaSyncDatabaseAction extends ActionBase {
@Override
public void run() throws Exception {
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- build(env);
+ build();
env.execute(String.format("KAFKA-Paimon Database Sync: %s", database));
}
}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseAction.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseAction.java
index b407e22a5..feb3075c8 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseAction.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseAction.java
@@ -34,7 +34,6 @@ import com.ververica.cdc.connectors.mongodb.source.MongoDBSource;
import com.ververica.cdc.connectors.mongodb.source.config.MongoDBSourceOptions;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import javax.annotation.Nullable;
@@ -117,7 +116,7 @@ public class MongoDBSyncDatabaseAction extends ActionBase {
}
@Override
- public void build(StreamExecutionEnvironment env) throws Exception {
+ public void build() throws Exception {
boolean caseSensitive = catalog.caseSensitive();
if (!caseSensitive) {
@@ -191,8 +190,7 @@ public class MongoDBSyncDatabaseAction extends ActionBase {
@Override
public void run() throws Exception {
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- build(env);
+ build();
env.execute(String.format("MongoDB-Paimon Database Sync: %s", database));
}
}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncTableAction.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncTableAction.java
index 95ff2ecb9..ba5b4b0ba 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncTableAction.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncTableAction.java
@@ -34,7 +34,6 @@ import com.ververica.cdc.connectors.mongodb.source.MongoDBSource;
import com.ververica.cdc.connectors.mongodb.source.config.MongoDBSourceOptions;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.util.ArrayList;
import java.util.Arrays;
@@ -109,7 +108,7 @@ public class MongoDBSyncTableAction extends ActionBase {
}
@Override
- public void build(StreamExecutionEnvironment env) throws Exception {
+ public void build() throws Exception {
checkArgument(
mongodbConfig.contains(MongoDBSourceOptions.COLLECTION),
String.format(
@@ -214,8 +213,7 @@ public class MongoDBSyncTableAction extends ActionBase {
@Override
public void run() throws Exception {
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- build(env);
+ build();
execute(env, String.format("MongoDB-Paimon Database Sync: %s", database));
}
}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
index 6ebbc22c8..8f1f5c830 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
@@ -41,7 +41,6 @@ import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.source.config.MySqlSourceOptions;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -188,7 +187,7 @@ public class MySqlSyncDatabaseAction extends ActionBase {
}
@Override
- public void build(StreamExecutionEnvironment env) throws Exception {
+ public void build() throws Exception {
checkArgument(
!mySqlConfig.contains(MySqlSourceOptions.TABLE_NAME),
MySqlSourceOptions.TABLE_NAME.key()
@@ -406,8 +405,7 @@ public class MySqlSyncDatabaseAction extends ActionBase {
@Override
public void run() throws Exception {
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- build(env);
+ build();
execute(env, String.format("MySQL-Paimon Database Sync: %s", database));
}
}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java
index 130caab01..bc6c99932 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java
@@ -39,7 +39,6 @@ import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.source.config.MySqlSourceOptions;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.time.ZoneId;
import java.util.ArrayList;
@@ -153,7 +152,7 @@ public class MySqlSyncTableAction extends ActionBase {
}
@Override
- public void build(StreamExecutionEnvironment env) throws Exception {
+ public void build() throws Exception {
checkArgument(
mySqlConfig.contains(MySqlSourceOptions.TABLE_NAME),
String.format(
@@ -321,8 +320,7 @@ public class MySqlSyncTableAction extends ActionBase {
@Override
public void run() throws Exception {
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- build(env);
+ build();
execute(env, String.format("MySQL-Paimon Table Sync: %s.%s", database, table));
}
}
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/CdcActionITCaseBase.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/CdcActionITCaseBase.java
index 919a04b51..6cc2de9ba 100644
--- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/CdcActionITCaseBase.java
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/CdcActionITCaseBase.java
@@ -185,8 +185,8 @@ public class CdcActionITCaseBase extends ActionITCaseBase {
return Arrays.asList(argKey, nullable.toString());
}
- public <T extends ActionBase> JobClient runActionWithDefaultEnv(T action) throws Exception {
- action.build(env);
+ public JobClient runActionWithDefaultEnv(ActionBase action) throws Exception {
+ action.withStreamExecutionEnvironment(env).build();
JobClient client = env.executeAsync();
waitJobRunning(client);
return client;
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionITCase.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionITCase.java
index bc2ee32b1..cc521b9ff 100644
--- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionITCase.java
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionITCase.java
@@ -858,7 +858,7 @@ public class MySqlSyncDatabaseActionITCase extends MySqlActionITCaseBase {
.includingTables("t.+")
.withMode(COMBINED.configString())
.build();
- action.build(env);
+ action.withStreamExecutionEnvironment(env).build();
if (Objects.nonNull(savepointPath)) {
StreamGraph streamGraph = env.getStreamGraph();
@@ -1167,7 +1167,7 @@ public class MySqlSyncDatabaseActionITCase extends MySqlActionITCaseBase {
.ignoreIncompatible(true)
.withMode(COMBINED.configString())
.build();
- action.build(env);
+ action.build();
assertThat(action.monitoredTables())
.containsOnly(
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/Action.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/Action.java
index 9c4d24211..3f4e92774 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/Action.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/Action.java
@@ -18,8 +18,6 @@
package org.apache.paimon.flink.action;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-
/** Abstract class for Flink actions. */
public interface Action {
@@ -36,8 +34,7 @@ public interface Action {
* <p>By default, this method is empty and can be overridden by subclasses to provide custom
* setup logic.
*
- * @param env The Flink Stream Execution Environment where the action will be executed.
* @throws Exception If an error occurs during the build process.
*/
- default void build(StreamExecutionEnvironment env) throws Exception {}
+ default void build() throws Exception {}
}
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 9c2687bd5..0d9833dac 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
@@ -48,8 +48,9 @@ public abstract class ActionBase implements Action {
protected final Catalog catalog;
protected final FlinkCatalog flinkCatalog;
protected final String catalogName = "paimon-" + UUID.randomUUID();
- protected final StreamExecutionEnvironment env;
- protected final StreamTableEnvironment batchTEnv;
+
+ protected StreamExecutionEnvironment env;
+ protected StreamTableEnvironment batchTEnv;
public ActionBase(String warehouse, Map<String, String> catalogConfig) {
catalogOptions = Options.fromMap(catalogConfig);
@@ -57,10 +58,21 @@ public abstract class ActionBase implements Action {
catalog = FlinkCatalogFactory.createPaimonCatalog(catalogOptions);
flinkCatalog = FlinkCatalogFactory.createCatalog(catalogName, catalog, catalogOptions);
- env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ // use the default env if user doesn't pass one
+ initFlinkEnv(StreamExecutionEnvironment.getExecutionEnvironment());
+ }
+
+ public ActionBase withStreamExecutionEnvironment(StreamExecutionEnvironment env) {
+ initFlinkEnv(env);
+ return this;
+ }
+
+ private void initFlinkEnv(StreamExecutionEnvironment env) {
+ this.env = env;
// we enable object reuse, we copy the un-reusable object ourselves.
- env.getConfig().enableObjectReuse();
- batchTEnv = StreamTableEnvironment.create(env, EnvironmentSettings.inBatchMode());
+ this.env.getConfig().enableObjectReuse();
+ batchTEnv = StreamTableEnvironment.create(this.env, EnvironmentSettings.inBatchMode());
// register flink catalog to table environment
batchTEnv.registerCatalog(flinkCatalog.getName(), flinkCatalog);
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 09746f5b1..d678dc2df 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
@@ -75,7 +75,7 @@ public class CompactAction extends TableActionBase {
}
@Override
- public void build(StreamExecutionEnvironment env) {
+ public void build() {
ReadableConfig conf = StreamExecutionEnvironmentUtils.getConfiguration(env);
boolean isStreaming =
conf.get(ExecutionOptions.RUNTIME_MODE) == RuntimeExecutionMode.STREAMING;
@@ -120,8 +120,7 @@ public class CompactAction extends TableActionBase {
@Override
public void run() throws Exception {
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- build(env);
+ build();
execute(env, "Compact job");
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java
index e07ec9bc9..3219994ee 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java
@@ -116,15 +116,15 @@ public class CompactDatabaseAction extends ActionBase {
}
@Override
- public void build(StreamExecutionEnvironment env) {
+ public void build() {
if (databaseCompactMode == MultiTablesSinkMode.DIVIDED) {
- buildForDividedMode(env);
+ buildForDividedMode();
} else {
- buildForCombinedMode(env);
+ buildForCombinedMode();
}
}
- private void buildForDividedMode(StreamExecutionEnvironment env) {
+ private void buildForDividedMode() {
try {
List<String> databases = catalog.listDatabases();
for (String databaseName : databases) {
@@ -188,7 +188,7 @@ public class CompactDatabaseAction extends ActionBase {
}
}
- private void buildForCombinedMode(StreamExecutionEnvironment env) {
+ private void buildForCombinedMode() {
ReadableConfig conf = StreamExecutionEnvironmentUtils.getConfiguration(env);
boolean isStreaming =
@@ -240,8 +240,7 @@ public class CompactDatabaseAction extends ActionBase {
@Override
public void run() throws Exception {
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- build(env);
+ build();
execute(env, "Compact database job");
}
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/SortCompactAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/SortCompactAction.java
index b665714d0..0a4f0138c 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/SortCompactAction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/SortCompactAction.java
@@ -31,7 +31,6 @@ import org.apache.paimon.table.FileStoreTable;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.configuration.ExecutionOptions;
import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.data.RowData;
import org.slf4j.Logger;
@@ -65,12 +64,12 @@ public class SortCompactAction extends CompactAction {
@Override
public void run() throws Exception {
- build(env);
+ build();
execute(env, "Sort Compact Job");
}
@Override
- public void build(StreamExecutionEnvironment env) {
+ public void build() {
// only support batch sort yet
if (env.getConfiguration().get(ExecutionOptions.RUNTIME_MODE)
!= RuntimeExecutionMode.BATCH) {
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MergeIntoProcedure.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MergeIntoProcedure.java
index e6658346b..450646803 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MergeIntoProcedure.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MergeIntoProcedure.java
@@ -221,6 +221,7 @@ public class MergeIntoProcedure extends ProcedureBase {
action.withMatchedDelete(matchedDeleteCondition);
}
+ action.withStreamExecutionEnvironment(procedureContext.getExecutionEnvironment());
MergeIntoActionFactory.validate(action);
DataStream<RowData> dataStream = action.buildDataStream();
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/ProcedureBase.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/ProcedureBase.java
index 91b8a1aa1..f83754c54 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/ProcedureBase.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/ProcedureBase.java
@@ -20,7 +20,7 @@ package org.apache.paimon.flink.procedure;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.factories.Factory;
-import org.apache.paimon.flink.action.Action;
+import org.apache.paimon.flink.action.ActionBase;
import org.apache.paimon.flink.utils.StreamExecutionEnvironmentUtils;
import org.apache.paimon.utils.StringUtils;
@@ -65,10 +65,11 @@ public abstract class ProcedureBase implements Procedure, Factory {
}
protected String[] execute(
- ProcedureContext procedureContext, Action action, String defaultJobName)
+ ProcedureContext procedureContext, ActionBase action, String defaultJobName)
throws Exception {
StreamExecutionEnvironment env = procedureContext.getExecutionEnvironment();
- action.build(env);
+ action.withStreamExecutionEnvironment(env);
+ action.build();
ReadableConfig conf = StreamExecutionEnvironmentUtils.getConfiguration(env);
String name = conf.getOptional(PipelineOptions.NAME).orElse(defaultJobName);
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ChangelogWithKeyFileStoreTableITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ChangelogWithKeyFileStoreTableITCase.java
index c2898856a..e49c830d0 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ChangelogWithKeyFileStoreTableITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ChangelogWithKeyFileStoreTableITCase.java
@@ -164,7 +164,7 @@ public class ChangelogWithKeyFileStoreTableITCase extends AbstractTestBase {
StreamExecutionEnvironment env = createStreamExecutionEnvironment(2000);
env.setParallelism(1);
env.setRestartStrategy(RestartStrategies.noRestart());
- new CompactAction(path, "default", "T").build(env);
+ new CompactAction(path, "default", "T").withStreamExecutionEnvironment(env).build();
JobClient client = env.executeAsync();
// write records for a while
@@ -408,7 +408,7 @@ public class ChangelogWithKeyFileStoreTableITCase extends AbstractTestBase {
StreamExecutionEnvironment env =
createStreamExecutionEnvironment(random.nextInt(1900) + 100);
env.setParallelism(2);
- new CompactAction(path, "default", "T").build(env);
+ new CompactAction(path, "default", "T").withStreamExecutionEnvironment(env).build();
env.executeAsync();
}
@@ -442,7 +442,7 @@ public class ChangelogWithKeyFileStoreTableITCase extends AbstractTestBase {
StreamExecutionEnvironment env =
createStreamExecutionEnvironment(random.nextInt(1900) + 100);
env.setParallelism(2);
- new CompactAction(path, "default", "T").build(env);
+ new CompactAction(path, "default", "T").withStreamExecutionEnvironment(env).build();
env.executeAsync();
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactActionITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactActionITCase.java
index eaea0e3e9..9206cc95f 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactActionITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactActionITCase.java
@@ -290,7 +290,8 @@ public class CompactActionITCase extends CompactActionITCaseBase {
new CompactAction(warehouse, database, tableName)
.withPartitions(getSpecifiedPartitions())
- .build(env);
+ .withStreamExecutionEnvironment(env)
+ .build();
if (isStreaming) {
env.executeAsync();
} else {
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactDatabaseActionITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactDatabaseActionITCase.java
index 0e14e1ed2..c0c0d4337 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactDatabaseActionITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactDatabaseActionITCase.java
@@ -134,7 +134,8 @@ public class CompactDatabaseActionITCase extends CompactActionITCaseBase {
StreamExecutionEnvironment env = buildDefaultEnv(false);
new CompactDatabaseAction(warehouse, Collections.emptyMap())
.withDatabaseCompactMode(mode)
- .build(env);
+ .withStreamExecutionEnvironment(env)
+ .build();
env.execute();
} else {
callProcedure(String.format("CALL sys.compact_database('', '%s')", mode), false, true);
@@ -208,7 +209,9 @@ public class CompactDatabaseActionITCase extends CompactActionITCaseBase {
if (ThreadLocalRandom.current().nextBoolean()) {
StreamExecutionEnvironment env = buildDefaultEnv(true);
if (mode.equals("divided")) {
- new CompactDatabaseAction(warehouse, new HashMap<>()).build(env);
+ new CompactDatabaseAction(warehouse, new HashMap<>())
+ .withStreamExecutionEnvironment(env)
+ .build();
} else {
// if CoreOptions.CONTINUOUS_DISCOVERY_INTERVAL.key() use default value, the cost
// time in combined mode will be over 1 min
@@ -217,7 +220,8 @@ public class CompactDatabaseActionITCase extends CompactActionITCaseBase {
.withTableOptions(
Collections.singletonMap(
CoreOptions.CONTINUOUS_DISCOVERY_INTERVAL.key(), "1s"))
- .build(env);
+ .withStreamExecutionEnvironment(env)
+ .build();
}
env.executeAsync();
} else {
@@ -484,7 +488,7 @@ public class CompactDatabaseActionITCase extends CompactActionITCaseBase {
Collections.singletonMap(
CoreOptions.CONTINUOUS_DISCOVERY_INTERVAL.key(), "1s"));
}
- action.build(env);
+ action.withStreamExecutionEnvironment(env).build();
env.execute();
} else {
if (mode.equals("divided")) {
@@ -584,7 +588,9 @@ public class CompactDatabaseActionITCase extends CompactActionITCaseBase {
if (ThreadLocalRandom.current().nextBoolean()) {
StreamExecutionEnvironment env = buildDefaultEnv(true);
- new CompactDatabaseAction(warehouse, new HashMap<>()).build(env);
+ new CompactDatabaseAction(warehouse, new HashMap<>())
+ .withStreamExecutionEnvironment(env)
+ .build();
env.executeAsync();
} else {
callProcedure("CALL sys.compact_database()");
@@ -655,7 +661,9 @@ public class CompactDatabaseActionITCase extends CompactActionITCaseBase {
if (ThreadLocalRandom.current().nextBoolean()) {
StreamExecutionEnvironment env = buildDefaultEnv(false);
- new CompactDatabaseAction(warehouse, new HashMap<>()).build(env);
+ new CompactDatabaseAction(warehouse, new HashMap<>())
+ .withStreamExecutionEnvironment(env)
+ .build();
env.execute();
} else {
callProcedure("CALL sys.compact_database()", false, true);