You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ro...@apache.org on 2024/06/01 14:47:58 UTC
(flink) 01/02: [FLINK-35199][table] Remove dynamic options and add initialization configuration to CreatePeriodicRefreshWorkflow
This is an automated email from the ASF dual-hosted git repository.
ron pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 309e3246e0232a0a363aa44ab6d5524133f8f548
Author: Feng Jin <ji...@gmail.com>
AuthorDate: Fri May 31 11:41:12 2024 +0800
[FLINK-35199][table] Remove dynamic options and add initialization configuration to CreatePeriodicRefreshWorkflow
---
.../scheduler/CreateEmbeddedSchedulerWorkflowHandler.java | 5 +++--
.../CreateEmbeddedSchedulerWorkflowRequestBody.java | 15 +++++++--------
.../table/gateway/workflow/EmbeddedWorkflowScheduler.java | 2 +-
.../apache/flink/table/gateway/workflow/WorkflowInfo.java | 12 +++++++++++-
.../table/gateway/workflow/QuartzSchedulerUtilsTest.java | 11 ++++++++---
.../src/test/resources/sql_gateway_rest_api_v3.snapshot | 2 +-
.../table/workflow/CreatePeriodicRefreshWorkflow.java | 10 +++++-----
7 files changed, 36 insertions(+), 21 deletions(-)
diff --git a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/handler/materializedtable/scheduler/CreateEmbeddedSchedulerWorkflowHandler.java b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/handler/materializedtable/scheduler/CreateEmbeddedSchedulerWorkflowHandler.java
index b52094a39e6..9a7071ab935 100644
--- a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/handler/materializedtable/scheduler/CreateEmbeddedSchedulerWorkflowHandler.java
+++ b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/handler/materializedtable/scheduler/CreateEmbeddedSchedulerWorkflowHandler.java
@@ -72,14 +72,15 @@ public class CreateEmbeddedSchedulerWorkflowHandler
String materializedTableIdentifier =
request.getRequestBody().getMaterializedTableIdentifier();
String cronExpression = request.getRequestBody().getCronExpression();
- Map<String, String> dynamicOptions = request.getRequestBody().getDynamicOptions();
+ Map<String, String> initConfig = request.getRequestBody().getInitConfig();
Map<String, String> executionConfig = request.getRequestBody().getExecutionConfig();
String customScheduleTime = request.getRequestBody().getCustomScheduleTime();
String restEndpointURL = request.getRequestBody().getRestEndpointUrl();
WorkflowInfo workflowInfo =
new WorkflowInfo(
materializedTableIdentifier,
- dynamicOptions == null ? Collections.emptyMap() : dynamicOptions,
+ Collections.emptyMap(),
+ initConfig == null ? Collections.emptyMap() : initConfig,
executionConfig == null ? Collections.emptyMap() : executionConfig,
customScheduleTime,
restEndpointURL);
diff --git a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/materializedtable/scheduler/CreateEmbeddedSchedulerWorkflowRequestBody.java b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/materializedtable/scheduler/CreateEmbeddedSchedulerWorkflowRequestBody.java
index e0628933560..d0ebf3201ba 100644
--- a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/materializedtable/scheduler/CreateEmbeddedSchedulerWorkflowRequestBody.java
+++ b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/message/materializedtable/scheduler/CreateEmbeddedSchedulerWorkflowRequestBody.java
@@ -34,7 +34,7 @@ public class CreateEmbeddedSchedulerWorkflowRequestBody implements RequestBody {
private static final String FIELD_NAME_MATERIALIZED_TABLE = "materializedTableIdentifier";
private static final String FIELD_NAME_CRON_EXPRESSION = "cronExpression";
- private static final String FIELD_NAME_DYNAMIC_OPTIONS = "dynamicOptions";
+ private static final String FIELD_NAME_INIT_CONFIG = "initConfig";
private static final String FIELD_NAME_EXECUTION_CONFIG = "executionConfig";
private static final String FIELD_NAME_SCHEDULE_TIME = "customScheduleTime";
private static final String FIELD_NAME_REST_ENDPOINT_URL = "restEndpointUrl";
@@ -45,9 +45,8 @@ public class CreateEmbeddedSchedulerWorkflowRequestBody implements RequestBody {
@JsonProperty(FIELD_NAME_CRON_EXPRESSION)
private final String cronExpression;
- @JsonProperty(FIELD_NAME_DYNAMIC_OPTIONS)
- @Nullable
- private final Map<String, String> dynamicOptions;
+ @JsonProperty(FIELD_NAME_INIT_CONFIG)
+ private final Map<String, String> initConfig;
@JsonProperty(FIELD_NAME_EXECUTION_CONFIG)
@Nullable
@@ -63,14 +62,14 @@ public class CreateEmbeddedSchedulerWorkflowRequestBody implements RequestBody {
public CreateEmbeddedSchedulerWorkflowRequestBody(
@JsonProperty(FIELD_NAME_MATERIALIZED_TABLE) String materializedTableIdentifier,
@JsonProperty(FIELD_NAME_CRON_EXPRESSION) String cronExpression,
- @Nullable @JsonProperty(FIELD_NAME_DYNAMIC_OPTIONS) Map<String, String> dynamicOptions,
+ @Nullable @JsonProperty(FIELD_NAME_INIT_CONFIG) Map<String, String> initConfig,
@Nullable @JsonProperty(FIELD_NAME_EXECUTION_CONFIG)
Map<String, String> executionConfig,
@Nullable @JsonProperty(FIELD_NAME_SCHEDULE_TIME) String customScheduleTime,
@JsonProperty(FIELD_NAME_REST_ENDPOINT_URL) String restEndpointUrl) {
this.materializedTableIdentifier = materializedTableIdentifier;
this.cronExpression = cronExpression;
- this.dynamicOptions = dynamicOptions;
+ this.initConfig = initConfig;
this.executionConfig = executionConfig;
this.customScheduleTime = customScheduleTime;
this.restEndpointUrl = restEndpointUrl;
@@ -81,8 +80,8 @@ public class CreateEmbeddedSchedulerWorkflowRequestBody implements RequestBody {
}
@Nullable
- public Map<String, String> getDynamicOptions() {
- return dynamicOptions;
+ public Map<String, String> getInitConfig() {
+ return initConfig;
}
@Nullable
diff --git a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/workflow/EmbeddedWorkflowScheduler.java b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/workflow/EmbeddedWorkflowScheduler.java
index 98e1c15d72d..5d92cbda4fe 100644
--- a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/workflow/EmbeddedWorkflowScheduler.java
+++ b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/workflow/EmbeddedWorkflowScheduler.java
@@ -100,7 +100,7 @@ public class EmbeddedWorkflowScheduler implements WorkflowScheduler<EmbeddedRefr
new CreateEmbeddedSchedulerWorkflowRequestBody(
materializedTableIdentifier.asSerializableString(),
periodicRefreshWorkflow.getCronExpression(),
- periodicRefreshWorkflow.getDynamicOptions(),
+ periodicRefreshWorkflow.getInitConfig(),
periodicRefreshWorkflow.getExecutionConfig(),
null,
periodicRefreshWorkflow.getRestEndpointUrl());
diff --git a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/workflow/WorkflowInfo.java b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/workflow/WorkflowInfo.java
index 0df6be22008..e31410739d6 100644
--- a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/workflow/WorkflowInfo.java
+++ b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/workflow/WorkflowInfo.java
@@ -37,6 +37,7 @@ public class WorkflowInfo {
private final String materializedTableIdentifier;
private final Map<String, String> dynamicOptions;
+ private final Map<String, String> initConfig;
private final Map<String, String> executionConfig;
private final @Nullable String customSchedulerTime;
@@ -47,11 +48,13 @@ public class WorkflowInfo {
public WorkflowInfo(
@JsonProperty("materializedTableIdentifier") String materializedTableIdentifier,
@JsonProperty("dynamicOptions") Map<String, String> dynamicOptions,
+ @JsonProperty("initConfig") Map<String, String> initConfig,
@JsonProperty("executionConfig") Map<String, String> executionConfig,
@JsonProperty("customSchedulerTime") @Nullable String customSchedulerTime,
@JsonProperty("restEndpointUrl") String restEndpointUrl) {
this.materializedTableIdentifier = materializedTableIdentifier;
this.dynamicOptions = dynamicOptions;
+ this.initConfig = initConfig;
this.executionConfig = executionConfig;
this.customSchedulerTime = customSchedulerTime;
this.restEndpointUrl = restEndpointUrl;
@@ -65,6 +68,10 @@ public class WorkflowInfo {
return dynamicOptions;
}
+ public Map<String, String> getInitConfig() {
+ return initConfig;
+ }
+
public Map<String, String> getExecutionConfig() {
return executionConfig;
}
@@ -88,7 +95,7 @@ public class WorkflowInfo {
}
WorkflowInfo that = (WorkflowInfo) o;
return Objects.equals(materializedTableIdentifier, that.materializedTableIdentifier)
- && Objects.equals(dynamicOptions, that.dynamicOptions)
+ && Objects.equals(initConfig, that.initConfig)
&& Objects.equals(executionConfig, that.executionConfig)
&& Objects.equals(customSchedulerTime, that.customSchedulerTime)
&& Objects.equals(restEndpointUrl, that.restEndpointUrl);
@@ -99,6 +106,7 @@ public class WorkflowInfo {
return Objects.hash(
materializedTableIdentifier,
dynamicOptions,
+ initConfig,
executionConfig,
customSchedulerTime,
restEndpointUrl);
@@ -112,6 +120,8 @@ public class WorkflowInfo {
+ '\''
+ ", dynamicOptions="
+ dynamicOptions
+ + ", initConfig="
+ + initConfig
+ ", executionConfig="
+ executionConfig
+ ", customSchedulerTime='"
diff --git a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/workflow/QuartzSchedulerUtilsTest.java b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/workflow/QuartzSchedulerUtilsTest.java
index fd0b9a748c5..414580b8698 100644
--- a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/workflow/QuartzSchedulerUtilsTest.java
+++ b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/workflow/QuartzSchedulerUtilsTest.java
@@ -48,14 +48,19 @@ public class QuartzSchedulerUtilsTest {
dynamicOptions.put("k1", "v1");
dynamicOptions.put("k2", "v2");
+ Map<String, String> initConfig = new HashMap<>();
+ initConfig.put("k3", "v4");
+ initConfig.put("k4", "v5");
+
Map<String, String> executionConfig = new HashMap<>();
- executionConfig.put("k3", "v3");
- executionConfig.put("k4", "v4");
+ executionConfig.put("k5", "v5");
+ executionConfig.put("k6", "v6");
String scheduleTime = "2023-04-04 20:00:00";
String url = "http://localhost:8083";
WorkflowInfo expected =
- new WorkflowInfo(identifier, dynamicOptions, executionConfig, scheduleTime, url);
+ new WorkflowInfo(
+ identifier, dynamicOptions, initConfig, executionConfig, scheduleTime, url);
WorkflowInfo serde = fromJson(toJson(expected), WorkflowInfo.class);
assertThat(serde).isEqualTo(expected);
diff --git a/flink-table/flink-sql-gateway/src/test/resources/sql_gateway_rest_api_v3.snapshot b/flink-table/flink-sql-gateway/src/test/resources/sql_gateway_rest_api_v3.snapshot
index f11c3a34cda..9ca90ea593c 100644
--- a/flink-table/flink-sql-gateway/src/test/resources/sql_gateway_rest_api_v3.snapshot
+++ b/flink-table/flink-sql-gateway/src/test/resources/sql_gateway_rest_api_v3.snapshot
@@ -459,7 +459,7 @@
"cronExpression" : {
"type" : "string"
},
- "dynamicOptions" : {
+ "initConfig" : {
"type" : "object",
"additionalProperties" : {
"type" : "string"
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/workflow/CreatePeriodicRefreshWorkflow.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/workflow/CreatePeriodicRefreshWorkflow.java
index f999faab3b9..43ebf516916 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/workflow/CreatePeriodicRefreshWorkflow.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/workflow/CreatePeriodicRefreshWorkflow.java
@@ -36,7 +36,7 @@ public class CreatePeriodicRefreshWorkflow implements CreateRefreshWorkflow {
private final ObjectIdentifier materializedTableIdentifier;
private final String descriptionStatement;
private final String cronExpression;
- private final @Nullable Map<String, String> dynamicOptions;
+ private final @Nullable Map<String, String> initConfig;
private final @Nullable Map<String, String> executionConfig;
// The SQL Gateway rest endpoint url that used for execute refresh operation
@@ -46,13 +46,13 @@ public class CreatePeriodicRefreshWorkflow implements CreateRefreshWorkflow {
ObjectIdentifier materializedTableIdentifier,
String descriptionStatement,
String cronExpression,
- Map<String, String> dynamicOptions,
+ Map<String, String> initConfig,
Map<String, String> executionConfig,
String restEndpointUrl) {
this.materializedTableIdentifier = materializedTableIdentifier;
this.descriptionStatement = descriptionStatement;
this.cronExpression = cronExpression;
- this.dynamicOptions = dynamicOptions;
+ this.initConfig = initConfig;
this.executionConfig = executionConfig;
this.restEndpointUrl = restEndpointUrl;
}
@@ -70,8 +70,8 @@ public class CreatePeriodicRefreshWorkflow implements CreateRefreshWorkflow {
}
@Nullable
- public Map<String, String> getDynamicOptions() {
- return dynamicOptions;
+ public Map<String, String> getInitConfig() {
+ return initConfig;
}
@Nullable