You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampark.apache.org by be...@apache.org on 2025/07/04 03:57:12 UTC
(streampark) branch dev updated: [Improve] Cleanup SQL DDL (#4268)
This is an automated email from the ASF dual-hosted git repository.
benjobs pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/streampark.git
The following commit(s) were added to refs/heads/dev by this push:
new cdd959abd [Improve] Cleanup SQL DDL (#4268)
cdd959abd is described below
commit cdd959abd1a780bef83cf49c722c6191d1bd66f7
Author: benjobs <be...@apache.org>
AuthorDate: Fri Jul 4 11:57:07 2025 +0800
[Improve] Cleanup SQL DDL (#4268)
* [Imporve] Cleanup SQL DDL
* [Improve] mysql ddl improvement
* [Improve] update known-dependencies.txt
---
.../streampark/common/constants/Constants.java | 2 -
.../streampark/common/enums/FlinkJobType.java | 7 +-
.../streampark-console-service/pom.xml | 6 -
.../src/main/assembly/assembly.xml | 2 -
.../src/main/assembly/script/data/mysql-data.sql | 44 +-
.../src/main/assembly/script/data/pgsql-data.sql | 38 +-
.../main/assembly/script/schema/mysql-schema.sql | 18 -
.../main/assembly/script/schema/pgsql-schema.sql | 21 -
.../main/assembly/script/upgrade/mysql/2.2.0.sql | 86 +--
.../main/assembly/script/upgrade/pgsql/2.2.2.sql | 91 ++-
.../console/core/bean/FlinkCatalogParams.java | 198 -------
.../console/core/entity/FlinkApplication.java | 21 +-
.../console/core/entity/FlinkCatalog.java | 107 ----
.../console/core/entity/SparkApplication.java | 12 -
.../core/entity/SparkApplicationConfig.java | 23 -
.../console/core/enums/ChangeTypeEnum.java | 4 +-
.../console/core/mapper/FlinkCatalogMapper.java | 27 -
.../impl/FlinkApplicationActionServiceImpl.java | 24 +-
.../impl/FlinkApplicationBackupServiceImpl.java | 6 +-
.../FlinkApplicationBuildPipelineServiceImpl.java | 23 +-
.../impl/FlinkApplicationConfigServiceImpl.java | 2 +-
.../impl/FlinkApplicationManageServiceImpl.java | 21 +-
.../service/impl/FlinkSavepointServiceImpl.java | 2 +-
.../console/core/util/ServiceHelper.java | 24 -
.../src/main/resources/db/data-h2.sql | 42 +-
.../console/base/util/JacksonUtilTest.java | 44 --
.../streampark-console-webapp/package.json | 3 +-
.../src/enums/flinkEnum.ts | 1 -
.../src/locales/lang/en/flink/app.ts | 2 -
.../src/locales/lang/zh-CN/flink/app.ts | 2 -
.../src/views/flink/app/Add.vue | 515 +++++++++--------
.../src/views/flink/app/EditStreamPark.vue | 613 ++++++++++-----------
.../src/views/flink/app/View.vue | 447 ++++++++-------
.../src/views/flink/app/components/FlinkSql.vue | 330 +++++------
.../src/views/flink/app/hooks/useAppTableAction.ts | 1 -
.../flink/app/hooks/useCreateAndEditSchema.ts | 21 +-
.../src/views/flink/app/hooks/useCreateSchema.ts | 33 +-
streampark-flink/pom.xml | 1 -
.../streampark-flink-cdcclient/pom.xml | 140 -----
.../apache/streampark/flink/cdc/cli/CDCClient.java | 80 ---
.../streampark/flink/cdc/cli/CDCExecutor.java | 69 ---
.../streampark/flink/cdc/cli/package-info.java | 18 -
.../flink/client/bean/SubmitRequest.scala | 1 -
.../flink/client/impl/YarnApplicationClient.scala | 2 +-
.../flink/client/trait/FlinkClientTrait.scala | 2 +-
.../impl/FlinkYarnApplicationBuildPipeline.scala | 2 +-
tools/dependencies/known-dependencies.txt | 16 +-
47 files changed, 1087 insertions(+), 2107 deletions(-)
diff --git a/streampark-common/src/main/java/org/apache/streampark/common/constants/Constants.java b/streampark-common/src/main/java/org/apache/streampark/common/constants/Constants.java
index c8f27bcb5..1282b155c 100644
--- a/streampark-common/src/main/java/org/apache/streampark/common/constants/Constants.java
+++ b/streampark-common/src/main/java/org/apache/streampark/common/constants/Constants.java
@@ -53,8 +53,6 @@ public final class Constants {
public static final String STREAMPARK_SPARKSQL_CLIENT_CLASS = "org.apache.streampark.spark.cli.SqlClient";
- public static final String STREAMPARK_FLINKCDC_CLIENT_CLASS = "org.apache.streampark.flink.cdc.cli.CDCClient";
-
public static final String PYTHON_EXECUTABLE = "venv.zip/venv/bin/python3";
public static final String SINGLE_SLASH = "/";
diff --git a/streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkJobType.java b/streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkJobType.java
index 2c50978ef..1e9835e2a 100644
--- a/streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkJobType.java
+++ b/streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkJobType.java
@@ -43,12 +43,7 @@ public enum FlinkJobType {
/**
* Py flink Mode
*/
- PYFLINK("Python Flink", 3),
-
- /**
- * Flink CDC
- */
- FLINK_CDC("Flink CDC", 4);
+ PYFLINK("Python Flink", 3);
private final String name;
diff --git a/streampark-console/streampark-console-service/pom.xml b/streampark-console/streampark-console-service/pom.xml
index 66d8a6463..3b94cd0ca 100644
--- a/streampark-console/streampark-console-service/pom.xml
+++ b/streampark-console/streampark-console-service/pom.xml
@@ -411,12 +411,6 @@
<version>${project.version}</version>
</dependency>
- <dependency>
- <groupId>org.apache.streampark</groupId>
- <artifactId>streampark-flink-cdcclient_${scala.binary.version}</artifactId>
- <version>${project.version}</version>
- </dependency>
-
<dependency>
<groupId>com.fasterxml.jackson.module</groupId>
<artifactId>jackson-module-scala_${scala.binary.version}</artifactId>
diff --git a/streampark-console/streampark-console-service/src/main/assembly/assembly.xml b/streampark-console/streampark-console-service/src/main/assembly/assembly.xml
index 5d19b0ab2..37667ac70 100644
--- a/streampark-console/streampark-console-service/src/main/assembly/assembly.xml
+++ b/streampark-console/streampark-console-service/src/main/assembly/assembly.xml
@@ -32,7 +32,6 @@
<excludes>
<exclude>org.apache.streampark:streampark-flink-sqlclient_${scala.binary.version}</exclude>
<exclude>org.apache.streampark:streampark-spark-sqlclient_${scala.binary.version}</exclude>
- <exclude>org.apache.streampark:streampark-flink-cdcclient_${scala.binary.version}</exclude>
<exclude>javax.servlet:servlet-api</exclude>
</excludes>
</dependencySet>
@@ -41,7 +40,6 @@
<includes>
<include>org.apache.streampark:streampark-flink-sqlclient_${scala.binary.version}</include>
<include>org.apache.streampark:streampark-spark-sqlclient_${scala.binary.version}</include>
- <include>org.apache.streampark:streampark-flink-cdcclient_${scala.binary.version}</include>
</includes>
</dependencySet>
</dependencySets>
diff --git a/streampark-console/streampark-console-service/src/main/assembly/script/data/mysql-data.sql b/streampark-console/streampark-console-service/src/main/assembly/script/data/mysql-data.sql
index 1d94ec494..2e8733690 100644
--- a/streampark-console/streampark-console-service/src/main/assembly/script/data/mysql-data.sql
+++ b/streampark-console/streampark-console-service/src/main/assembly/script/data/mysql-data.sql
@@ -64,9 +64,9 @@ INSERT INTO `t_app` (`id`, `job_type`, `create_time`, `modify_time`) VALUES (100
-- Records of t_spark_app
-- ----------------------------
insert into `t_spark_app` (
- `id`, `team_id`, `job_type`, `app_type`, `app_name`, `deploy_mode`, `resource_from`, `main_class`,
- `yarn_queue`, `k8s_image_pull_policy`, `k8s_namespace`, `state`, `option_state`, `user_id`,
- `description`, `tracking`, `release`, `build`, `create_time`, `modify_time`, `tags`)
+ `id`, `team_id`, `job_type`, `app_type`, `app_name`, `deploy_mode`, `resource_from`, `main_class`,
+ `yarn_queue`, `k8s_image_pull_policy`, `k8s_namespace`, `state`, `option_state`, `user_id`,
+ `description`, `tracking`, `release`, `build`, `create_time`, `modify_time`, `tags`)
values (100001, 100000, 2, 4, 'Spark SQL Demo', 2, 2, 'org.apache.streampark.spark.cli.SqlClient', 'default', 0, 'default', 0, 0, 100000, 'Spark SQL Demo', 0, 1, 1, now(), now(), 'streampark,test');
-- ----------------------------
@@ -79,7 +79,6 @@ insert into `t_spark_effective` values (100000, 100001, 4, 100000, now());
-- ----------------------------
insert into `t_spark_sql` values (100000, 100001, 'eNq1jr0OgjAURnee4m4FY/oCTJVUg/KT9F7cK2kQiy2W+P6KMQ6yuDh9+YZzcjIlBUkgsSkkXCbv0N9Da0ifBgOx01cDSCqvdmsIpuu9e98kavA54EPH9ajbs+HTqIPl023gsyeN8gqlIsgrqhfmoygaiTEre2vYGliDgiW/IXvd2hdymIls0d87+5f6jxdlITOCFWxVXX5npg92MWtB', null, null, 1, 1, now());
-
-- ----------------------------
-- Records of t_menu
-- ----------------------------
@@ -190,25 +189,6 @@ insert into `t_menu` values (150502, 150500, 'member add', null, null, 'member:a
insert into `t_menu` values (150503, 150500, 'member update', null, null, 'member:update', null, '1', 1, null, now(), now());
insert into `t_menu` values (150504, 150500, 'member delete', null, null, 'member:delete', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150601, 150600, 'catalog view', null, null, 'catalog:view', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150602, 150600, 'catalog create', null, null, 'catalog:create', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150603, 150600, 'catalog update', null, null, 'catalog:update', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150604, 150600, 'catalog delete', null, null, 'catalog:delete', null, '1', 1, null, now(), now());
-
-insert into `t_menu` values (150605, 150600, 'database view', null, null, 'database:view', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150606, 150600, 'database create', null, null, 'database:create', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150607, 150600, 'database delete', null, null, 'database:delete', null, '1', 1, null, now(), now());
-
-insert into `t_menu` values (150608, 150600, 'table view', null, null, 'table:view', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150609, 150600, 'table create', null, null, 'table:create', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150610, 150600, 'table update', null, null, 'table:update', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150611, 150600, 'table view', null, null, 'table:column:add', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150612, 150600, 'table column list', null, null, 'table:column:list', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150613, 150600, 'table column drop', null, null, 'table:column:drop', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150614, 150600, 'table option add', null, null, 'option:add', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150615, 150600, 'table option remove', null, null, 'option:remove', null, '1', 1, null, now(), now());
-
--- -------
-- ----------------------------
-- Records of t_role
-- ----------------------------
@@ -324,27 +304,11 @@ insert into `t_role_menu` values (100102, 100002, 140401);
insert into `t_role_menu` values (100103, 100002, 140402);
insert into `t_role_menu` values (100104, 100002, 140403);
insert into `t_role_menu` values (100105, 100002, 150000);
+insert into `t_role_menu` values (100106, 100002, 150500);
insert into `t_role_menu` values (100107, 100002, 150501);
insert into `t_role_menu` values (100108, 100002, 150502);
insert into `t_role_menu` values (100109, 100002, 150503);
insert into `t_role_menu` values (100110, 100002, 150504);
-insert into `t_role_menu` values (100111, 100002, 150601);
-insert into `t_role_menu` values (100112, 100002, 150602);
-insert into `t_role_menu` values (100113, 100002, 150603);
-insert into `t_role_menu` values (100114, 100002, 150604);
-insert into `t_role_menu` values (100115, 100002, 150605);
-insert into `t_role_menu` values (100116, 100002, 150606);
-insert into `t_role_menu` values (100117, 100002, 150607);
-insert into `t_role_menu` values (100118, 100002, 150608);
-insert into `t_role_menu` values (100119, 100002, 150609);
-insert into `t_role_menu` values (100120, 100002, 150610);
-insert into `t_role_menu` values (100121, 100002, 150611);
-insert into `t_role_menu` values (100122, 100002, 150612);
-insert into `t_role_menu` values (100123, 100002, 150613);
-insert into `t_role_menu` values (100124, 100002, 150614);
-insert into `t_role_menu` values (100125, 100002, 150615);
-insert into `t_role_menu` values (100126, 100002, 150600);
-insert into `t_role_menu` values (100127, 100001, 150600);
-- ----------------------------
-- Records of t_setting
diff --git a/streampark-console/streampark-console-service/src/main/assembly/script/data/pgsql-data.sql b/streampark-console/streampark-console-service/src/main/assembly/script/data/pgsql-data.sql
index 46686f66b..a3f2d6c35 100644
--- a/streampark-console/streampark-console-service/src/main/assembly/script/data/pgsql-data.sql
+++ b/streampark-console/streampark-console-service/src/main/assembly/script/data/pgsql-data.sql
@@ -50,9 +50,9 @@ insert into "public"."t_flink_sql" values (100000, 100000, 'eNqlUUtPhDAQvu+vmFs1
-- Records of t_spark_app
-- ----------------------------
insert into "public"."t_spark_app" (
- "id", "team_id", "job_type", "app_type", "app_name", "deploy_mode", "resource_from", "main_class",
- "yarn_queue", "k8s_image_pull_policy", "k8s_namespace", "state", "option_state", "user_id",
- "description", "tracking", "release", "build", "create_time", "modify_time", "tags")
+ "id", "team_id", "job_type", "app_type", "app_name", "deploy_mode", "resource_from", "main_class",
+ "yarn_queue", "k8s_image_pull_policy", "k8s_namespace", "state", "option_state", "user_id",
+ "description", "tracking", "release", "build", "create_time", "modify_time", "tags")
values (100000, 100000, 2, 4, 'Spark SQL Demo', 2, 2, 'org.apache.streampark.spark.cli.SqlClient', 'default', 0, 'default', 0, 0, 100000, 'Spark SQL Demo', 0, 1, 1, now(), now(), 'streampark,test');
-- ----------------------------
@@ -153,22 +153,7 @@ insert into "public"."t_menu" values (130504, 130500, 'link delete', null, null,
insert into "public"."t_menu" values (130601, 130600, 'add yarn queue', null, null, 'yarnQueue:create', '', '1', '0', null, now(), now());
insert into "public"."t_menu" values (130602, 130600, 'edit yarn queue', null, null, 'yarnQueue:update', '', '1', '0', null, now(), now());
insert into "public"."t_menu" values (130603, 130600, 'delete yarn queue', null, null, 'yarnQueue:delete', '', '1', '0', null, now(), now());
-insert into "public"."t_menu" values (130701, 130700, 'catalog view', null, null, 'catalog:view', '', '1', '0', null, now(), now());
-insert into "public"."t_menu" values (130702, 130700, 'catalog create', null, null, 'catalog:create', '', '1', '0', null, now(), now());
-insert into "public"."t_menu" values (130703, 130700, 'catalog delete', null, null, 'catalog:delete', '', '1', '0', null, now(), now());
-insert into "public"."t_menu" values (130704, 130700, 'catalog update', null, null, 'catalog:update', '', '1', '0', null, now(), now());
-insert into "public"."t_menu" values (150605, 150600, 'database view', null, null, 'database:view', null, '1', 1, null, now(), now());
-insert into "public"."t_menu" values (150606, 150600, 'database create', null, null, 'database:create', null, '1', 1, null, now(), now());
-insert into "public"."t_menu" values (150607, 150600, 'database delete', null, null, 'database:delete', null, '1', 1, null, now(), now());
-insert into "public"."t_menu" values (150608, 150600, 'table view', null, null, 'table:view', null, '1', 1, null, now(), now());
-insert into "public"."t_menu" values (150609, 150600, 'table create', null, null, 'table:create', null, '1', 1, null, now(), now());
-insert into "public"."t_menu" values (150610, 150600, 'table update', null, null, 'table:update', null, '1', 1, null, now(), now());
-insert into "public"."t_menu" values (150611, 150600, 'table view', null, null, 'table:column:add', null, '1', 1, null, now(), now());
-insert into "public"."t_menu" values (150612, 150600, 'table column list', null, null, 'table:column:list', null, '1', 1, null, now(), now());
-insert into "public"."t_menu" values (150613, 150600, 'table column drop', null, null, 'table:column:drop', null, '1', 1, null, now(), now());
-insert into "public"."t_menu" values (150614, 150600, 'table option add', null, null, 'option:add', null, '1', 1, null, now(), now());
-insert into "public"."t_menu" values (150615, 150600, 'table option remove', null, null, 'option:remove', null, '1', 1, null, now(), now());
-- ----------------------------
-- Records of t_role
-- ----------------------------
@@ -273,23 +258,6 @@ insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130600);
insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130601);
insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130602);
insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130603);
-insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130701);
-insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130702);
-insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130703);
-insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130704);
-insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 150605);
-insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 150606);
-insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 150607);
-insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 150608);
-insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 150609);
-insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 150610);
-insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 150611);
-insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 150612);
-insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 150613);
-insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 150614);
-insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 150615);
-insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 150600);
-insert into "public"."t_role_menu" (role_id, menu_id) values (100001, 150600);
-- ----------------------------
-- Records of t_setting
diff --git a/streampark-console/streampark-console-service/src/main/assembly/script/schema/mysql-schema.sql b/streampark-console/streampark-console-service/src/main/assembly/script/schema/mysql-schema.sql
index e2c4a0e87..db29202ce 100644
--- a/streampark-console/streampark-console-service/src/main/assembly/script/schema/mysql-schema.sql
+++ b/streampark-console/streampark-console-service/src/main/assembly/script/schema/mysql-schema.sql
@@ -542,8 +542,6 @@ create table `t_yarn_queue` (
unique key `unq_team_id_queue_label` (`team_id`, `queue_label`) using btree
) engine = innodb default charset = utf8mb4 collate = utf8mb4_general_ci;
-
-
-- ----------------------------
-- Table of t_resource
-- ----------------------------
@@ -671,22 +669,6 @@ create table `t_spark_log` (
primary key (`id`) using btree
) engine=innodb auto_increment=100000 default charset=utf8mb4 collate=utf8mb4_general_ci;
--- ----------------------------
--- table structure for t_flink_catalog
--- ----------------------------
-drop table if exists `t_flink_catalog`;
-CREATE TABLE `t_flink_catalog` (
- `id` BIGINT AUTO_INCREMENT PRIMARY KEY,
- `team_id` BIGINT NOT NULL,
- `user_id` BIGINT DEFAULT NULL,
- `catalog_type` VARCHAR(255) NOT NULL,
- `catalog_name` VARCHAR(255) NOT NULL,
- `configuration` TEXT,
- `create_time` TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
- `update_time` TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
- UNIQUE KEY uniq_catalog_name (`catalog_name`)
-) ENGINE=InnoDB auto_increment=100000 default charset=utf8mb4 collate=utf8mb4_general_ci;
-
-- ----------------------------
-- table structure for t_spark_effective
-- ----------------------------
diff --git a/streampark-console/streampark-console-service/src/main/assembly/script/schema/pgsql-schema.sql b/streampark-console/streampark-console-service/src/main/assembly/script/schema/pgsql-schema.sql
index 8a5a88475..a638e7eeb 100644
--- a/streampark-console/streampark-console-service/src/main/assembly/script/schema/pgsql-schema.sql
+++ b/streampark-console/streampark-console-service/src/main/assembly/script/schema/pgsql-schema.sql
@@ -817,27 +817,6 @@ create table "public"."t_external_link" (
;
alter table "public"."t_external_link" add constraint "t_external_link_pkey" primary key ("id");
--- ----------------------------
--- table structure for t_flink_catalog
--- ----------------------------
-
-create sequence "public"."streampark_t_flink_catalog_id_seq"
- increment 1 start 10000 cache 1 minvalue 10000 maxvalue 9223372036854775807;
-
-CREATE TABLE "public"."t_flink_catalog" (
- "id" int8 not null default nextval('streampark_t_flink_catalog_id_seq'::regclass),
- "team_id" BIGINT NOT NULL,
- "user_id" BIGINT DEFAULT NULL,
- "catalog_type" VARCHAR(255) NOT NULL,
- "catalog_name" VARCHAR(255) NOT NULL,
- "configuration" TEXT,
- "create_time" TIMESTAMP WITHOUT TIME ZONE DEFAULT CURRENT_TIMESTAMP,
- "update_time" TIMESTAMP WITHOUT TIME ZONE DEFAULT CURRENT_TIMESTAMP,
- CONSTRAINT uniq_catalog_name UNIQUE (catalog_name)
-);
-alter table "public"."t_flink_catalog" add constraint "t_flink_catalog_pkey" primary key ("id");
-
-
-- ----------------------------
-- table structure for t_yarn_queue
-- ----------------------------
diff --git a/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/mysql/2.2.0.sql b/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/mysql/2.2.0.sql
index 755b7d731..ba95f45fa 100644
--- a/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/mysql/2.2.0.sql
+++ b/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/mysql/2.2.0.sql
@@ -34,37 +34,33 @@ primary key(`id`)
alter table `t_flink_app`
- modify column `id` not null;
+modify column `id` not null;
add column `k8s_name` varchar(63) collate utf8mb4_general_ci default null,
- -- modify_time change with duration #3188
- modify column `modify_time` datetime not null default current_timestamp comment 'modify time';
+-- modify_time change with duration #3188
+modify column `modify_time` datetime not null default current_timestamp comment 'modify time';
alter table t_app_backup rename to t_flink_app_backup;
alter table t_flink_log rename to t_app_log;
alter table `t_app_log`
- change column `yarn_app_id` `cluster_id` varchar(64) default null,
- change column `job_manager_url` `tracking_url` varchar(255) default null,
- change column `option_time` `create_time` datetime default null,
- add column `job_type` tinyint default null,
- add column `user_id` bigint default null comment 'operator user id';
+change column `yarn_app_id` `cluster_id` varchar(64) default null,
+change column `job_manager_url` `tracking_url` varchar(255) default null,
+add column `job_type` tinyint default null,
+add column `user_id` bigint default null comment 'operator user id';
alter table `t_flink_project`
- add column `salt` varchar(26) collate utf8mb4_general_ci default null comment 'password salt',
- modify column `password` varchar(512) collate utf8mb4_general_ci default null comment 'password';
-
-ALTER TABLE `t_flink_project`
- change column `branches` `refs` varchar(255) collate utf8mb4_general_ci default null;
+add column `salt` varchar(26) collate utf8mb4_general_ci default null comment 'password salt',
+modify column `password` varchar(512) collate utf8mb4_general_ci default null comment 'password';
alter table `t_flink_sql`
- add column `team_resource` varchar(64) default null;
+add column `team_resource` varchar(64) default null;
alter table `t_flink_cluster`
- add column `job_manager_url` varchar(150) default null comment 'url address of jobmanager' after `address`,
- add column `start_time` datetime default null comment 'start time',
- add column `end_time` datetime default null comment 'end time',
- add column `alert_id` bigint default null comment 'alert id';
+add column `job_manager_url` varchar(150) default null comment 'url address of jobmanager' after `address`,
+add column `start_time` datetime default null comment 'start time',
+add column `end_time` datetime default null comment 'end time',
+add column `alert_id` bigint default null comment 'alert id';
-- ----------------------------
-- Table of t_resource
@@ -108,58 +104,4 @@ insert into `t_role_menu` (role_id, menu_id) values (100002, 120403);
alter table `t_user` modify column `password` varchar(64) collate utf8mb4_general_ci default null comment 'password';
alter table `t_user` modify column `login_type` tinyint default 0 comment 'login type 0:password 1:ldap 2:sso';
--- ----------------------------
--- table structure for t_flink_catalog
--- ----------------------------
-drop table if exists `t_flink_catalog`;
-CREATE TABLE `t_flink_catalog` (
- `id` BIGINT AUTO_INCREMENT PRIMARY KEY,
- `team_id` BIGINT NOT NULL,
- `user_id` BIGINT DEFAULT NULL,
- `catalog_type` VARCHAR(255) NOT NULL,
- `catalog_name` VARCHAR(255) NOT NULL,
- `configuration` TEXT,
- `create_time` TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
- `update_time` TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
- UNIQUE KEY uniq_catalog_name (`catalog_name`)
-) ENGINE=InnoDB auto_increment=100000 default charset=utf8mb4 collate=utf8mb4_general_ci;
-
-insert into `t_menu` values (150601, 150600, 'catalog view', null, null, 'catalog:view', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150602, 150600, 'catalog create', null, null, 'catalog:create', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150603, 150600, 'catalog update', null, null, 'catalog:update', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150604, 150600, 'catalog delete', null, null, 'catalog:delete', null, '1', 1, null, now(), now());
-
-insert into `t_menu` values (150605, 150600, 'database view', null, null, 'database:view', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150606, 150600, 'database create', null, null, 'database:create', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150607, 150600, 'database delete', null, null, 'database:delete', null, '1', 1, null, now(), now());
-
-insert into `t_menu` values (150608, 150600, 'table view', null, null, 'table:view', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150609, 150600, 'table create', null, null, 'table:create', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150610, 150600, 'table update', null, null, 'table:update', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150611, 150600, 'table view', null, null, 'table:column:add', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150612, 150600, 'table column list', null, null, 'table:column:list', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150613, 150600, 'table column drop', null, null, 'table:column:drop', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150614, 150600, 'table option add', null, null, 'option:add', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150615, 150600, 'table option remove', null, null, 'option:remove', null, '1', 1, null, now(), now());
-
--- -------
-
-insert into `t_role_menu` values (100107, 100002, 150601);
-insert into `t_role_menu` values (100108, 100002, 150602);
-insert into `t_role_menu` values (100109, 100002, 150603);
-insert into `t_role_menu` values (100110, 100002, 150604);
-insert into `t_role_menu` values (100115, 100002, 150605);
-insert into `t_role_menu` values (100116, 100002, 150606);
-insert into `t_role_menu` values (100117, 100002, 150607);
-insert into `t_role_menu` values (100118, 100002, 150608);
-insert into `t_role_menu` values (100119, 100002, 150609);
-insert into `t_role_menu` values (100120, 100002, 150610);
-insert into `t_role_menu` values (100121, 100002, 150611);
-insert into `t_role_menu` values (100122, 100002, 150612);
-insert into `t_role_menu` values (100123, 100002, 150613);
-insert into `t_role_menu` values (100124, 100002, 150614);
-insert into `t_role_menu` values (100125, 100002, 150615);
-insert into `t_role_menu` values (100126, 100002, 150600);
-insert into `t_role_menu` values (100127, 100001, 150600);
-
set foreign_key_checks = 1;
diff --git a/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/pgsql/2.2.2.sql b/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/pgsql/2.2.2.sql
index d97733b5a..3ffc55059 100644
--- a/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/pgsql/2.2.2.sql
+++ b/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/pgsql/2.2.2.sql
@@ -15,27 +15,86 @@
* limitations under the License.
*/
-alter table "public"."t_flink_app"
- add column "k8s_name" varchar(63) collate "pg_catalog"."default";
+-- ----------------------------
+-- Table of t_resource
+-- ----------------------------
+create sequence "public"."streampark_t_resource_id_seq"
+ increment 1 start 10000 cache 1 minvalue 10000 maxvalue 9223372036854775807;
-alter table "public"."t_flink_log"
- add column "user_id" int8 collate "pg_catalog"."default";
+create table "public"."t_resource" (
+"id" int8 not null default nextval('streampark_t_resource_id_seq'::regclass),
+"resource_name" varchar(128) collate "pg_catalog"."default" not null,
+"resource_type" int4,
+"resource_path" varchar(255) default null,
+"resource" text collate "pg_catalog"."default",
+"engine_type" int4,
+"main_class" varchar(255) collate "pg_catalog"."default",
+"description" text collate "pg_catalog"."default" default null,
+"creator_id" int8 not null,
+"connector_required_options" text default null,
+"connector_optional_options" text default null,
+"team_id" int8 not null,
+"create_time" timestamp(6),
+"modify_time" timestamp(6)
+);
-comment on column "public"."t_flink_log"."user_id" is 'operator user id';
+comment on column "public"."t_resource"."id" is 'Resource id';
+comment on column "public"."t_resource"."resource_name" is 'Resource name';
+comment on column "public"."t_resource"."resource_type" is '0:app 1:common 2:connector 3:format 4:udf';
+comment on column "public"."t_resource"."resource" is 'resource content, including jars and poms';
+comment on column "public"."t_resource"."engine_type" is 'compute engine type, 0:apache flink 1:apache spark';
+comment on column "public"."t_resource"."main_class" is 'The program main class';
+comment on column "public"."t_resource"."description" is 'More detailed description of resource';
+comment on column "public"."t_resource"."creator_id" is 'user id of creator';
+comment on column "public"."t_resource"."team_id" is 'team id';
+comment on column "public"."t_resource"."create_time" is 'creation time';
+comment on column "public"."t_resource"."modify_time" is 'modify time';
-alter table "public"."t_flink_project"
- add column `salt` varchar(26) collate "pg_catalog"."default";
+alter table "public"."t_resource" add constraint "t_resource_pkey" primary key ("id");
+create index "un_team_dname_inx" on "public"."t_resource" using btree (
+ "team_id" "pg_catalog"."int8_ops" asc nulls last,
+ "resource_name" collate "pg_catalog"."default" "pg_catalog"."text_ops" asc nulls last
+ );
-comment on column "public"."t_flink_project"."salt" is 'password salt';
+alter table "public"."t_flink_sql"
+ add column "team_resource" varchar(64) default null;
-alter table "public"."t_flink_project"
- alter column `password` type varchar(512) collate "pg_catalog"."default";
+alter table "public"."t_flink_cluster"
+ add column "job_manager_url" varchar(150) collate "pg_catalog"."default",
+add column "start_time" timestamp(6) collate "pg_catalog"."default",
+add column "end_time" timestamp(6) collate "pg_catalog"."default",
+add column "alert_id" int8 collate "pg_catalog"."default";
-ALTER TABLE "public"."t_flink_project"
- ALTER COLUMN "branches" TYPE varchar(255) collate "pg_catalog"."default";
+insert into "public"."t_menu" values (120400, 120000, 'menu.resource', '/flink/resource', 'flink/resource/View', null, 'apartment', '0', '1', 3, now(), now());
+insert into "public"."t_menu" values (110401, 110400, 'add', null, null, 'token:add', null, '1', '1', null, now(), now());
+insert into "public"."t_menu" values (110402, 110400, 'delete', null, null, 'token:delete', null, '1', '1', null, now(), now());
+insert into "public"."t_menu" values (110403, 110400, 'view', null, null, 'token:view', null, '1', '1', null, now(), now());
-ALTER TABLE "public"."t_flink_project"
- rename column "branches" to "refs";
+insert into "public"."t_role_menu" (role_id, menu_id) values (100001, 120400);
+insert into "public"."t_role_menu" (role_id, menu_id) values (100001, 120401);
+insert into "public"."t_role_menu" (role_id, menu_id) values (100001, 120402);
+insert into "public"."t_role_menu" (role_id, menu_id) values (100001, 120403);
-ALTER TABLE "public"."t_app_log"
- rename column "option_time" to "create_time";
+insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 120400);
+insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 120401);
+insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 120402);
+insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 120403);
+
+-- add sso as login type
+alter table "public"."t_user" alter column "password" TYPE varchar(64) collate "pg_catalog"."default";
+comment on column "public"."t_user"."login_type" is 'login type 0:password 1:ldap 2:sso';
+
+insert into "public"."t_menu" values (120500, 130000, 'setting.flinkGateway', '/setting/FlinkGateway', 'setting/FlinkGateway/index', null, 'apartment', '0', '1', 3, now(), now());
+insert into "public"."t_menu" values (110501, 110500, 'add', null, null, 'gateway:add', null, '1', '1', null, now(), now());
+insert into "public"."t_menu" values (110502, 110500, 'update', null, null, 'gateway:update', null, '1', '1', null, now(), now());
+insert into "public"."t_menu" values (110503, 110500, 'delete', null, null, 'gateway:delete', null, '1', '1', null, now(), now());
+
+insert into "public"."t_role_menu" (role_id, menu_id) values (100001, 120500);
+insert into "public"."t_role_menu" (role_id, menu_id) values (100001, 120501);
+insert into "public"."t_role_menu" (role_id, menu_id) values (100001, 120502);
+insert into "public"."t_role_menu" (role_id, menu_id) values (100001, 120503);
+
+insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 120500);
+insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 120501);
+insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 120502);
+insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 120503);
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/bean/FlinkCatalogParams.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/bean/FlinkCatalogParams.java
deleted file mode 100644
index e1ea94d25..000000000
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/bean/FlinkCatalogParams.java
+++ /dev/null
@@ -1,198 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.streampark.console.core.bean;
-
-import org.apache.streampark.common.enums.CatalogType;
-import org.apache.streampark.console.base.util.JacksonUtils;
-import org.apache.streampark.console.core.entity.FlinkCatalog;
-
-import com.fasterxml.jackson.annotation.JsonProperty;
-import com.fasterxml.jackson.core.JsonProcessingException;
-import lombok.Getter;
-import lombok.Setter;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.BeanUtils;
-
-import javax.validation.constraints.NotBlank;
-
-import java.io.Serializable;
-import java.util.Date;
-
-@Getter
-@Setter
-@Slf4j
-public class FlinkCatalogParams implements Serializable {
-
- private Long id;
-
- private Long teamId;
-
- private String catalogName;
-
- private CatalogType catalogType;
-
- /** creator */
- private Long userId;
-
- private Date createTime;
-
- private Date updateTime;
-
- private FlinkJDBCCatalog flinkJDBCCatalog;
-
- private FlinkHiveCatalog flinkHiveCatalog;
-
- private FlinkPaimonCatalog flinkPaimonCatalog;
-
- private String customCatalogConfig;
-
- public static FlinkCatalogParams of(FlinkCatalog flinkCatalog) {
- if (flinkCatalog == null) {
- return null;
- }
- FlinkCatalogParams flinkCatalogParams = new FlinkCatalogParams();
- BeanUtils.copyProperties(flinkCatalog, flinkCatalogParams, "configuration");
- try {
- switch (flinkCatalog.getCatalogType()) {
- case MYSQL:
- case PGSQL:
- case ORACLE:
- case JDBC:
- flinkCatalogParams.setFlinkJDBCCatalog(
- JacksonUtils.read(flinkCatalog.getConfiguration(), FlinkJDBCCatalog.class));
- break;
- case HIVE:
- flinkCatalogParams.setFlinkHiveCatalog(
- JacksonUtils.read(flinkCatalog.getConfiguration(), FlinkHiveCatalog.class));
- break;
- case PAIMON:
- flinkCatalogParams.setFlinkPaimonCatalog(
- JacksonUtils.read(flinkCatalog.getConfiguration(), FlinkPaimonCatalog.class));
- break;
- case CUSTOM:
- flinkCatalogParams.setCustomCatalogConfig(flinkCatalog.getConfiguration());
- break;
- }
- } catch (JsonProcessingException e) {
- log.error("Flink catalog params json read failed", e);
- throw new RuntimeException("Flink catalog params json read failed");
- }
-
- return flinkCatalogParams;
- }
-
- @Getter
- @Setter
- public static class FlinkJDBCCatalog implements Serializable {
-
- @NotBlank
- private String type;
-
- @NotBlank
- @JsonProperty("default-database")
- private String defaultDatabase;
-
- @NotBlank
- private String username;
- @NotBlank
- private String password;
-
- @NotBlank
- @JsonProperty("base-url")
- private String baseUrl;
- }
-
- @Getter
- @Setter
- public static class FlinkHiveCatalog implements Serializable {
-
- @NotBlank
- private String type;
-
- @JsonProperty("hive-conf-dir")
- private String hiveConfDir;
-
- @JsonProperty("default-database")
- private String defaultDatabase;
-
- @JsonProperty("hive-version")
- private String hiveVersion;
-
- @JsonProperty("hadoop-conf-dir")
- private String hadoopConfDir;
-
- public String getType() {
- return type;
- }
-
- public void setType(String type) {
- this.type = type;
- }
-
- public String getHiveConfDir() {
- return hiveConfDir;
- }
-
- public void setHiveConfDir(String hiveConfDir) {
- this.hiveConfDir = hiveConfDir;
- }
-
- public String getDefaultDatabase() {
- return defaultDatabase;
- }
-
- public void setDefaultDatabase(String defaultDatabase) {
- this.defaultDatabase = defaultDatabase;
- }
-
- public String getHiveVersion() {
- return hiveVersion;
- }
-
- public void setHiveVersion(String hiveVersion) {
- this.hiveVersion = hiveVersion;
- }
-
- public String getHadoopConfDir() {
- return hadoopConfDir;
- }
-
- public void setHadoopConfDir(String hadoopConfDir) {
- this.hadoopConfDir = hadoopConfDir;
- }
- }
-
- @Getter
- @Setter
- public static class FlinkPaimonCatalog implements Serializable {
-
- @NotBlank
- private String type;
- @NotBlank
- private String warehouse;
- @NotBlank
- private String metastore; // hive filesystem
- private String uri;
-
- @JsonProperty("hive-conf-dir")
- private String hiveConfDir;
-
- @JsonProperty("hadoop-conf-dir")
- private String hadoopConfDir;
- }
-}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/FlinkApplication.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/FlinkApplication.java
index c7fa347fa..021612f1d 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/FlinkApplication.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/FlinkApplication.java
@@ -470,8 +470,6 @@ public class FlinkApplication extends BaseEntity {
switch (flinkJobType) {
case FLINK_SQL:
return Constants.STREAMPARK_FLINKSQL_CLIENT_CLASS;
- case FLINK_CDC:
- return Constants.STREAMPARK_FLINKCDC_CLIENT_CLASS;
case PYFLINK:
return Constants.PYTHON_FLINK_DRIVER_CLASS_NAME;
case FLINK_JAR:
@@ -505,35 +503,34 @@ public class FlinkApplication extends BaseEntity {
}
@JsonIgnore
- public boolean isJobTypeFlinkSqlOrCDC() {
- return FlinkJobType.FLINK_SQL.getMode().equals(this.getJobType()) ||
- FlinkJobType.FLINK_CDC.getMode().equals(this.getJobType());
+ public boolean isFlinkSql() {
+ return FlinkJobType.FLINK_SQL.getMode().equals(this.getJobType());
}
@JsonIgnore
- public boolean isJobTypeFlinkJar() {
+ public boolean isFlinkJar() {
return FlinkJobType.FLINK_JAR.getMode().equals(this.getJobType());
}
@JsonIgnore
- public boolean isJobTypeFlinkJarOrPyFlink() {
+ public boolean isFlinkJarOrPyFlink() {
return FlinkJobType.FLINK_JAR.getMode().equals(this.getJobType())
|| FlinkJobType.PYFLINK.getMode().equals(this.getJobType());
}
@JsonIgnore
- public boolean isResourceFromUpload() {
- return isJobTypeFlinkJarOrPyFlink()
+ public boolean isUploadResource() {
+ return isFlinkJarOrPyFlink()
&& ResourceFromEnum.UPLOAD.getValue().equals(this.getResourceFrom());
}
@JsonIgnore
- public boolean isResourceFromBuild() {
- return isJobTypeFlinkJarOrPyFlink()
+ public boolean isBuildResource() {
+ return isFlinkJarOrPyFlink()
&& ResourceFromEnum.BUILD.getValue().equals(this.getResourceFrom());
}
- public boolean isAppTypeStreamPark() {
+ public boolean isStreamParkType() {
return this.getAppType() == ApplicationType.STREAMPARK_FLINK.getType();
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/FlinkCatalog.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/FlinkCatalog.java
deleted file mode 100644
index 9fdc8342c..000000000
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/FlinkCatalog.java
+++ /dev/null
@@ -1,107 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.streampark.console.core.entity;
-
-import org.apache.streampark.common.enums.CatalogType;
-import org.apache.streampark.console.base.util.JacksonUtils;
-import org.apache.streampark.console.core.bean.FlinkCatalogParams;
-
-import com.baomidou.mybatisplus.annotation.FieldStrategy;
-import com.baomidou.mybatisplus.annotation.IdType;
-import com.baomidou.mybatisplus.annotation.TableField;
-import com.baomidou.mybatisplus.annotation.TableId;
-import com.baomidou.mybatisplus.annotation.TableName;
-import com.fasterxml.jackson.core.JsonProcessingException;
-import lombok.Getter;
-import lombok.Setter;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.BeanUtils;
-
-import java.io.Serializable;
-import java.util.Date;
-
-/** catalog store */
-@Getter
-@Setter
-@TableName("t_flink_catalog")
-@Slf4j
-public class FlinkCatalog implements Serializable {
-
- @TableId(type = IdType.AUTO)
- private Long id;
-
- private Long teamId;
-
- private String catalogName;
-
- private CatalogType catalogType;
-
- private String configuration;
-
- @TableField(updateStrategy = FieldStrategy.IGNORED)
- private Date createTime;
-
- private Date updateTime;
-
- /** creator */
- private Long userId;
-
- public static FlinkCatalog of(FlinkCatalogParams flinkCatalogParams) {
- if (flinkCatalogParams == null)
- return null;
- FlinkCatalog flinkCatalog = new FlinkCatalog();
-
- BeanUtils.copyProperties(
- flinkCatalogParams,
- flinkCatalog,
- "flinkJDBCCatalog",
- "flinkHiveCatalog",
- "flinkPaimonCatalog",
- "customCatalogConfig");
-
- if (null == flinkCatalogParams.getCatalogType()) {
- return flinkCatalog;
- }
- try {
- switch (flinkCatalogParams.getCatalogType()) {
- case MYSQL:
- case PGSQL:
- case ORACLE:
- case JDBC:
- flinkCatalog.setConfiguration(
- JacksonUtils.write(flinkCatalogParams.getFlinkJDBCCatalog()));
- break;
- case HIVE:
- flinkCatalog.setConfiguration(
- JacksonUtils.write(flinkCatalogParams.getFlinkHiveCatalog()));
- break;
- case PAIMON:
- flinkCatalog.setConfiguration(
- JacksonUtils.write(flinkCatalogParams.getFlinkPaimonCatalog()));
- break;
- case CUSTOM:
- flinkCatalog.setConfiguration(flinkCatalogParams.getCustomCatalogConfig());
- break;
- }
- } catch (JsonProcessingException e) {
- log.error("Flink catalog json read failed", e);
- throw new RuntimeException("Flink catalog json read failed");
- }
- return flinkCatalog;
- }
-}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/SparkApplication.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/SparkApplication.java
index f048c5712..01815b765 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/SparkApplication.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/SparkApplication.java
@@ -44,7 +44,6 @@ import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
-import com.baomidou.mybatisplus.core.toolkit.support.SFunction;
import com.fasterxml.jackson.annotation.JsonIgnore;
import lombok.Getter;
import lombok.Setter;
@@ -525,15 +524,4 @@ public class SparkApplication extends BaseEntity {
return Objects.hash(id);
}
- public static class SFunc {
-
- public static final SFunction<SparkApplication, Long> ID = SparkApplication::getId;
- public static final SFunction<SparkApplication, String> APP_ID = SparkApplication::getClusterId;
- public static final SFunction<SparkApplication, Date> START_TIME = SparkApplication::getStartTime;
- public static final SFunction<SparkApplication, Date> END_TIME = SparkApplication::getEndTime;
- public static final SFunction<SparkApplication, Long> DURATION = SparkApplication::getDuration;
- public static final SFunction<SparkApplication, Integer> STATE = SparkApplication::getState;
- public static final SFunction<SparkApplication, String> OPTIONS = SparkApplication::getOptions;
- public static final SFunction<SparkApplication, Integer> EXECUTION_MODE = SparkApplication::getDeployMode;
- }
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/SparkApplicationConfig.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/SparkApplicationConfig.java
index 371ad145e..a9dae69f7 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/SparkApplicationConfig.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/SparkApplicationConfig.java
@@ -18,8 +18,6 @@
package org.apache.streampark.console.core.entity;
import org.apache.streampark.common.util.DeflaterUtils;
-import org.apache.streampark.common.util.PropertiesUtils;
-import org.apache.streampark.console.core.enums.ConfigFileTypeEnum;
import com.baomidou.mybatisplus.annotation.FieldStrategy;
import com.baomidou.mybatisplus.annotation.IdType;
@@ -30,12 +28,8 @@ import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
-import javax.annotation.Nullable;
-
import java.util.Base64;
import java.util.Date;
-import java.util.HashMap;
-import java.util.Map;
@Getter
@Setter
@TableName("t_spark_config")
@@ -75,21 +69,4 @@ public class SparkApplicationConfig {
application.setFormat(this.format);
}
- @Nullable
- private Map<String, String> readConfig() {
- ConfigFileTypeEnum fileType = ConfigFileTypeEnum.of(this.format);
- if (fileType == null) {
- return null;
- }
- switch (fileType) {
- case YAML:
- return PropertiesUtils.fromYamlTextAsJava(DeflaterUtils.unzipString(this.content));
- case PROPERTIES:
- return PropertiesUtils.fromPropertiesTextAsJava(DeflaterUtils.unzipString(this.content));
- case HOCON:
- return PropertiesUtils.fromHoconTextAsJava(DeflaterUtils.unzipString(this.content));
- default:
- return new HashMap<>();
- }
- }
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/enums/ChangeTypeEnum.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/enums/ChangeTypeEnum.java
index a7032d344..699ff9356 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/enums/ChangeTypeEnum.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/enums/ChangeTypeEnum.java
@@ -32,10 +32,10 @@ public enum ChangeTypeEnum {
DEPENDENCY(1, "[DEPENDENCY], Dependency has changed"),
/** Represents a change in SQL. */
- SQL(2, "[SQL], Flink Sql is changed"),
+ SQL(2, "[SQL], SQL is changed"),
/** Represents a change in both dependencies and SQL. */
- ALL(3, "[ALL], Dependency and Flink Sql all changed"),
+ ALL(3, "[ALL], Dependency and SQL both changed"),
/** Represents a change in team resources. */
TEAM_RESOURCE(4, "[TEAM_RESOURCE], Team resource has changed");
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/mapper/FlinkCatalogMapper.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/mapper/FlinkCatalogMapper.java
deleted file mode 100644
index 94183ede7..000000000
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/mapper/FlinkCatalogMapper.java
+++ /dev/null
@@ -1,27 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.streampark.console.core.mapper;
-
-import org.apache.streampark.console.core.entity.FlinkCatalog;
-
-import com.baomidou.mybatisplus.core.mapper.BaseMapper;
-
-/** catalog mapper */
-public interface FlinkCatalogMapper extends BaseMapper<FlinkCatalog> {
-
-}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationActionServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationActionServiceImpl.java
index df86e4215..d79911d4c 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationActionServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationActionServiceImpl.java
@@ -205,7 +205,7 @@ public class FlinkApplicationActionServiceImpl
// 3) restore related status
LambdaUpdateWrapper<FlinkApplication> updateWrapper = Wrappers.lambdaUpdate();
updateWrapper.eq(FlinkApplication::getId, application.getId());
- if (application.isJobTypeFlinkSqlOrCDC()) {
+ if (application.isFlinkSql()) {
updateWrapper.set(FlinkApplication::getRelease, ReleaseStateEnum.FAILED.get());
} else {
updateWrapper.set(FlinkApplication::getRelease, ReleaseStateEnum.NEED_RELEASE.get());
@@ -421,7 +421,7 @@ public class FlinkApplicationActionServiceImpl
applicationManageService.toEffective(application);
Map<String, Object> extraParameter = new HashMap<>(0);
- if (application.isJobTypeFlinkSqlOrCDC()) {
+ if (application.isFlinkSql()) {
FlinkSql flinkSql = flinkSqlService.getEffective(application.getId(), true);
// Get the sql of the replaced placeholder
String realSql = variableService.replaceVariable(application.getTeamId(), flinkSql.getSql());
@@ -659,24 +659,6 @@ public class FlinkApplicationActionServiceImpl
}
break;
- case FLINK_CDC:
- log.info("the current job id: {}", application.getId());
- FlinkSql flinkCDC = flinkSqlService.getEffective(application.getId(), false);
- AssertUtils.notNull(flinkCDC);
- // 1) dist_userJar
- String cdcDistJar = ServiceHelper.getFlinkCDCClientJar(flinkEnv);
- // 2) appConfig
- appConf =
- applicationConfig == null
- ? null
- : String.format("yaml://%s", applicationConfig.getContent());
- // 3) client
- if (FlinkDeployMode.YARN_APPLICATION == deployModeEnum) {
- String clientPath = Workspace.remote().APP_CLIENT();
- flinkUserJar = String.format("%s/%s", clientPath, cdcDistJar);
- }
- break;
-
case PYFLINK:
Resource resource =
resourceService.findByResourceName(application.getTeamId(), application.getJar());
@@ -695,7 +677,7 @@ public class FlinkApplicationActionServiceImpl
break;
case FLINK_JAR:
- if (application.isResourceFromUpload()) {
+ if (application.isUploadResource()) {
appConf =
String.format(
"json://{\"%s\":\"%s\"}",
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationBackupServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationBackupServiceImpl.java
index 2b5fc93dd..9a84d354f 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationBackupServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationBackupServiceImpl.java
@@ -89,7 +89,7 @@ public class FlinkApplicationBackupServiceImpl
// If necessary, perform the backup first
if (bakParam.isBackup()) {
application.setBackUpDescription(bakParam.getDescription());
- if (application.isJobTypeFlinkSqlOrCDC()) {
+ if (application.isFlinkSql()) {
FlinkSql flinkSql = flinkSqlService.getEffective(application.getId(), false);
backup(application, flinkSql);
} else {
@@ -107,7 +107,7 @@ public class FlinkApplicationBackupServiceImpl
effectiveService.saveOrUpdate(
bakParam.getAppId(), EffectiveTypeEnum.CONFIG, bakParam.getId());
// if flink sql task, will be rollback sql and dependencies
- if (application.isJobTypeFlinkSqlOrCDC()) {
+ if (application.isFlinkSql()) {
effectiveService.saveOrUpdate(
bakParam.getAppId(), EffectiveTypeEnum.FLINKSQL, bakParam.getSqlId());
}
@@ -190,7 +190,7 @@ public class FlinkApplicationBackupServiceImpl
@Override
public void backup(FlinkApplication appParam, FlinkSql flinkSqlParam) {
// basic configuration file backup
- String appHome = (appParam.isJobTypeFlinkJar() && appParam.isResourceFromBuild())
+ String appHome = (appParam.isFlinkJar() && appParam.isBuildResource())
? appParam.getDistHome()
: appParam.getAppHome();
FsOperator fsOperator = appParam.getFsOperator();
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationBuildPipelineServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationBuildPipelineServiceImpl.java
index 4fd369c51..bd7e24647 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationBuildPipelineServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationBuildPipelineServiceImpl.java
@@ -192,7 +192,7 @@ public class FlinkApplicationBuildPipelineServiceImpl
return true;
}
// rollback
- if (app.isNeedRollback() && app.isJobTypeFlinkSqlOrCDC()) {
+ if (app.isNeedRollback() && app.isFlinkSql()) {
flinkSqlService.rollback(app);
}
@@ -200,7 +200,7 @@ public class FlinkApplicationBuildPipelineServiceImpl
FlinkSql newFlinkSql = flinkSqlService.getCandidate(app.getId(), CandidateTypeEnum.NEW);
FlinkSql effectiveFlinkSql = flinkSqlService.getEffective(app.getId(), false);
FlinkJobType jobType = app.getJobTypeEnum();
- if (jobType == FlinkJobType.FLINK_SQL || jobType == FlinkJobType.PYFLINK || jobType == FlinkJobType.FLINK_CDC) {
+ if (jobType == FlinkJobType.FLINK_SQL || jobType == FlinkJobType.PYFLINK) {
FlinkSql flinkSql = newFlinkSql == null ? effectiveFlinkSql : newFlinkSql;
AssertUtils.notNull(flinkSql);
app.setDependency(flinkSql.getDependency());
@@ -236,12 +236,12 @@ public class FlinkApplicationBuildPipelineServiceImpl
// 2) some preparatory work
String appUploads = app.getWorkspace().APP_UPLOADS();
- if (app.isJobTypeFlinkJarOrPyFlink()) {
+ if (app.isFlinkJarOrPyFlink()) {
// flinkJar upload jar to appHome...
String appHome = app.getAppHome();
FsOperator fsOperator = app.getFsOperator();
fsOperator.delete(appHome);
- if (app.isResourceFromUpload()) {
+ if (app.isUploadResource()) {
String uploadJar = appUploads.concat("/").concat(app.getJar());
File localJar = new File(
String.format(
@@ -325,10 +325,10 @@ public class FlinkApplicationBuildPipelineServiceImpl
// If the current task is not running, or the task has just been added, directly
// set
// the candidate version to the official version
- if (app.isJobTypeFlinkSqlOrCDC()) {
+ if (app.isFlinkSql()) {
applicationManageService.toEffective(app);
} else {
- if (app.isAppTypeStreamPark()) {
+ if (app.isStreamParkType()) {
FlinkApplicationConfig config =
applicationConfigService.getLatest(app.getId());
if (config != null) {
@@ -341,7 +341,7 @@ public class FlinkApplicationBuildPipelineServiceImpl
}
// backup.
if (!app.isNeedRollback()) {
- if (app.isJobTypeFlinkSqlOrCDC() && newFlinkSql != null) {
+ if (app.isFlinkSql() && newFlinkSql != null) {
backUpService.backup(app, newFlinkSql);
} else {
backUpService.backup(app, null);
@@ -575,7 +575,7 @@ public class FlinkApplicationBuildPipelineServiceImpl
app.getLocalAppHome(),
mainClass,
flinkUserJar,
- app.isJobTypeFlinkJar(),
+ app.isFlinkJar(),
app.getDeployModeEnum(),
app.getJobTypeEnum(),
flinkEnv.getFlinkVersion(),
@@ -608,13 +608,6 @@ public class FlinkApplicationBuildPipelineServiceImpl
return String.format("%s/%s", clientPath, sqlDistJar);
}
return Workspace.local().APP_CLIENT().concat("/").concat(sqlDistJar);
- case FLINK_CDC:
- String cdcDistJar = ServiceHelper.getFlinkCDCClientJar(flinkEnv);
- if (app.getDeployModeEnum() == FlinkDeployMode.YARN_APPLICATION) {
- String clientPath = Workspace.remote().APP_CLIENT();
- return String.format("%s/%s", clientPath, cdcDistJar);
- }
- return Workspace.local().APP_CLIENT().concat("/").concat(cdcDistJar);
default:
throw new UnsupportedOperationException(
"[StreamPark] unsupported JobType: " + app.getJobTypeEnum());
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationConfigServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationConfigServiceImpl.java
index a6447f9ec..21d84e5f4 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationConfigServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationConfigServiceImpl.java
@@ -99,7 +99,7 @@ public class FlinkApplicationConfigServiceImpl
public synchronized void update(FlinkApplication appParam, Boolean latest) {
// flink sql job
FlinkApplicationConfig latestConfig = getLatest(appParam.getId());
- if (appParam.isJobTypeFlinkSqlOrCDC()) {
+ if (appParam.isFlinkSql()) {
updateForFlinkSqlJob(appParam, latest, latestConfig);
} else {
updateForNonFlinkSqlJob(appParam, latest, latestConfig);
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationManageServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationManageServiceImpl.java
index 158296f86..ddbb27cf1 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationManageServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationManageServiceImpl.java
@@ -159,7 +159,7 @@ public class FlinkApplicationManageServiceImpl extends ServiceImpl<FlinkApplicat
if (config != null) {
this.configService.toEffective(appParam.getId(), config.getId());
}
- if (appParam.isJobTypeFlinkSqlOrCDC()) {
+ if (appParam.isFlinkSql()) {
FlinkSql flinkSql = flinkSqlService.getCandidate(appParam.getId(), null);
if (flinkSql != null) {
flinkSqlService.toEffective(appParam.getId(), flinkSql.getId());
@@ -345,7 +345,7 @@ public class FlinkApplicationManageServiceImpl extends ServiceImpl<FlinkApplicat
appParam.getTeamId());
appParam.doSetHotParams();
- if (appParam.isResourceFromUpload()) {
+ if (appParam.isUploadResource()) {
String jarPath = String.format(
"%s/%d/%s", Workspace.local().APP_UPLOADS(), appParam.getTeamId(), appParam.getJar());
if (!new File(jarPath).exists()) {
@@ -360,8 +360,7 @@ public class FlinkApplicationManageServiceImpl extends ServiceImpl<FlinkApplicat
boolean saveSuccess = save(appParam);
if (saveSuccess) {
FlinkJobType jobType = appParam.getJobTypeEnum();
- if (jobType == FlinkJobType.FLINK_SQL || jobType == FlinkJobType.PYFLINK
- || jobType == FlinkJobType.FLINK_CDC) {
+ if (jobType == FlinkJobType.FLINK_SQL || jobType == FlinkJobType.PYFLINK) {
FlinkSql flinkSql = new FlinkSql(appParam);
flinkSqlService.create(flinkSql);
}
@@ -442,7 +441,7 @@ public class FlinkApplicationManageServiceImpl extends ServiceImpl<FlinkApplicat
boolean saved = save(newApp);
if (saved) {
- if (newApp.isJobTypeFlinkSqlOrCDC()) {
+ if (newApp.isFlinkSql()) {
FlinkSql copyFlinkSql = flinkSqlService.getLatestFlinkSql(appParam.getId(), true);
newApp.setFlinkSql(copyFlinkSql.getSql());
newApp.setDependency(copyFlinkSql.getDependency());
@@ -493,7 +492,7 @@ public class FlinkApplicationManageServiceImpl extends ServiceImpl<FlinkApplicat
application.setRelease(ReleaseStateEnum.NEED_RELEASE.get());
// 1) jar job jar file changed
- if (application.isResourceFromUpload()) {
+ if (application.isUploadResource()) {
if (!Objects.equals(application.getJar(), appParam.getJar())) {
application.setBuild(true);
} else {
@@ -577,12 +576,12 @@ public class FlinkApplicationManageServiceImpl extends ServiceImpl<FlinkApplicat
}
// Flink Sql job...
- if (application.isJobTypeFlinkSqlOrCDC()) {
+ if (application.isFlinkSql()) {
updateFlinkSqlJob(application, appParam);
return true;
}
- if (application.isAppTypeStreamPark()) {
+ if (application.isStreamParkType()) {
configService.update(appParam, application.isRunning());
} else {
application.setJar(appParam.getJar());
@@ -715,7 +714,7 @@ public class FlinkApplicationManageServiceImpl extends ServiceImpl<FlinkApplicat
update.update();
// backup
- if (appParam.isJobTypeFlinkSqlOrCDC()) {
+ if (appParam.isFlinkSql()) {
FlinkSql newFlinkSql = flinkSqlService.getCandidate(appParam.getId(), CandidateTypeEnum.NEW);
if (!appParam.isNeedRollback() && newFlinkSql != null) {
backUpService.backup(appParam, newFlinkSql);
@@ -746,7 +745,7 @@ public class FlinkApplicationManageServiceImpl extends ServiceImpl<FlinkApplicat
if (config != null) {
config.setToApplication(application);
}
- if (application.isJobTypeFlinkSqlOrCDC()) {
+ if (application.isFlinkSql()) {
FlinkSql flinkSql = flinkSqlService.getEffective(application.getId(), true);
if (flinkSql == null) {
flinkSql = flinkSqlService.getCandidate(application.getId(), CandidateTypeEnum.NEW);
@@ -754,7 +753,7 @@ public class FlinkApplicationManageServiceImpl extends ServiceImpl<FlinkApplicat
}
flinkSql.setToApplication(application);
} else {
- if (application.isResourceFromBuild()) {
+ if (application.isBuildResource()) {
String path = this.projectService.getAppConfPath(application.getProjectId(), application.getModule());
application.setConfPath(path);
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/FlinkSavepointServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/FlinkSavepointServiceImpl.java
index 5420c8167..8f2ab73cb 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/FlinkSavepointServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/FlinkSavepointServiceImpl.java
@@ -342,7 +342,7 @@ public class FlinkSavepointServiceImpl extends ServiceImpl<FlinkSavepointMapper,
@VisibleForTesting
@Nullable
public String getSavepointFromConfig(FlinkApplication application) {
- if (!application.isAppTypeStreamPark() && !application.isJobTypeFlinkSqlOrCDC()) {
+ if (!application.isStreamParkType() && !application.isFlinkSql()) {
return null;
}
FlinkApplicationConfig applicationConfig = configService.getEffective(application.getId());
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/util/ServiceHelper.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/util/ServiceHelper.java
index 02a6e8ab3..252b7364e 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/util/ServiceHelper.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/util/ServiceHelper.java
@@ -39,8 +39,6 @@ public class ServiceHelper {
private static String flinkSqlClientJar = null;
- private static String flinkCDCClientJar = null;
-
private static String sparkSqlClientJar = null;
public static User getLoginUser() {
@@ -60,28 +58,6 @@ public class ServiceHelper {
return null;
}
- public static String getFlinkCDCClientJar(FlinkEnv flinkEnv) {
- if (flinkCDCClientJar == null) {
- File localClient = WebUtils.getAppClientDir();
- ApiAlertException.throwIfFalse(localClient.exists(),
- "[StreamPark]" + localClient + " no exists. please check.");
- String regex = String.format("streampark-flink-cdcclient_%s-.*\\.jar", flinkEnv.getScalaVersion());
-
- List<String> jars = Arrays.stream(Objects.requireNonNull(localClient.list()))
- .filter(x -> x.matches(regex))
- .collect(Collectors.toList());
- ApiAlertException.throwIfTrue(
- jars.isEmpty(),
- "[StreamPark] can't found streampark-flink-cdcclient jar in " + localClient);
-
- ApiAlertException.throwIfTrue(
- jars.size() > 1,
- "[StreamPark] found multiple streampark-flink-cdclient jar in " + localClient);
- flinkCDCClientJar = jars.get(0);
- }
- return flinkCDCClientJar;
- }
-
public static String getFlinkSqlClientJar(FlinkEnv flinkEnv) {
if (flinkSqlClientJar == null) {
File localClient = WebUtils.getAppClientDir();
diff --git a/streampark-console/streampark-console-service/src/main/resources/db/data-h2.sql b/streampark-console/streampark-console-service/src/main/resources/db/data-h2.sql
index d76d1ede5..7dac02fb6 100644
--- a/streampark-console/streampark-console-service/src/main/resources/db/data-h2.sql
+++ b/streampark-console/streampark-console-service/src/main/resources/db/data-h2.sql
@@ -62,9 +62,9 @@ VALUES (100001, 2, now(), now());
-- Records of t_spark_app
-- ----------------------------
insert into `t_spark_app` (
- `id`, `team_id`, `job_type`, `app_type`, `app_name`, `deploy_mode`, `resource_from`, `main_class`,
- `yarn_queue`, `k8s_image_pull_policy`, `k8s_namespace`, `state`, `option_state`, `user_id`,
- `description`, `tracking`, `release`, `build`, `create_time`, `modify_time`, `tags`)
+ `id`, `team_id`, `job_type`, `app_type`, `app_name`, `deploy_mode`, `resource_from`, `main_class`,
+ `yarn_queue`, `k8s_image_pull_policy`, `k8s_namespace`, `state`, `option_state`, `user_id`,
+ `description`, `tracking`, `release`, `build`, `create_time`, `modify_time`, `tags`)
values (100001, 100000, 2, 4, 'Spark SQL Demo', 2, 2, 'org.apache.streampark.spark.cli.SqlClient', 'default', 0, 'default', 0, 0, 100000, 'Spark SQL Demo', 0, 1, 1, now(), now(), 'streampark,test');
-- ----------------------------
@@ -187,24 +187,6 @@ insert into `t_menu` values (150502, 150500, 'member add', null, null, 'member:a
insert into `t_menu` values (150503, 150500, 'member update', null, null, 'member:update', null, '1', 1, null, now(), now());
insert into `t_menu` values (150504, 150500, 'member delete', null, null, 'member:delete', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150601, 150600, 'catalog view', null, null, 'catalog:view', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150602, 150600, 'catalog create', null, null, 'catalog:create', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150603, 150600, 'catalog update', null, null, 'catalog:update', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150604, 150600, 'catalog delete', null, null, 'catalog:delete', null, '1', 1, null, now(), now());
-
-insert into `t_menu` values (150605, 150600, 'database view', null, null, 'database:view', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150606, 150600, 'database create', null, null, 'database:create', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150607, 150600, 'database delete', null, null, 'database:delete', null, '1', 1, null, now(), now());
-
-insert into `t_menu` values (150608, 150600, 'table view', null, null, 'table:view', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150609, 150600, 'table create', null, null, 'table:create', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150610, 150600, 'table update', null, null, 'table:update', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150611, 150600, 'table view', null, null, 'table:column:add', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150612, 150600, 'table column list', null, null, 'table:column:list', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150613, 150600, 'table column drop', null, null, 'table:column:drop', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150614, 150600, 'table option add', null, null, 'option:add', null, '1', 1, null, now(), now());
-insert into `t_menu` values (150615, 150600, 'table option remove', null, null, 'option:remove', null, '1', 1, null, now(), now());
-
-- ----------------------------
-- Records of t_role
-- ----------------------------
@@ -325,24 +307,6 @@ insert into `t_role_menu` values (100107, 100002, 150501);
insert into `t_role_menu` values (100108, 100002, 150502);
insert into `t_role_menu` values (100109, 100002, 150503);
insert into `t_role_menu` values (100110, 100002, 150504);
-insert into `t_role_menu` values (100111, 100002, 150601);
-insert into `t_role_menu` values (100112, 100002, 150602);
-insert into `t_role_menu` values (100113, 100002, 150603);
-insert into `t_role_menu` values (100114, 100002, 150604);
-insert into `t_role_menu` values (100115, 100002, 150605);
-insert into `t_role_menu` values (100116, 100002, 150606);
-insert into `t_role_menu` values (100117, 100002, 150607);
-insert into `t_role_menu` values (100118, 100002, 150608);
-insert into `t_role_menu` values (100119, 100002, 150609);
-insert into `t_role_menu` values (100120, 100002, 150610);
-insert into `t_role_menu` values (100121, 100002, 150611);
-insert into `t_role_menu` values (100122, 100002, 150612);
-insert into `t_role_menu` values (100123, 100002, 150613);
-insert into `t_role_menu` values (100124, 100002, 150614);
-insert into `t_role_menu` values (100125, 100002, 150615);
-insert into `t_role_menu` values (100126, 100002, 150600);
-insert into `t_role_menu` values (100127, 100001, 150600);
-
-- ----------------------------
-- Records of t_setting
diff --git a/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/base/util/JacksonUtilTest.java b/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/base/util/JacksonUtilTest.java
deleted file mode 100644
index 470b76f86..000000000
--- a/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/base/util/JacksonUtilTest.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.streampark.console.base.util;
-
-import org.apache.streampark.console.core.bean.FlinkCatalogParams;
-
-import com.fasterxml.jackson.core.JsonProcessingException;
-import lombok.extern.slf4j.Slf4j;
-import org.junit.jupiter.api.Test;
-
-import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
-
-@Slf4j
-public class JacksonUtilTest {
-
- @Test
- public void flinkCatalogConvertTest() throws JsonProcessingException {
- FlinkCatalogParams.FlinkJDBCCatalog flinkJDBCCatalog =
- new FlinkCatalogParams.FlinkJDBCCatalog();
- flinkJDBCCatalog.setType("jdbc");
- flinkJDBCCatalog.setDefaultDatabase("streampark");
- flinkJDBCCatalog.setPassword("12345678");
- flinkJDBCCatalog.setUsername("root");
- flinkJDBCCatalog.setBaseUrl("jdbc:mysql://127.0.0.1:3306");
- String flinkJdbcCatalog = JacksonUtils.write(flinkJDBCCatalog);
- log.info("catalogString is {}", flinkJdbcCatalog);
- assertThat(flinkJdbcCatalog.contains("base-url")).isTrue();
- }
-}
diff --git a/streampark-console/streampark-console-webapp/package.json b/streampark-console/streampark-console-webapp/package.json
index d7cfeb0fd..d28736ad4 100644
--- a/streampark-console/streampark-console-webapp/package.json
+++ b/streampark-console/streampark-console-webapp/package.json
@@ -65,8 +65,7 @@
"vue": "^3.3.4",
"vue-i18n": "^9.2.2",
"vue-router": "^4.2.4",
- "vue-types": "^5.1.0",
- "js-yaml": "^4.1.0"
+ "vue-types": "^5.1.0"
},
"devDependencies": {
"@iconify/json": "^2.2.89",
diff --git a/streampark-console/streampark-console-webapp/src/enums/flinkEnum.ts b/streampark-console/streampark-console-webapp/src/enums/flinkEnum.ts
index d7e72ff6c..7c078d5e5 100644
--- a/streampark-console/streampark-console-webapp/src/enums/flinkEnum.ts
+++ b/streampark-console/streampark-console-webapp/src/enums/flinkEnum.ts
@@ -151,7 +151,6 @@ export enum JobTypeEnum {
JAR = 1,
SQL = 2,
PYFLINK = 3,
- CDC = 4,
}
export enum ConfigTypeEnum {
diff --git a/streampark-console/streampark-console-webapp/src/locales/lang/en/flink/app.ts b/streampark-console/streampark-console-webapp/src/locales/lang/en/flink/app.ts
index 7840b7da4..ab0bc56b8 100644
--- a/streampark-console/streampark-console-webapp/src/locales/lang/en/flink/app.ts
+++ b/streampark-console/streampark-console-webapp/src/locales/lang/en/flink/app.ts
@@ -205,10 +205,8 @@ export default {
editStreamPark: {
success: 'update successful',
flinkSqlRequired: 'Flink Sql is required',
- yamlRequired: 'Yaml is required',
appidCheck: 'appid can not be empty',
sqlCheck: 'SQL check error',
- yamlCheck: 'Yaml check error',
},
operation: {
edit: 'Edit Job',
diff --git a/streampark-console/streampark-console-webapp/src/locales/lang/zh-CN/flink/app.ts b/streampark-console/streampark-console-webapp/src/locales/lang/zh-CN/flink/app.ts
index 5ba959233..11a8b417a 100644
--- a/streampark-console/streampark-console-webapp/src/locales/lang/zh-CN/flink/app.ts
+++ b/streampark-console/streampark-console-webapp/src/locales/lang/zh-CN/flink/app.ts
@@ -195,8 +195,6 @@ export default {
flinkSqlRequired: 'Flink Sql 为必填项',
appidCheck: 'appid 不能为空',
sqlCheck: 'SQL 检查错误',
- yamlRequired: 'Yaml 不能为空',
- yamlCheck: 'Yaml 检查错误',
},
operation: {
edit: '编辑作业',
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/Add.vue b/streampark-console/streampark-console-webapp/src/views/flink/app/Add.vue
index 74e334961..6927e71ba 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/Add.vue
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/Add.vue
@@ -15,304 +15,296 @@
limitations under the License.
-->
<script lang="ts">
- export default {
- name: 'AppCreate',
- };
+export default {
+ name: 'AppCreate',
+};
</script>
<script setup lang="ts" name="AppCreate">
- import { useGo } from '/@/hooks/web/usePage';
- import ProgramArgs from './components/ProgramArgs.vue';
- import { Switch } from 'ant-design-vue';
- import { onMounted, reactive, ref, unref } from 'vue';
- import { PageWrapper } from '/@/components/Page';
- import { createAsyncComponent } from '/@/utils/factory/createAsyncComponent';
+import { useGo } from '/@/hooks/web/usePage';
+import ProgramArgs from './components/ProgramArgs.vue';
+import { Switch } from 'ant-design-vue';
+import { onMounted, reactive, ref, unref } from 'vue';
+import { PageWrapper } from '/@/components/Page';
+import { createAsyncComponent } from '/@/utils/factory/createAsyncComponent';
- import { BasicForm, useForm } from '/@/components/Form';
- import { SettingTwoTone } from '@ant-design/icons-vue';
- import { useDrawer } from '/@/components/Drawer';
- import Mergely from './components/Mergely.vue';
- import { handleConfTemplate } from '/@/api/flink/config';
- import { fetchAppConf, fetchCreate } from '/@/api/flink/app';
- import options from './data/option';
- import { useCreateSchema } from './hooks/useCreateSchema';
- import { getAppConfType, handleSubmitParams } from './utils';
- import { useMessage } from '/@/hooks/web/useMessage';
- import { createLocalStorage } from '/@/utils/cache';
- import { buildUUID } from '/@/utils/uuid';
- import { useI18n } from '/@/hooks/web/useI18n';
- import VariableReview from './components/VariableReview.vue';
- import PomTemplateTab from './components/PodTemplate/PomTemplateTab.vue';
- import UseSysHadoopConf from './components/UseSysHadoopConf.vue';
- import { CreateParams } from '/@/api/flink/app.type';
- import { decodeByBase64, encryptByBase64 } from '/@/utils/cipher';
- import {
- AppTypeEnum,
- ClusterStateEnum,
- DeployMode,
- JobTypeEnum,
- ResourceFromEnum,
- } from '/@/enums/flinkEnum';
+import { BasicForm, useForm } from '/@/components/Form';
+import { SettingTwoTone } from '@ant-design/icons-vue';
+import { useDrawer } from '/@/components/Drawer';
+import Mergely from './components/Mergely.vue';
+import { handleConfTemplate } from '/@/api/flink/config';
+import { fetchAppConf, fetchCreate } from '/@/api/flink/app';
+import options from './data/option';
+import { useCreateSchema } from './hooks/useCreateSchema';
+import { getAppConfType, handleSubmitParams } from './utils';
+import { useMessage } from '/@/hooks/web/useMessage';
+import { createLocalStorage } from '/@/utils/cache';
+import { buildUUID } from '/@/utils/uuid';
+import { useI18n } from '/@/hooks/web/useI18n';
+import VariableReview from './components/VariableReview.vue';
+import PomTemplateTab from './components/PodTemplate/PomTemplateTab.vue';
+import UseSysHadoopConf from './components/UseSysHadoopConf.vue';
+import { CreateParams } from '/@/api/flink/app.type';
+import { decodeByBase64, encryptByBase64 } from '/@/utils/cipher';
+import {
+ AppTypeEnum,
+ ClusterStateEnum,
+ DeployMode,
+ JobTypeEnum,
+ ResourceFromEnum,
+} from '/@/enums/flinkEnum';
- const FlinkSqlEditor = createAsyncComponent(() => import('./components/FlinkSql.vue'), {
- loading: true,
- });
- const Dependency = createAsyncComponent(() => import('./components/Dependency.vue'), {
- loading: true,
- });
+const FlinkSqlEditor = createAsyncComponent(() => import('./components/FlinkSql.vue'), {
+ loading: true,
+});
+const Dependency = createAsyncComponent(() => import('./components/Dependency.vue'), {
+ loading: true,
+});
- const go = useGo();
- const flinkSql = ref();
- const dependencyRef = ref();
- const submitLoading = ref(false);
+const go = useGo();
+const flinkSql = ref();
+const dependencyRef = ref();
+const submitLoading = ref(false);
- const { t } = useI18n();
- const { createMessage } = useMessage();
- const optionsKeyMapping = new Map();
- const ls = createLocalStorage();
- options.forEach((item) => {
- optionsKeyMapping.set(item.key, item);
- });
+const { t } = useI18n();
+const { createMessage } = useMessage();
+const optionsKeyMapping = new Map();
+const ls = createLocalStorage();
+options.forEach((item) => {
+ optionsKeyMapping.set(item.key, item);
+});
- const k8sTemplate = reactive({
- podTemplate: '',
- jmPodTemplate: '',
- tmPodTemplate: '',
- });
+const k8sTemplate = reactive({
+ podTemplate: '',
+ jmPodTemplate: '',
+ tmPodTemplate: '',
+});
- const { flinkEnvs, flinkClusters, getCreateFormSchema, suggestions } =
- useCreateSchema(dependencyRef);
+const { flinkEnvs, flinkClusters, getCreateFormSchema, suggestions } =
+ useCreateSchema(dependencyRef);
- const [registerAppForm, { setFieldsValue, getFieldsValue, submit }] = useForm({
- labelCol: { lg: { span: 5, offset: 0 }, sm: { span: 7, offset: 0 } },
- wrapperCol: { lg: { span: 16, offset: 0 }, sm: { span: 17, offset: 0 } },
- baseColProps: { span: 24 },
- colon: true,
- showActionButtonGroup: false,
- });
+const [registerAppForm, { setFieldsValue, getFieldsValue, submit }] = useForm({
+ labelCol: { lg: { span: 5, offset: 0 }, sm: { span: 7, offset: 0 } },
+ wrapperCol: { lg: { span: 16, offset: 0 }, sm: { span: 17, offset: 0 } },
+ baseColProps: { span: 24 },
+ colon: true,
+ showActionButtonGroup: false,
+});
- const [registerConfDrawer, { openDrawer: openConfDrawer }] = useDrawer();
- const [registerReviewDrawer, { openDrawer: openReviewDrawer }] = useDrawer();
+const [registerConfDrawer, { openDrawer: openConfDrawer }] = useDrawer();
+const [registerReviewDrawer, { openDrawer: openReviewDrawer }] = useDrawer();
- /* Initialize the form */
- async function handleInitForm() {
- const defaultValue = {
- resolveOrder: 0,
- k8sRestExposedType: 0,
- };
- options.forEach((item) => {
- defaultValue[item.key] = item.defaultValue;
- });
- const v = flinkEnvs.value.filter((v) => v.isDefault)[0];
- if (v) {
- Object.assign(defaultValue, { versionId: v.id });
- }
- await setFieldsValue(defaultValue);
+/* Initialize the form */
+async function handleInitForm() {
+ const defaultValue = {
+ resolveOrder: 0,
+ k8sRestExposedType: 0,
+ };
+ options.forEach((item) => {
+ defaultValue[item.key] = item.defaultValue;
+ });
+ const v = flinkEnvs.value.filter((v) => v.isDefault)[0];
+ if (v) {
+ Object.assign(defaultValue, { versionId: v.id });
}
+ await setFieldsValue(defaultValue);
+}
- /* Open the sqlConf drawer */
- async function handleSQLConf(checked: boolean, model: Recordable) {
- if (checked) {
- if (model.configOverride) {
- openConfDrawer(true, {
- configOverride: model.configOverride,
- });
- } else {
- const res = await handleConfTemplate();
- openConfDrawer(true, {
- configOverride: decodeByBase64(res),
- });
- }
+/* Open the sqlConf drawer */
+async function handleSQLConf(checked: boolean, model: Recordable) {
+ if (checked) {
+ if (model.configOverride) {
+ openConfDrawer(true, {
+ configOverride: model.configOverride,
+ });
} else {
- openConfDrawer(false);
- setFieldsValue({ isSetConfig: false, configOverride: null });
+ const res = await handleConfTemplate();
+ openConfDrawer(true, {
+ configOverride: decodeByBase64(res),
+ });
}
+ } else {
+ openConfDrawer(false);
+ setFieldsValue({ isSetConfig: false, configOverride: null });
}
+}
- function handleEditConfClose() {
- const formValue = getFieldsValue();
- if (!formValue.configOverride) {
- setFieldsValue({ isSetConfig: false });
- }
+function handleEditConfClose() {
+ const formValue = getFieldsValue();
+ if (!formValue.configOverride) {
+ setFieldsValue({ isSetConfig: false });
}
+}
- function handleCluster(values: Recordable) {
- let flinkClusterId =
- values.deployMode == DeployMode.YARN_SESSION
- ? values.yarnSessionClusterId
- : values.flinkClusterId;
- const cluster =
- unref(flinkClusters).filter((c) => {
- if (flinkClusterId) {
- return c.id == flinkClusterId && c.clusterState === ClusterStateEnum.RUNNING;
- }
- })[0] || null;
- if (cluster) {
- Object.assign(values, { flinkClusterId: cluster.id });
- if (values.deployMode == DeployMode.KUBERNETES_SESSION) {
- Object.assign(values, { clusterId: cluster.clusterId });
+function handleCluster(values: Recordable) {
+ let flinkClusterId =
+ values.deployMode == DeployMode.YARN_SESSION
+ ? values.yarnSessionClusterId
+ : values.flinkClusterId;
+ const cluster =
+ unref(flinkClusters).filter((c) => {
+ if (flinkClusterId) {
+ return c.id == flinkClusterId && c.clusterState === ClusterStateEnum.RUNNING;
}
+ })[0] || null;
+ if (cluster) {
+ Object.assign(values, { flinkClusterId: cluster.id });
+ if (values.deployMode == DeployMode.KUBERNETES_SESSION) {
+ Object.assign(values, { clusterId: cluster.clusterId });
}
}
+}
- /* custom mode */
- async function handleSubmitCustomJob(values) {
- handleCluster(values);
- // Trigger a pom confirmation operation.
- await unref(dependencyRef)?.handleApplyPom();
- // common params...
- const dependency: { pom?: string; jar?: string } = {};
- const dependencyRecords = unref(dependencyRef)?.dependencyRecords;
- const uploadJars = unref(dependencyRef)?.uploadJars;
- if (unref(dependencyRecords) && unref(dependencyRecords).length > 0) {
- Object.assign(dependency, {
- pom: unref(dependencyRecords),
- });
- }
- if (uploadJars && unref(uploadJars).length > 0) {
- Object.assign(dependency, {
- jar: unref(uploadJars),
- });
- }
- const params = {
- jobType: values.jobType,
- projectId: values.project || null,
- module: values.module || null,
- dependency:
- dependency.pom === undefined && dependency.jar === undefined
- ? null
- : JSON.stringify(dependency),
- appType: values.appType,
- };
- handleSubmitParams(params, values, k8sTemplate);
- // common params...
- const resourceFrom = values.resourceFrom;
- if (resourceFrom) {
- if (resourceFrom == ResourceFromEnum.PROJECT) {
- params['resourceFrom'] = ResourceFromEnum.PROJECT;
- //streampark flink
- if (values.appType == AppTypeEnum.STREAMPARK_FLINK) {
- const configVal = values.config;
- params['format'] = getAppConfType(configVal);
- if (values.configOverride == null) {
- params['config'] = await fetchAppConf({
- config: configVal,
- });
- } else {
- params['config'] = decodeByBase64(values.configOverride);
- }
+/* custom mode */
+async function handleSubmitCustomJob(values) {
+ handleCluster(values);
+ // Trigger a pom confirmation operation.
+ await unref(dependencyRef)?.handleApplyPom();
+ // common params...
+ const dependency: { pom?: string; jar?: string } = {};
+ const dependencyRecords = unref(dependencyRef)?.dependencyRecords;
+ const uploadJars = unref(dependencyRef)?.uploadJars;
+ if (unref(dependencyRecords) && unref(dependencyRecords).length > 0) {
+ Object.assign(dependency, {
+ pom: unref(dependencyRecords),
+ });
+ }
+ if (uploadJars && unref(uploadJars).length > 0) {
+ Object.assign(dependency, {
+ jar: unref(uploadJars),
+ });
+ }
+ const params = {
+ jobType: values.jobType,
+ projectId: values.project || null,
+ module: values.module || null,
+ dependency:
+ dependency.pom === undefined && dependency.jar === undefined
+ ? null
+ : JSON.stringify(dependency),
+ appType: values.appType,
+ };
+ handleSubmitParams(params, values, k8sTemplate);
+ // common params...
+ const resourceFrom = values.resourceFrom;
+ if (resourceFrom) {
+ if (resourceFrom == ResourceFromEnum.PROJECT) {
+ params['resourceFrom'] = ResourceFromEnum.PROJECT;
+ //streampark flink
+ if (values.appType == AppTypeEnum.STREAMPARK_FLINK) {
+ const configVal = values.config;
+ params['format'] = getAppConfType(configVal);
+ if (values.configOverride == null) {
+ params['config'] = await fetchAppConf({
+ config: configVal,
+ });
} else {
- params['jar'] = values.jar || null;
- params['mainClass'] = values.mainClass || null;
+ params['config'] = decodeByBase64(values.configOverride);
}
- handleCreateApp(params);
} else {
- // from upload
- Object.assign(params, {
- resourceFrom: ResourceFromEnum.UPLOAD,
- appType: AppTypeEnum.APACHE_FLINK,
- jar: values.uploadJobJar,
- mainClass: values.mainClass,
- });
- handleCreateApp(params);
+ params['jar'] = values.jar || null;
+ params['mainClass'] = values.mainClass || null;
}
- }
- }
- /* flink sql mode */
- async function handleSubmitSQL(values: Recordable) {
- // Trigger a pom confirmation operation.
- await unref(dependencyRef)?.handleApplyPom();
- // common params...
- const dependency: { pom?: string; jar?: string } = {};
- const dependencyRecords = unref(dependencyRef)?.dependencyRecords;
- const uploadJars = unref(dependencyRef)?.uploadJars;
- if (unref(dependencyRecords) && unref(dependencyRecords).length > 0) {
- Object.assign(dependency, {
- pom: unref(dependencyRecords),
- });
- }
- if (uploadJars && unref(uploadJars).length > 0) {
- Object.assign(dependency, {
- jar: unref(uploadJars),
- });
- }
-
- let config = values.configOverride;
- if (config != null && config.trim() != '') {
- config = encryptByBase64(config);
+ handleCreateApp(params);
} else {
- config = null;
+ // from upload
+ Object.assign(params, {
+ resourceFrom: ResourceFromEnum.UPLOAD,
+ appType: AppTypeEnum.APACHE_FLINK,
+ jar: values.uploadJobJar,
+ mainClass: values.mainClass,
+ });
+ handleCreateApp(params);
}
+ }
+}
+/* flink sql mode */
+async function handleSubmitSQL(values: Recordable) {
+ // Trigger a pom confirmation operation.
+ await unref(dependencyRef)?.handleApplyPom();
+ // common params...
+ const dependency: { pom?: string; jar?: string } = {};
+ const dependencyRecords = unref(dependencyRef)?.dependencyRecords;
+ const uploadJars = unref(dependencyRef)?.uploadJars;
+ if (unref(dependencyRecords) && unref(dependencyRecords).length > 0) {
+ Object.assign(dependency, {
+ pom: unref(dependencyRecords),
+ });
+ }
+ if (uploadJars && unref(uploadJars).length > 0) {
+ Object.assign(dependency, {
+ jar: unref(uploadJars),
+ });
+ }
- handleCluster(values);
- const params = {
- jobType: values.jobType,
- flinkSql: values.flinkSql,
- appType: AppTypeEnum.STREAMPARK_FLINK,
- config,
- format: values.isSetConfig ? 1 : null,
- teamResource: JSON.stringify(values.teamResource),
- dependency:
- dependency.pom === undefined && dependency.jar === undefined
- ? null
- : JSON.stringify(dependency),
- };
- handleSubmitParams(params, values, k8sTemplate);
- await handleCreateApp(params);
+ let config = values.configOverride;
+ if (config != null && config.trim() != '') {
+ config = encryptByBase64(config);
+ } else {
+ config = null;
}
- /* Submit to create */
- async function handleAppCreate(formValue: Recordable) {
- try {
- submitLoading.value = true;
- if (formValue.jobType == JobTypeEnum.SQL || formValue.jobType == JobTypeEnum.CDC) {
- if (formValue.flinkSql == null || formValue.flinkSql.trim() === '') {
- const errorMsg =
- formValue.jobType == JobTypeEnum.SQL
- ? t('flink.app.editStreamPark.flinkSqlRequired')
- : t('flink.app.editStreamPark.yamlRequired');
- createMessage.warning(errorMsg);
- } else {
- const access = await flinkSql?.value?.handleVerifySql();
- if (!access) {
- const errorMsg =
- formValue.jobType == JobTypeEnum.SQL
- ? t('flink.app.editStreamPark.sqlCheck')
- : t('flink.app.editStreamPark.yamlCheck');
- createMessage.warning(errorMsg);
- throw new Error(access);
- }
- }
- await handleSubmitSQL(formValue);
+
+ handleCluster(values);
+ const params = {
+ jobType: JobTypeEnum.SQL,
+ flinkSql: values.flinkSql,
+ appType: AppTypeEnum.STREAMPARK_FLINK,
+ config,
+ format: values.isSetConfig ? 1 : null,
+ teamResource: JSON.stringify(values.teamResource),
+ dependency:
+ dependency.pom === undefined && dependency.jar === undefined
+ ? null
+ : JSON.stringify(dependency),
+ };
+ handleSubmitParams(params, values, k8sTemplate);
+ await handleCreateApp(params);
+}
+/* Submit to create */
+async function handleAppCreate(formValue: Recordable) {
+ try {
+ submitLoading.value = true;
+ if (formValue.jobType == JobTypeEnum.SQL) {
+ if (formValue.flinkSql == null || formValue.flinkSql.trim() === '') {
+ createMessage.warning(t('flink.app.editStreamPark.flinkSqlRequired'));
} else {
- await handleSubmitCustomJob(formValue);
- }
- } catch (error) {
- submitLoading.value = false;
- }
- }
- /* send create request */
- async function handleCreateApp(params: Recordable) {
- const param = {};
- for (const k in params) {
- const v = params[k];
- if (v != null) {
- param[k] = v;
+ const access = await flinkSql?.value?.handleVerifySql();
+ if (!access) {
+ createMessage.warning(t('flink.app.editStreamPark.sqlCheck'));
+ throw new Error(access);
+ }
}
+ await handleSubmitSQL(formValue);
+ } else {
+ await handleSubmitCustomJob(formValue);
}
- const socketId = buildUUID();
- ls.set('DOWN_SOCKET_ID', socketId);
- Object.assign(param, { socketId });
- const { data } = await fetchCreate(param as CreateParams);
+ } catch (error) {
submitLoading.value = false;
- if (data.data) {
- go('/flink/app');
- } else {
- createMessage.error(data.message);
+ }
+}
+/* send create request */
+async function handleCreateApp(params: Recordable) {
+ const param = {};
+ for (const k in params) {
+ const v = params[k];
+ if (v != null) {
+ param[k] = v;
}
}
+ const socketId = buildUUID();
+ ls.set('DOWN_SOCKET_ID', socketId);
+ Object.assign(param, { socketId });
+ const { data } = await fetchCreate(param as CreateParams);
+ submitLoading.value = false;
+ if (data.data) {
+ go('/flink/app');
+ } else {
+ createMessage.error(data.message);
+ }
+}
- onMounted(async () => {
- handleInitForm();
- });
+onMounted(async () => {
+ handleInitForm();
+});
</script>
<template>
@@ -324,7 +316,6 @@
v-model:value="model[field]"
:versionId="model['versionId']"
:suggestions="suggestions"
- :jobType="Number(model['jobType'])"
@preview="(value) => openReviewDrawer(true, { value, suggestions })"
/>
</template>
@@ -384,5 +375,5 @@
</PageWrapper>
</template>
<style lang="less">
- @import url('./styles/Add.less');
+@import url('./styles/Add.less');
</style>
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/EditStreamPark.vue b/streampark-console/streampark-console-webapp/src/views/flink/app/EditStreamPark.vue
index 8eb133a80..2d1fa1900 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/EditStreamPark.vue
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/EditStreamPark.vue
@@ -15,348 +15,340 @@
limitations under the License.
-->
<script lang="ts">
- export default {
- name: 'EditStreamPark',
- };
+export default {
+ name: 'EditStreamPark',
+};
</script>
<script setup lang="ts" name="EditStreamPark">
- import { PageWrapper } from '/@/components/Page';
- import { BasicForm, useForm } from '/@/components/Form';
- import { onMounted, reactive, ref, nextTick, unref } from 'vue';
- import { AppListRecord } from '/@/api/flink/app.type';
- import configOptions from './data/option';
- import { fetchUpdate, fetchGet } from '/@/api/flink/app';
- import { fetchUpload } from '/@/api/resource/upload';
- import { useRoute } from 'vue-router';
- import { getAppConfType, handleSubmitParams, handleTeamResource } from './utils';
- import { fetchFlinkHistory } from '/@/api/flink/flinkSql';
- import { decodeByBase64, encryptByBase64 } from '/@/utils/cipher';
- import PomTemplateTab from './components/PodTemplate/PomTemplateTab.vue';
- import UploadJobJar from './components/UploadJobJar.vue';
- import FlinkSqlEditor from './components/FlinkSql.vue';
- import Dependency from './components/Dependency.vue';
- import Different from './components/AppDetail/Different.vue';
- import Mergely from './components/Mergely.vue';
- import AppConf from './components/AppConf';
- import UseSysHadoopConf from './components/UseSysHadoopConf.vue';
- import CompareConf from './components/CompareConf';
- import { useMessage } from '/@/hooks/web/useMessage';
- import { fetchConfHistory } from '/@/api/flink/config';
- import { useDrawer } from '/@/components/Drawer';
- import { useEditStreamParkSchema } from './hooks/useEditStreamPark';
- import { useEdit } from './hooks/useEdit';
- import { useI18n } from '/@/hooks/web/useI18n';
- import { useGo } from '/@/hooks/web/usePage';
- import ProgramArgs from './components/ProgramArgs.vue';
- import VariableReview from './components/VariableReview.vue';
- import { DeployMode, JobTypeEnum, UseStrategyEnum } from '/@/enums/flinkEnum';
+import { PageWrapper } from '/@/components/Page';
+import { BasicForm, useForm } from '/@/components/Form';
+import { onMounted, reactive, ref, nextTick, unref } from 'vue';
+import { AppListRecord } from '/@/api/flink/app.type';
+import configOptions from './data/option';
+import { fetchUpdate, fetchGet } from '/@/api/flink/app';
+import { fetchUpload } from '/@/api/resource/upload';
+import { useRoute } from 'vue-router';
+import { getAppConfType, handleSubmitParams, handleTeamResource } from './utils';
+import { fetchFlinkHistory } from '/@/api/flink/flinkSql';
+import { decodeByBase64, encryptByBase64 } from '/@/utils/cipher';
+import PomTemplateTab from './components/PodTemplate/PomTemplateTab.vue';
+import UploadJobJar from './components/UploadJobJar.vue';
+import FlinkSqlEditor from './components/FlinkSql.vue';
+import Dependency from './components/Dependency.vue';
+import Different from './components/AppDetail/Different.vue';
+import Mergely from './components/Mergely.vue';
+import AppConf from './components/AppConf';
+import UseSysHadoopConf from './components/UseSysHadoopConf.vue';
+import CompareConf from './components/CompareConf';
+import { useMessage } from '/@/hooks/web/useMessage';
+import { fetchConfHistory } from '/@/api/flink/config';
+import { useDrawer } from '/@/components/Drawer';
+import { useEditStreamParkSchema } from './hooks/useEditStreamPark';
+import { useEdit } from './hooks/useEdit';
+import { useI18n } from '/@/hooks/web/useI18n';
+import { useGo } from '/@/hooks/web/usePage';
+import ProgramArgs from './components/ProgramArgs.vue';
+import VariableReview from './components/VariableReview.vue';
+import { DeployMode, JobTypeEnum, UseStrategyEnum } from '/@/enums/flinkEnum';
- const route = useRoute();
- const go = useGo();
- const { t } = useI18n();
- const { createMessage } = useMessage();
- const app = reactive<Partial<AppListRecord>>({});
- const flinkSqlHistory = ref<any[]>([]);
- const submitLoading = ref<boolean>(false);
+const route = useRoute();
+const go = useGo();
+const { t } = useI18n();
+const { createMessage } = useMessage();
+const app = reactive<Partial<AppListRecord>>({});
+const flinkSqlHistory = ref<any[]>([]);
+const submitLoading = ref<boolean>(false);
- const configVersions = ref<Array<{ id: string }>>([]);
+const configVersions = ref<Array<{ id: string }>>([]);
- const uploadLoading = ref(false);
- const uploadJar = ref('');
- const dependencyRef = ref();
- const programArgRef = ref();
- const podTemplateRef = ref();
+const uploadLoading = ref(false);
+const uploadJar = ref('');
+const dependencyRef = ref();
+const programArgRef = ref();
+const podTemplateRef = ref();
- const k8sTemplate = reactive({
- podTemplate: '',
- jmPodTemplate: '',
- tmPodTemplate: '',
- });
+const k8sTemplate = reactive({
+ podTemplate: '',
+ jmPodTemplate: '',
+ tmPodTemplate: '',
+});
- const { handleResetApplication, defaultOptions } = useEdit();
- const {
- alerts,
- flinkEnvs,
- flinkSql,
- getEditStreamParkFormSchema,
- registerDifferentDrawer,
- suggestions,
- } = useEditStreamParkSchema(configVersions, flinkSqlHistory, dependencyRef);
+const { handleResetApplication, defaultOptions } = useEdit();
+const {
+ alerts,
+ flinkEnvs,
+ flinkSql,
+ getEditStreamParkFormSchema,
+ registerDifferentDrawer,
+ suggestions,
+} = useEditStreamParkSchema(configVersions, flinkSqlHistory, dependencyRef);
- const [registerForm, { setFieldsValue, getFieldsValue, submit }] = useForm({
- labelWidth: 120,
- colon: true,
- baseColProps: { span: 24 },
- labelCol: { lg: { span: 5, offset: 0 }, sm: { span: 7, offset: 0 } },
- wrapperCol: { lg: { span: 16, offset: 0 }, sm: { span: 17, offset: 0 } },
- showActionButtonGroup: false,
- });
+const [registerForm, { setFieldsValue, getFieldsValue, submit }] = useForm({
+ labelWidth: 120,
+ colon: true,
+ baseColProps: { span: 24 },
+ labelCol: { lg: { span: 5, offset: 0 }, sm: { span: 7, offset: 0 } },
+ wrapperCol: { lg: { span: 16, offset: 0 }, sm: { span: 17, offset: 0 } },
+ showActionButtonGroup: false,
+});
- const [registerConfDrawer, { openDrawer: openConfDrawer }] = useDrawer();
- const [registerReviewDrawer, { openDrawer: openReviewDrawer }] = useDrawer();
+const [registerConfDrawer, { openDrawer: openConfDrawer }] = useDrawer();
+const [registerReviewDrawer, { openDrawer: openReviewDrawer }] = useDrawer();
- /* Form reset */
- function handleReset(deployMode?: string) {
- let selectAlertId = '';
- if (app.alertId) {
- selectAlertId = unref(alerts).filter((t) => t.id == app.alertId)[0]?.id;
- }
- nextTick(() => {
- const resetParams = handleResetApplication();
- const defaultParams = {
- jobName: app.jobName,
- tags: app.tags,
- args: app.args || '',
- description: app.description,
- dynamicProperties: app.dynamicProperties,
- resolveOrder: app.resolveOrder,
- versionId: app.versionId || null,
- teamResource: handleTeamResource(app.teamResource),
- k8sRestExposedType: app.k8sRestExposedType,
- yarnQueue: app.yarnQueue,
- restartSize: app.restartSize,
- alertId: selectAlertId,
- checkPointFailure: {
- cpMaxFailureInterval: app.cpMaxFailureInterval,
- cpFailureRateInterval: app.cpFailureRateInterval,
- cpFailureAction: app.cpFailureAction,
- },
- flinkImage: app.flinkImage,
- k8sNamespace: app.k8sNamespace,
- ...resetParams,
- };
- switch (app.deployMode) {
- case DeployMode.STANDALONE:
- defaultParams['remoteClusterId'] = app.flinkClusterId;
- break;
- case DeployMode.YARN_SESSION:
- defaultParams['yarnSessionClusterId'] = app.flinkClusterId;
- break;
- case DeployMode.KUBERNETES_SESSION:
- defaultParams['k8sSessionClusterId'] = app.flinkClusterId;
- break;
- default:
- break;
- }
- if (!deployMode) {
- Object.assign(defaultParams, { deployMode: app.deployMode });
- }
- setFieldsValue(defaultParams);
- app.args && programArgRef.value?.setContent(app.args);
- });
+/* Form reset */
+function handleReset(deployMode?: string) {
+ let selectAlertId = '';
+ if (app.alertId) {
+ selectAlertId = unref(alerts).filter((t) => t.id == app.alertId)[0]?.id;
}
- /* Custom job upload */
- async function handleCustomJobRequest(data) {
- const formData = new FormData();
- formData.append('file', data.file);
- try {
- const resp = await fetchUpload(formData);
- uploadJar.value = data.file.name;
- uploadLoading.value = false;
- setFieldsValue({ jar: uploadJar.value, mainClass: resp.mainClass });
- } catch (error) {
- console.error(error);
- uploadLoading.value = false;
+ nextTick(() => {
+ const resetParams = handleResetApplication();
+ const defaultParams = {
+ jobName: app.jobName,
+ tags: app.tags,
+ args: app.args || '',
+ description: app.description,
+ dynamicProperties: app.dynamicProperties,
+ resolveOrder: app.resolveOrder,
+ versionId: app.versionId || null,
+ teamResource: handleTeamResource(app.teamResource),
+ k8sRestExposedType: app.k8sRestExposedType,
+ yarnQueue: app.yarnQueue,
+ restartSize: app.restartSize,
+ alertId: selectAlertId,
+ checkPointFailure: {
+ cpMaxFailureInterval: app.cpMaxFailureInterval,
+ cpFailureRateInterval: app.cpFailureRateInterval,
+ cpFailureAction: app.cpFailureAction,
+ },
+ flinkImage: app.flinkImage,
+ k8sNamespace: app.k8sNamespace,
+ ...resetParams,
+ };
+ switch (app.deployMode) {
+ case DeployMode.STANDALONE:
+ defaultParams['remoteClusterId'] = app.flinkClusterId;
+ break;
+ case DeployMode.YARN_SESSION:
+ defaultParams['yarnSessionClusterId'] = app.flinkClusterId;
+ break;
+ case DeployMode.KUBERNETES_SESSION:
+ defaultParams['k8sSessionClusterId'] = app.flinkClusterId;
+ break;
+ default:
+ break;
+ }
+ if (!deployMode) {
+ Object.assign(defaultParams, { deployMode: app.deployMode });
}
+ setFieldsValue(defaultParams);
+ app.args && programArgRef.value?.setContent(app.args);
+ });
+}
+/* Custom job upload */
+async function handleCustomJobRequest(data) {
+ const formData = new FormData();
+ formData.append('file', data.file);
+ try {
+ const resp = await fetchUpload(formData);
+ uploadJar.value = data.file.name;
+ uploadLoading.value = false;
+ setFieldsValue({ jar: uploadJar.value, mainClass: resp.mainClass });
+ } catch (error) {
+ console.error(error);
+ uploadLoading.value = false;
}
+}
- /* Handling update parameters */
- async function handleAppUpdate(values) {
- try {
- submitLoading.value = true;
- if (app.jobType == JobTypeEnum.SQL || app.jobType == JobTypeEnum.CDC) {
- if (values.flinkSql == null || values.flinkSql.trim() === '') {
- const errorMsg =
- app.jobType == JobTypeEnum.SQL
- ? t('flink.app.editStreamPark.flinkSqlRequired')
- : t('flink.app.editStreamPark.yamlRequired');
- createMessage.warning(errorMsg);
- } else {
- const access = await flinkSql?.value?.handleVerifySql();
- if (!access) {
- const errorMsg =
- app.jobType == JobTypeEnum.SQL
- ? t('flink.app.editStreamPark.sqlCheck')
- : t('flink.app.editStreamPark.yamlCheck');
- createMessage.warning(errorMsg);
- throw new Error(access);
- }
- handleSubmitSQL(values);
- }
+/* Handling update parameters */
+async function handleAppUpdate(values) {
+ try {
+ submitLoading.value = true;
+ if (app.jobType == JobTypeEnum.SQL) {
+ if (values.flinkSql == null || values.flinkSql.trim() === '') {
+ createMessage.warning(t('flink.app.editStreamPark.flinkSqlRequired'));
} else {
- handleSubmitCustomJob(values);
+ const access = await flinkSql?.value?.handleVerifySql();
+ if (!access) {
+ createMessage.warning(t('flink.app.editStreamPark.sqlCheck'));
+ throw new Error(access);
+ }
+ handleSubmitSQL(values);
}
- } catch (error) {
- console.error(error);
- submitLoading.value = false;
+ } else {
+ handleSubmitCustomJob(values);
}
+ } catch (error) {
+ console.error(error);
+ submitLoading.value = false;
}
+}
- async function handleSubmitSQL(values: Recordable) {
- try {
- // Trigger a pom confirmation operation.
- await unref(dependencyRef)?.handleApplyPom();
- // common params...
- const dependency: { pom?: string; jar?: string } = {};
- const dependencyRecords = unref(dependencyRef)?.dependencyRecords;
- const uploadJars = unref(dependencyRef)?.uploadJars;
- if (unref(dependencyRecords) && unref(dependencyRecords).length > 0) {
- Object.assign(dependency, {
- pom: unref(dependencyRecords),
- });
- }
- if (uploadJars && unref(uploadJars).length > 0) {
- Object.assign(dependency, {
- jar: unref(uploadJars),
- });
- }
- let config = values.configOverride;
- if (config != null && config.trim() !== '') {
- config = encryptByBase64(config);
- } else {
- config = null;
- }
- const params = {
- id: app.id,
- sqlId: values.flinkSqlHistory || app.sqlId || null,
- flinkSql: values.flinkSql,
- config,
- format: values.isSetConfig ? 1 : null,
- dependency:
- dependency.pom === undefined && dependency.jar === undefined
- ? null
- : JSON.stringify(dependency),
- };
- handleSubmitParams(params, values, k8sTemplate);
- await handleUpdateApp(params);
- } catch (error) {
- createMessage.error('edit error');
- submitLoading.value = false;
+async function handleSubmitSQL(values: Recordable) {
+ try {
+ // Trigger a pom confirmation operation.
+ await unref(dependencyRef)?.handleApplyPom();
+ // common params...
+ const dependency: { pom?: string; jar?: string } = {};
+ const dependencyRecords = unref(dependencyRef)?.dependencyRecords;
+ const uploadJars = unref(dependencyRef)?.uploadJars;
+ if (unref(dependencyRecords) && unref(dependencyRecords).length > 0) {
+ Object.assign(dependency, {
+ pom: unref(dependencyRecords),
+ });
+ }
+ if (uploadJars && unref(uploadJars).length > 0) {
+ Object.assign(dependency, {
+ jar: unref(uploadJars),
+ });
+ }
+ let config = values.configOverride;
+ if (config != null && config.trim() !== '') {
+ config = encryptByBase64(config);
+ } else {
+ config = null;
}
+ const params = {
+ id: app.id,
+ sqlId: values.flinkSqlHistory || app.sqlId || null,
+ flinkSql: values.flinkSql,
+ config,
+ format: values.isSetConfig ? 1 : null,
+ dependency:
+ dependency.pom === undefined && dependency.jar === undefined
+ ? null
+ : JSON.stringify(dependency),
+ };
+ handleSubmitParams(params, values, k8sTemplate);
+ await handleUpdateApp(params);
+ } catch (error) {
+ createMessage.error('edit error');
+ submitLoading.value = false;
}
- /* Submit an update */
- async function handleSubmitCustomJob(values: Recordable) {
- try {
- const format =
- values.strategy == UseStrategyEnum.USE_EXIST
- ? app.format
- : getAppConfType(values.config || '');
- let config = values.configOverride || app.config;
- if (config != null && config.trim() !== '') {
- config = encryptByBase64(config);
- } else {
- config = null;
- }
- const configId = values.strategy == UseStrategyEnum.USE_EXIST ? app.configId : null;
- const params = {
- id: app.id,
- format: format,
- configId,
- config,
- };
- handleSubmitParams(params, values, k8sTemplate);
- handleUpdateApp(params);
- } catch (error) {
- console.error('error', error);
- submitLoading.value = false;
+}
+/* Submit an update */
+async function handleSubmitCustomJob(values: Recordable) {
+ try {
+ const format =
+ values.strategy == UseStrategyEnum.USE_EXIST
+ ? app.format
+ : getAppConfType(values.config || '');
+ let config = values.configOverride || app.config;
+ if (config != null && config.trim() !== '') {
+ config = encryptByBase64(config);
+ } else {
+ config = null;
}
+ const configId = values.strategy == UseStrategyEnum.USE_EXIST ? app.configId : null;
+ const params = {
+ id: app.id,
+ format: format,
+ configId,
+ config,
+ };
+ handleSubmitParams(params, values, k8sTemplate);
+ handleUpdateApp(params);
+ } catch (error) {
+ console.error('error', error);
+ submitLoading.value = false;
}
+}
- /* Send submission interface */
- async function handleUpdateApp(params: Recordable) {
- try {
- const updated = await fetchUpdate(params);
- if (updated) {
- go('/flink/app');
- }
- } catch (error) {
- console.error('error', error);
- } finally {
- submitLoading.value = false;
+/* Send submission interface */
+async function handleUpdateApp(params: Recordable) {
+ try {
+ const updated = await fetchUpdate(params);
+ if (updated) {
+ go('/flink/app');
}
+ } catch (error) {
+ console.error('error', error);
+ } finally {
+ submitLoading.value = false;
}
+}
- /* initialization information */
- async function handleStreamParkInfo() {
- const appId = route.query.appId;
- const res = await fetchGet({ id: appId as string });
- let configId = '';
- const confVersion = await fetchConfHistory({ id: route.query.appId });
- confVersion.forEach((conf: Recordable) => {
- if (conf.effective) {
- configId = conf.id;
- }
- });
- configVersions.value = confVersion;
- Object.assign(app, res);
- Object.assign(defaultOptions, JSON.parse(app.options || '{}'));
-
- if (app.jobType == JobTypeEnum.SQL || app.jobType == JobTypeEnum.CDC) {
- fetchFlinkHistory({ id: appId }).then((res) => {
- flinkSqlHistory.value = res;
- });
- }
- let isSetConfig = false;
- let configOverride = '';
- if (app.config && app.config.trim() !== '') {
- configOverride = decodeByBase64(app.config);
- isSetConfig = true;
+/* initialization information */
+async function handleStreamParkInfo() {
+ const appId = route.query.appId;
+ const res = await fetchGet({ id: appId as string });
+ let configId = '';
+ const confVersion = await fetchConfHistory({ id: route.query.appId });
+ confVersion.forEach((conf: Recordable) => {
+ if (conf.effective) {
+ configId = conf.id;
}
- const defaultFormValue = { isSetConfig, configOverride };
- configOptions.forEach((item) => {
- Object.assign(defaultFormValue, {
- [item.key]: item.defaultValue,
- });
- });
-
- setFieldsValue({
- jobType: res.jobType,
- appType: res.appType,
- deployMode: res.deployMode,
- flinkSql: res.flinkSql ? decodeByBase64(res.flinkSql) : '',
- dependency: '',
- module: res.module,
- configId,
- sqlId: app.sqlId,
- flinkSqlHistory: app.sqlId,
- versionId: app.versionId,
- projectName: app.projectName,
- project: app.projectId,
- ...defaultFormValue,
- });
- nextTick(() => {
- unref(flinkSql)?.setContent(decodeByBase64(res.flinkSql));
+ });
+ configVersions.value = confVersion;
+ Object.assign(app, res);
+ Object.assign(defaultOptions, JSON.parse(app.options || '{}'));
- setTimeout(() => {
- unref(dependencyRef)?.setDefaultValue(JSON.parse(res.dependency || '{}'));
- unref(podTemplateRef)?.handleChoicePodTemplate('ptVisual', res.k8sPodTemplate);
- unref(podTemplateRef)?.handleChoicePodTemplate('jmPtVisual', res.k8sJmPodTemplate);
- unref(podTemplateRef)?.handleChoicePodTemplate('tmPtVisual', res.k8sTmPodTemplate);
- }, 1000);
+ if (app.jobType == JobTypeEnum.SQL) {
+ fetchFlinkHistory({ id: appId }).then((res) => {
+ flinkSqlHistory.value = res;
});
- handleReset();
}
-
- function handleMergely(configOverride: string) {
- openConfDrawer(true, {
- configOverride,
- });
+ let isSetConfig = false;
+ let configOverride = '';
+ if (app.config && app.config.trim() !== '') {
+ configOverride = decodeByBase64(app.config);
+ isSetConfig = true;
}
+ const defaultFormValue = { isSetConfig, configOverride };
+ configOptions.forEach((item) => {
+ Object.assign(defaultFormValue, {
+ [item.key]: item.defaultValue,
+ });
+ });
- function handleEditConfClose() {
- const formValue = getFieldsValue();
- if (!formValue.configOverride) {
- setFieldsValue({ isSetConfig: false });
- }
- }
- onMounted(() => {
- if (!route?.query?.appId) {
- go('/flink/app');
- createMessage.warning(t('flink.app.editStreamPark.appidCheck'));
- return;
- }
- handleStreamParkInfo();
+ setFieldsValue({
+ jobType: res.jobType,
+ appType: res.appType,
+ deployMode: res.deployMode,
+ flinkSql: res.flinkSql ? decodeByBase64(res.flinkSql) : '',
+ dependency: '',
+ module: res.module,
+ configId,
+ sqlId: app.sqlId,
+ flinkSqlHistory: app.sqlId,
+ versionId: app.versionId,
+ projectName: app.projectName,
+ project: app.projectId,
+ ...defaultFormValue,
+ });
+ nextTick(() => {
+ unref(flinkSql)?.setContent(decodeByBase64(res.flinkSql));
+
+ setTimeout(() => {
+ unref(dependencyRef)?.setDefaultValue(JSON.parse(res.dependency || '{}'));
+ unref(podTemplateRef)?.handleChoicePodTemplate('ptVisual', res.k8sPodTemplate);
+ unref(podTemplateRef)?.handleChoicePodTemplate('jmPtVisual', res.k8sJmPodTemplate);
+ unref(podTemplateRef)?.handleChoicePodTemplate('tmPtVisual', res.k8sTmPodTemplate);
+ }, 1000);
+ });
+ handleReset();
+}
+
+function handleMergely(configOverride: string) {
+ openConfDrawer(true, {
+ configOverride,
});
+}
+
+function handleEditConfClose() {
+ const formValue = getFieldsValue();
+ if (!formValue.configOverride) {
+ setFieldsValue({ isSetConfig: false });
+ }
+}
+onMounted(() => {
+ if (!route?.query?.appId) {
+ go('/flink/app');
+ createMessage.warning(t('flink.app.editStreamPark.appidCheck'));
+ return;
+ }
+ handleStreamParkInfo();
+});
</script>
<template>
<PageWrapper contentBackground content-class="p-26px app_controller">
@@ -376,7 +368,7 @@
<template #args="{ model }">
<ProgramArgs
ref="programArgRef"
- v-if="model.args != null"
+ v-if="model.args != null && model.args != undefined"
v-model:value="model.args"
:suggestions="suggestions"
@preview="(value) => openReviewDrawer(true, { value, suggestions })"
@@ -391,7 +383,6 @@
v-model:value="model[field]"
:versionId="model['versionId']"
:suggestions="suggestions"
- :jobType="Number(model['jobType'])"
@preview="(value) => openReviewDrawer(true, { value, suggestions })"
/>
</template>
@@ -434,5 +425,5 @@
</PageWrapper>
</template>
<style lang="less">
- @import url('./styles/Add.less');
+@import url('./styles/Add.less');
</style>
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/View.vue b/streampark-console/streampark-console-webapp/src/views/flink/app/View.vue
index e33a45a82..c973db64d 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/View.vue
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/View.vue
@@ -15,250 +15,250 @@
limitations under the License.
-->
<script lang="ts" setup name="AppView">
- import { PlusOutlined } from '@ant-design/icons-vue';
- import { nextTick, ref, onUnmounted, onMounted } from 'vue';
- import { useAppTableAction } from './hooks/useAppTableAction';
- import { useI18n } from '/@/hooks/web/useI18n';
- import { JobTypeEnum, OptionStateEnum, ReleaseStateEnum } from '/@/enums/flinkEnum';
- import { useDebounceFn, useTimeoutFn } from '@vueuse/core';
- import {
- Form,
- Button,
- Select,
- Input,
- Tooltip,
- Badge,
- Tag,
- Popover,
- Row,
- Col,
- } from 'ant-design-vue';
- import { fetchAppRecord } from '/@/api/flink/app';
- import { useTable } from '/@/components/Table';
- import { PageWrapper } from '/@/components/Page';
- import { BasicTable, TableAction } from '/@/components/Table';
- import { AppListRecord } from '/@/api/flink/app.type';
- import { releaseTitleMap } from './data';
- import { useDrawer } from '/@/components/Drawer';
- import { useModal } from '/@/components/Modal';
+import { PlusOutlined } from '@ant-design/icons-vue';
+import { nextTick, ref, onUnmounted, onMounted } from 'vue';
+import { useAppTableAction } from './hooks/useAppTableAction';
+import { useI18n } from '/@/hooks/web/useI18n';
+import { JobTypeEnum, OptionStateEnum, ReleaseStateEnum } from '/@/enums/flinkEnum';
+import { useDebounceFn, useTimeoutFn } from '@vueuse/core';
+import {
+ Form,
+ Button,
+ Select,
+ Input,
+ Tooltip,
+ Badge,
+ Tag,
+ Popover,
+ Row,
+ Col,
+} from 'ant-design-vue';
+import { fetchAppRecord } from '/@/api/flink/app';
+import { useTable } from '/@/components/Table';
+import { PageWrapper } from '/@/components/Page';
+import { BasicTable, TableAction } from '/@/components/Table';
+import { AppListRecord } from '/@/api/flink/app.type';
+import { releaseTitleMap } from './data';
+import { useDrawer } from '/@/components/Drawer';
+import { useModal } from '/@/components/Modal';
- import StartApplicationModal from './components/AppView/StartApplicationModal.vue';
- import StopApplicationModal from './components/AppView/StopApplicationModal.vue';
- import LogModal from './components/AppView/LogModal.vue';
- import BuildDrawer from './components/AppView/BuildDrawer.vue';
- import AppDashboard from './components/AppView/AppDashboard.vue';
- import State, {
- buildStatusMap,
- optionStateMap,
- releaseStateMap,
- stateMap,
- } from './components/State';
- import { useSavepoint } from './hooks/useSavepoint';
- import { useAppTableColumns } from './hooks/useAppTableColumns';
- import AppTableResize from './components/AppResize.vue';
- import { useRouter } from 'vue-router';
+import StartApplicationModal from './components/AppView/StartApplicationModal.vue';
+import StopApplicationModal from './components/AppView/StopApplicationModal.vue';
+import LogModal from './components/AppView/LogModal.vue';
+import BuildDrawer from './components/AppView/BuildDrawer.vue';
+import AppDashboard from './components/AppView/AppDashboard.vue';
+import State, {
+ buildStatusMap,
+ optionStateMap,
+ releaseStateMap,
+ stateMap,
+} from './components/State';
+import { useSavepoint } from './hooks/useSavepoint';
+import { useAppTableColumns } from './hooks/useAppTableColumns';
+import AppTableResize from './components/AppResize.vue';
+import { useRouter } from 'vue-router';
- defineOptions({
- name: 'AppView',
- });
- const { t } = useI18n();
- const searchRef = ref<Recordable>({
- tags: undefined,
- owner: undefined,
- jobType: undefined,
- });
- const optionApps = {
- starting: new Map(),
- stopping: new Map(),
- release: new Map(),
- savepointing: new Map(),
- };
+defineOptions({
+ name: 'AppView',
+});
+const { t } = useI18n();
+const searchRef = ref<Recordable>({
+ tags: undefined,
+ owner: undefined,
+ jobType: undefined,
+});
+const optionApps = {
+ starting: new Map(),
+ stopping: new Map(),
+ release: new Map(),
+ savepointing: new Map(),
+};
- const titleLenRef = ref({
- maxState: '',
- maxRelease: '',
- maxBuild: '',
- });
+const titleLenRef = ref({
+ maxState: '',
+ maxRelease: '',
+ maxBuild: '',
+});
- const appDashboardRef = ref<any>();
- const noData = ref<boolean>();
+const appDashboardRef = ref<any>();
+const noData = ref<boolean>();
- const currentTablePage = ref(1);
- const router = useRouter();
- const { onTableColumnResize, tableColumnWidth, getAppColumns } = useAppTableColumns();
- const { openSavepoint } = useSavepoint(handleOptionApp);
- const [registerStartModal, { openModal: openStartModal }] = useModal();
- const [registerStopModal, { openModal: openStopModal }] = useModal();
- const [registerLogModal, { openModal: openLogModal }] = useModal();
- const [registerBuildDrawer, { openDrawer: openBuildDrawer }] = useDrawer();
+const currentTablePage = ref(1);
+const router = useRouter();
+const { onTableColumnResize, tableColumnWidth, getAppColumns } = useAppTableColumns();
+const { openSavepoint } = useSavepoint(handleOptionApp);
+const [registerStartModal, { openModal: openStartModal }] = useModal();
+const [registerStopModal, { openModal: openStopModal }] = useModal();
+const [registerLogModal, { openModal: openLogModal }] = useModal();
+const [registerBuildDrawer, { openDrawer: openBuildDrawer }] = useDrawer();
- const [registerTable, { reload, getLoading, setPagination }] = useTable({
- rowKey: 'id',
- api: fetchAppRecord,
- beforeFetch: (params) => {
- if (Reflect.has(params, 'state')) {
- if (params.state && params.state.length > 0) {
- params['stateArray'] = [...params.state];
- }
- delete params.state;
+const [registerTable, { reload, getLoading, setPagination }] = useTable({
+ rowKey: 'id',
+ api: fetchAppRecord,
+ beforeFetch: (params) => {
+ if (Reflect.has(params, 'state')) {
+ if (params.state && params.state.length > 0) {
+ params['stateArray'] = [...params.state];
}
- currentTablePage.value = params.pageNum;
- Object.assign(params, searchRef.value);
- sessionStorage.setItem('appPageNo', params.pageNum);
- return params;
- },
- afterFetch: (dataSource) => {
- const timestamp = new Date().getTime();
- noData.value = dataSource.length == 0;
- dataSource.forEach((x) => {
- x.expanded = [
- {
- jmMemory: x.jmMemory,
- tmMemory: x.tmMemory,
- totalTM: x.totalTM,
- totalSlot: x.totalSlot,
- availableSlot: x.availableSlot,
- },
- ];
- if (x['optionState'] === OptionStateEnum.NONE) {
- if (optionApps.starting.get(x.id)) {
- if (timestamp - optionApps.starting.get(x.id) > 2000 * 2) {
- optionApps.starting.delete(x.id);
- }
- }
- if (optionApps.stopping.get(x.id)) {
- if (timestamp - optionApps.stopping.get(x.id) > 2000) {
- optionApps.stopping.delete(x.id);
- }
+ delete params.state;
+ }
+ currentTablePage.value = params.pageNum;
+ Object.assign(params, searchRef.value);
+ sessionStorage.setItem('appPageNo', params.pageNum);
+ return params;
+ },
+ afterFetch: (dataSource) => {
+ const timestamp = new Date().getTime();
+ noData.value = dataSource.length == 0;
+ dataSource.forEach((x) => {
+ x.expanded = [
+ {
+ jmMemory: x.jmMemory,
+ tmMemory: x.tmMemory,
+ totalTM: x.totalTM,
+ totalSlot: x.totalSlot,
+ availableSlot: x.availableSlot,
+ },
+ ];
+ if (x['optionState'] === OptionStateEnum.NONE) {
+ if (optionApps.starting.get(x.id)) {
+ if (timestamp - optionApps.starting.get(x.id) > 2000 * 2) {
+ optionApps.starting.delete(x.id);
}
- if (optionApps.release.get(x.id)) {
- if (timestamp - optionApps.release.get(x.id) > 2000) {
- optionApps.release.delete(x.id);
- }
+ }
+ if (optionApps.stopping.get(x.id)) {
+ if (timestamp - optionApps.stopping.get(x.id) > 2000) {
+ optionApps.stopping.delete(x.id);
}
- if (optionApps.savepointing.get(x.id)) {
- if (timestamp - optionApps.savepointing.get(x.id) > 2000) {
- optionApps.savepointing.delete(x.id);
- }
+ }
+ if (optionApps.release.get(x.id)) {
+ if (timestamp - optionApps.release.get(x.id) > 2000) {
+ optionApps.release.delete(x.id);
}
}
- });
- const stateLenMap = dataSource.reduce(
- (
- prev: {
- maxState: string;
- maxRelease: string;
- maxBuild: string;
- },
- cur: any,
- ) => {
- const { state, optionState, release, buildStatus } = cur;
- // state title len
- if (optionState === OptionStateEnum.NONE) {
- const stateStr = stateMap[state]?.title;
- if (stateStr && stateStr.length > prev.maxState.length) {
- prev.maxState = stateStr;
- }
- } else {
- const stateStr = optionStateMap[optionState]?.title;
- if (stateStr && stateStr.length > prev.maxState.length) {
- prev.maxState = stateStr;
- }
+ if (optionApps.savepointing.get(x.id)) {
+ if (timestamp - optionApps.savepointing.get(x.id) > 2000) {
+ optionApps.savepointing.delete(x.id);
}
- //release title len
- const releaseStr = releaseStateMap[release]?.title;
- if (releaseStr && releaseStr.length > prev.maxRelease.length) {
- prev.maxRelease = releaseStr;
+ }
+ }
+ });
+ const stateLenMap = dataSource.reduce(
+ (
+ prev: {
+ maxState: string;
+ maxRelease: string;
+ maxBuild: string;
+ },
+ cur: any,
+ ) => {
+ const { state, optionState, release, buildStatus } = cur;
+ // state title len
+ if (optionState === OptionStateEnum.NONE) {
+ const stateStr = stateMap[state]?.title;
+ if (stateStr && stateStr.length > prev.maxState.length) {
+ prev.maxState = stateStr;
}
- //build title len
- const buildStr = buildStatusMap[buildStatus]?.title;
- if (buildStr && buildStr.length > prev.maxBuild.length) {
- prev.maxBuild = buildStr;
+ } else {
+ const stateStr = optionStateMap[optionState]?.title;
+ if (stateStr && stateStr.length > prev.maxState.length) {
+ prev.maxState = stateStr;
}
- return prev;
- },
- {
- maxState: '',
- maxRelease: '',
- maxBuild: '',
- },
- );
- Object.assign(titleLenRef.value, stateLenMap);
- return dataSource;
- },
- fetchSetting: { listField: 'records' },
- immediate: true,
- canResize: false,
- showIndexColumn: false,
- showTableSetting: false,
- useSearchForm: false,
- tableSetting: { fullScreen: true, redo: false },
- actionColumn: {
- dataIndex: 'operation',
- title: t('component.table.operation'),
- width: 180,
- },
- });
+ }
+ //release title len
+ const releaseStr = releaseStateMap[release]?.title;
+ if (releaseStr && releaseStr.length > prev.maxRelease.length) {
+ prev.maxRelease = releaseStr;
+ }
+ //build title len
+ const buildStr = buildStatusMap[buildStatus]?.title;
+ if (buildStr && buildStr.length > prev.maxBuild.length) {
+ prev.maxBuild = buildStr;
+ }
+ return prev;
+ },
+ {
+ maxState: '',
+ maxRelease: '',
+ maxBuild: '',
+ },
+ );
+ Object.assign(titleLenRef.value, stateLenMap);
+ return dataSource;
+ },
+ fetchSetting: { listField: 'records' },
+ immediate: true,
+ canResize: false,
+ showIndexColumn: false,
+ showTableSetting: false,
+ useSearchForm: false,
+ tableSetting: { fullScreen: true, redo: false },
+ actionColumn: {
+ dataIndex: 'operation',
+ title: t('component.table.operation'),
+ width: 180,
+ },
+});
- const { getTableActions, tagsOptions, users } = useAppTableAction(
- openStartModal,
- openStopModal,
- openSavepoint,
- openLogModal,
- openBuildDrawer,
- handlePageDataReload,
- optionApps,
- );
+const { getTableActions, tagsOptions, users } = useAppTableAction(
+ openStartModal,
+ openStopModal,
+ openSavepoint,
+ openLogModal,
+ openBuildDrawer,
+ handlePageDataReload,
+ optionApps,
+);
- /* view */
- function handleJobView(app: AppListRecord) {
- router.push({ path: '/flink/app/detail', query: { appId: app.id } });
- }
+/* view */
+function handleJobView(app: AppListRecord) {
+ router.push({ path: '/flink/app/detail', query: { appId: app.id } });
+}
- /* Update options data */
- function handleOptionApp(data: {
- type: 'starting' | 'stopping' | 'release' | 'savepointing';
- key: any;
- value: any;
- }) {
- optionApps[data.type].set(data.key, data.value);
- }
+/* Update options data */
+function handleOptionApp(data: {
+ type: 'starting' | 'stopping' | 'release' | 'savepointing';
+ key: any;
+ value: any;
+}) {
+ optionApps[data.type].set(data.key, data.value);
+}
- function handlePageDataReload(polling = false) {
- nextTick(() => {
- appDashboardRef.value?.handleDashboard(false);
- reload({ polling });
- });
+function handlePageDataReload(polling = false) {
+ nextTick(() => {
+ appDashboardRef.value?.handleDashboard(false);
+ reload({ polling });
+ });
+}
+
+const handleResetReload = useDebounceFn(() => {
+ setPagination({
+ current: 1,
+ });
+ reload();
+}, 500);
+
+const { start, stop } = useTimeoutFn(() => {
+ if (!getLoading()) {
+ handlePageDataReload(true);
}
+ start();
+}, 2000);
- const handleResetReload = useDebounceFn(() => {
+onMounted(() => {
+ // If there is a page, jump to the page number of the record
+ const currentPage = sessionStorage.getItem('appPageNo');
+ if (currentPage) {
setPagination({
- current: 1,
+ current: Number(currentPage) || 1,
});
- reload();
- }, 500);
-
- const { start, stop } = useTimeoutFn(() => {
- if (!getLoading()) {
- handlePageDataReload(true);
- }
- start();
- }, 2000);
-
- onMounted(() => {
- // If there is a page, jump to the page number of the record
- const currentPage = sessionStorage.getItem('appPageNo');
- if (currentPage) {
- setPagination({
- current: Number(currentPage) || 1,
- });
- sessionStorage.removeItem('appPageNo');
- }
- });
+ sessionStorage.removeItem('appPageNo');
+ }
+});
- onUnmounted(() => {
- stop();
- });
+onUnmounted(() => {
+ stop();
+});
</script>
<template>
<PageWrapper contentFullHeight content-class="flex flex-col">
@@ -310,7 +310,6 @@
:options="[
{ label: 'JAR', value: JobTypeEnum.JAR },
{ label: 'SQL', value: JobTypeEnum.SQL },
- { label: 'FlinkCDC', value: JobTypeEnum.CDC },
]"
/>
</Form.Item>
@@ -350,7 +349,6 @@
<template v-if="column.dataIndex === 'jobName'">
<span class="app_type app_jar" v-if="record['jobType'] === JobTypeEnum.JAR"> JAR </span>
<span class="app_type app_sql" v-if="record['jobType'] === JobTypeEnum.SQL"> SQL </span>
- <span class="app_type app_sql" v-if="record['jobType'] === JobTypeEnum.CDC"> CDC </span>
<span class="link cursor-pointer" @click="handleJobView(record)">
<Popover :title="t('common.detailText')">
<template #content>
@@ -363,7 +361,6 @@
<Tag color="blue">
<span v-if="record['jobType'] == JobTypeEnum.JAR"> JAR </span>
<span v-if="record['jobType'] == JobTypeEnum.SQL"> SQL </span>
- <span v-if="record['jobType'] == JobTypeEnum.CDC"> FlinkCDC </span>
</Tag>
</div>
<div class="pt-2px flex">
@@ -436,5 +433,5 @@
</PageWrapper>
</template>
<style lang="less">
- @import url('./styles/View.less');
+@import url('./styles/View.less');
</style>
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/components/FlinkSql.vue b/streampark-console/streampark-console-webapp/src/views/flink/app/components/FlinkSql.vue
index d8fa3486c..d166b3f4b 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/components/FlinkSql.vue
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/components/FlinkSql.vue
@@ -15,200 +15,176 @@
limitations under the License.
-->
<script lang="ts">
- export default {
- name: 'FlinkSql',
- components: { SvgIcon },
- };
+export default {
+ name: 'FlinkSql',
+ components: { SvgIcon },
+};
</script>
<script setup lang="ts" name="FlinkSql">
- import { computed, reactive, ref, watchEffect } from 'vue';
- import { Tooltip } from 'ant-design-vue';
- import { FullscreenExitOutlined } from '@ant-design/icons-vue';
- import { getMonacoOptions } from '../data';
- import { Icon, SvgIcon } from '/@/components/Icon';
- import { useMonaco } from '/@/hooks/web/useMonaco';
- import { Button } from 'ant-design-vue';
- import { isEmpty } from '/@/utils/is';
- import { useMessage } from '/@/hooks/web/useMessage';
- import { fetchFlinkSqlVerify } from '/@/api/flink/flinkSql';
- import { format } from '../FlinkSqlFormatter';
- import { useI18n } from '/@/hooks/web/useI18n';
- import { useFullContent } from '/@/hooks/event/useFullscreen';
- import { JobTypeEnum } from '/@/enums/flinkEnum';
- import YAML from 'js-yaml';
- const ButtonGroup = Button.Group;
- const { t } = useI18n();
+import { computed, reactive, ref, watchEffect } from 'vue';
+import { Tooltip } from 'ant-design-vue';
+import { FullscreenExitOutlined } from '@ant-design/icons-vue';
+import { getMonacoOptions } from '../data';
+import { Icon, SvgIcon } from '/@/components/Icon';
+import { useMonaco } from '/@/hooks/web/useMonaco';
+import { Button } from 'ant-design-vue';
+import { isEmpty } from '/@/utils/is';
+import { useMessage } from '/@/hooks/web/useMessage';
+import { fetchFlinkSqlVerify } from '/@/api/flink/flinkSql';
+import { format } from '../FlinkSqlFormatter';
+import { useI18n } from '/@/hooks/web/useI18n';
+import { useFullContent } from '/@/hooks/event/useFullscreen';
+const ButtonGroup = Button.Group;
+const { t } = useI18n();
- const flinkSql = ref();
- const verifyRes = reactive({
- errorMsg: '',
- verified: false,
- errorStart: 0,
- errorEnd: 0,
- });
+const flinkSql = ref();
+const verifyRes = reactive({
+ errorMsg: '',
+ verified: false,
+ errorStart: 0,
+ errorEnd: 0,
+});
- const { toggle, fullContentClass, fullEditorClass, fullScreenStatus } = useFullContent();
- const emit = defineEmits(['update:value', 'preview']);
- const { createMessage } = useMessage();
+const { toggle, fullContentClass, fullEditorClass, fullScreenStatus } = useFullContent();
+const emit = defineEmits(['update:value', 'preview']);
+const { createMessage } = useMessage();
- const props = defineProps({
- value: {
- type: String,
- default: '',
- },
- appId: {
- type: String as PropType<Nullable<string>>,
- },
- versionId: {
- type: String as PropType<Nullable<string>>,
- },
- suggestions: {
- type: Array as PropType<Array<{ text: string; description: string }>>,
- default: () => [],
- },
- jobType: {
- type: Number,
- },
- });
- const defaultValue = '';
+const props = defineProps({
+ value: {
+ type: String,
+ default: '',
+ },
+ appId: {
+ type: String as PropType<Nullable<string>>,
+ },
+ versionId: {
+ type: String as PropType<Nullable<string>>,
+ },
+ suggestions: {
+ type: Array as PropType<Array<{ text: string; description: string }>>,
+ default: () => [],
+ },
+});
+const defaultValue = '';
- /* verify */
- async function handleVerifySql() {
- if (isEmpty(props.value)) {
- verifyRes.errorMsg = 'empty sql';
- return false;
- }
+/* verify */
+async function handleVerifySql() {
+ if (isEmpty(props.value)) {
+ verifyRes.errorMsg = 'empty sql';
+ return false;
+ }
- if (!props.versionId) {
- createMessage.error(t('flink.app.dependencyError'));
- return false;
- } else {
- if (props.jobType === JobTypeEnum.CDC) {
- try {
- YAML.load(props.value);
- verifyRes.verified = true;
- verifyRes.errorMsg = '';
- syntaxError();
- return true;
- } catch (error) {
- verifyRes.errorStart = 0;
- verifyRes.errorEnd = 0;
- verifyRes.errorMsg = `${error.name}: ${error.reason} at line ${error.mark.line},cloumn ${error.mark.column}`;
- syntaxError();
- return false;
- }
+ if (!props.versionId) {
+ createMessage.error(t('flink.app.dependencyError'));
+ return false;
+ } else {
+ try {
+ const { data } = await fetchFlinkSqlVerify({
+ sql: props.value,
+ versionId: props.versionId,
+ });
+ const success = data.data === true || data.data === 'true';
+ if (success) {
+ verifyRes.verified = true;
+ verifyRes.errorMsg = '';
+ syntaxError();
+ return true;
} else {
- try {
- const { data } = await fetchFlinkSqlVerify({
- sql: props.value,
- versionId: props.versionId,
- });
- const success = data.data === true || data.data === 'true';
- if (success) {
- verifyRes.verified = true;
- verifyRes.errorMsg = '';
- syntaxError();
- return true;
- } else {
- verifyRes.errorStart = parseInt(data.start);
- verifyRes.errorEnd = parseInt(data.end);
- switch (data.type) {
- case 4:
- verifyRes.errorMsg = 'Unsupported sql';
- break;
- case 5:
- verifyRes.errorMsg = "SQL is not endWith ';'";
- break;
- default:
- verifyRes.errorMsg = data.message;
- break;
- }
- syntaxError();
- return false;
- }
- } catch (error) {
- console.error(error);
- return false;
+ verifyRes.errorStart = parseInt(data.start);
+ verifyRes.errorEnd = parseInt(data.end);
+ switch (data.type) {
+ case 4:
+ verifyRes.errorMsg = 'Unsupported sql';
+ break;
+ case 5:
+ verifyRes.errorMsg = "SQL is not endWith ';'";
+ break;
+ default:
+ verifyRes.errorMsg = data.message;
+ break;
}
+ syntaxError();
+ return false;
}
+ } catch (error) {
+ console.error(error);
+ return false;
}
}
+}
- async function syntaxError() {
- const editor = await getInstance();
- if (editor) {
- const model = editor.getModel();
- const monaco = await getMonacoInstance();
- if (verifyRes.errorMsg) {
- try {
- monaco.editor.setModelMarkers(model, 'sql', [
- {
- startLineNumber: verifyRes.errorStart,
- endLineNumber: verifyRes.errorEnd,
- severity: monaco.MarkerSeverity.Error,
- message: verifyRes.errorMsg,
- },
- ]);
- } catch (e) {
- console.log(e);
- }
- } else {
- monaco.editor.setModelMarkers(model, 'sql', []);
+async function syntaxError() {
+ const editor = await getInstance();
+ if (editor) {
+ const model = editor.getModel();
+ const monaco = await getMonacoInstance();
+ if (verifyRes.errorMsg) {
+ try {
+ monaco.editor.setModelMarkers(model, 'sql', [
+ {
+ startLineNumber: verifyRes.errorStart,
+ endLineNumber: verifyRes.errorEnd,
+ severity: monaco.MarkerSeverity.Error,
+ message: verifyRes.errorMsg,
+ },
+ ]);
+ } catch (e) {
+ console.log(e);
}
+ } else {
+ monaco.editor.setModelMarkers(model, 'sql', []);
}
}
- /* format */
- function handleFormatSql() {
- if (isEmpty(props.value)) return;
- if (props.jobType === JobTypeEnum.CDC) {
- return false;
- }
- const formatSql = format(props.value);
- setContent(formatSql);
- }
- /* full screen */
- // function handleBigScreen() {
- // toggle();
- // unref(flinkSql).style.width = '0';
- // setTimeout(() => {
- // unref(flinkSql).style.width = '100%';
- // unref(flinkSql).style.height = isFullscreen.value ? 'calc(100vh - 50px)' : '550px';
- // }, 500);
- // }
- const { onChange, setContent, getInstance, getMonacoInstance, setMonacoSuggest } = useMonaco(
- flinkSql,
- {
- language: 'sql',
- code: props.value || defaultValue,
- options: {
- minimap: { enabled: true },
- ...(getMonacoOptions(false) as any),
- autoClosingBrackets: 'never',
- },
+}
+/* format */
+function handleFormatSql() {
+ if (isEmpty(props.value)) return;
+ const formatSql = format(props.value);
+ setContent(formatSql);
+}
+/* full screen */
+// function handleBigScreen() {
+// toggle();
+// unref(flinkSql).style.width = '0';
+// setTimeout(() => {
+// unref(flinkSql).style.width = '100%';
+// unref(flinkSql).style.height = isFullscreen.value ? 'calc(100vh - 50px)' : '550px';
+// }, 500);
+// }
+const { onChange, setContent, getInstance, getMonacoInstance, setMonacoSuggest } = useMonaco(
+ flinkSql,
+ {
+ language: 'sql',
+ code: props.value || defaultValue,
+ options: {
+ minimap: { enabled: true },
+ ...(getMonacoOptions(false) as any),
+ autoClosingBrackets: 'never',
},
- );
+ },
+);
- watchEffect(() => {
- if (props.suggestions.length > 0) {
- setMonacoSuggest(props.suggestions);
- }
- });
- const canPreview = computed(() => {
- return /\${.+}/.test(props.value);
- });
- const flinkEditorClass = computed(() => {
- return {
- ...fullEditorClass.value,
- ['syntax-' + (verifyRes.errorMsg ? 'false' : 'true')]: true,
- };
- });
+watchEffect(() => {
+ if (props.suggestions.length > 0) {
+ setMonacoSuggest(props.suggestions);
+ }
+});
+const canPreview = computed(() => {
+ return /\${.+}/.test(props.value);
+});
+const flinkEditorClass = computed(() => {
+ return {
+ ...fullEditorClass.value,
+ ['syntax-' + (verifyRes.errorMsg ? 'false' : 'true')]: true,
+ };
+});
- onChange((data) => {
- emit('update:value', data);
- });
+onChange((data) => {
+ emit('update:value', data);
+});
- defineExpose({ handleVerifySql, setContent });
+defineExpose({ handleVerifySql, setContent });
</script>
<template>
@@ -242,13 +218,7 @@
<Icon icon="ant-design:eye-outlined" />
{{ t('flink.app.flinkSql.preview') }}
</a-button>
- <a-button
- v-if="props.jobType === JobTypeEnum.SQL"
- class="flinksql-tool-item"
- size="small"
- type="default"
- @click="handleFormatSql"
- >
+ <a-button class="flinksql-tool-item" size="small" type="default" @click="handleFormatSql">
<Icon icon="ant-design:thunderbolt-outlined" />
{{ t('flink.app.flinkSql.format') }}
</a-button>
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useAppTableAction.ts b/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useAppTableAction.ts
index e634d5311..167b63fde 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useAppTableAction.ts
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useAppTableAction.ts
@@ -314,7 +314,6 @@ export const useAppTableAction = (
{ label: 'JAR', value: JobTypeEnum.JAR },
{ label: 'SQL', value: JobTypeEnum.SQL },
{ label: 'PYFLINK', value: JobTypeEnum.PYFLINK },
- { label: 'FLINK CDC', value: JobTypeEnum.CDC },
],
onChange: handlePageDataReload.bind(null, false),
},
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useCreateAndEditSchema.ts b/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useCreateAndEditSchema.ts
index 2356d0667..62f20eab5 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useCreateAndEditSchema.ts
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useCreateAndEditSchema.ts
@@ -112,21 +112,12 @@ export const useCreateAndEditSchema = (
ifShow: ({ values }) => values?.jobType == JobTypeEnum.SQL,
rules: [{ required: true, message: t('flink.app.addAppTips.flinkSqlIsRequiredMessage') }],
},
- {
- field: 'flinkSql',
- label: 'CDC YAML',
- component: 'Input',
- slot: 'flinkSql',
- ifShow: ({ values }) => values?.jobType == JobTypeEnum.CDC,
- rules: [{ required: true, message: t('flink.app.addAppTips.flinkSqlIsRequiredMessage') }],
- },
{
field: 'teamResource',
label: t('flink.app.resource'),
component: 'Select',
render: ({ model }) => renderStreamParkResource({ model, resources: unref(teamResource) }),
- ifShow: ({ values }) =>
- values.jobType == JobTypeEnum.SQL || values.jobType == JobTypeEnum.CDC,
+ ifShow: ({ values }) => values.jobType == JobTypeEnum.SQL,
},
{
field: 'dependency',
@@ -141,8 +132,7 @@ export const useCreateAndEditSchema = (
label: t('flink.app.appConf'),
component: 'Switch',
ifShow: ({ values }) =>
- (values?.jobType == JobTypeEnum.SQL || values?.jobType == JobTypeEnum.CDC) &&
- !isK8sDeployMode(values.deployMode),
+ values?.jobType == JobTypeEnum.SQL && !isK8sDeployMode(values.deployMode),
render({ model, field }) {
return renderIsSetConfig(model, field, registerConfDrawer, openConfDrawer);
},
@@ -487,10 +477,7 @@ export const useCreateAndEditSchema = (
component: 'InputTextArea',
defaultValue: '',
slot: 'args',
- ifShow: ({ values }) =>
- edit?.mode
- ? true
- : values.jobType == JobTypeEnum.JAR && values.jobType == JobTypeEnum.PYFLINK,
+ ifShow: ({ values }) => (edit?.mode ? true : values.jobType != JobTypeEnum.SQL),
},
{
field: 'hadoopUser',
@@ -527,8 +514,6 @@ export const useCreateAndEditSchema = (
],
},
);
- } else if (model.jobType == JobTypeEnum.CDC) {
- return getAlertSvgIcon('cdc', 'Flink CDC');
} else {
return getAlertSvgIcon('fql', 'Flink SQL');
}
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useCreateSchema.ts b/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useCreateSchema.ts
index f6cca358e..59bdad48a 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useCreateSchema.ts
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useCreateSchema.ts
@@ -39,32 +39,25 @@ const getJobTypeOptions = () => {
return [
{
label: h('div', {}, [
- h(SvgIcon, { name: 'fjar', color: '#108ee9', size: 16 }, ''),
+ h(SvgIcon, { name: 'fjar', color: '#108ee9' }, ''),
h('span', { class: 'pl-10px' }, 'Flink JAR'),
]),
value: String(JobTypeEnum.JAR),
},
{
label: h('div', {}, [
- h(SvgIcon, { name: 'fql', color: '#108ee9', size: 16 }, ''),
+ h(SvgIcon, { name: 'fql', color: '#108ee9' }, ''),
h('span', { class: 'pl-10px' }, 'Flink SQL'),
]),
value: String(JobTypeEnum.SQL),
},
{
label: h('div', {}, [
- h(SvgIcon, { name: 'py', color: '#108ee9', size: 16 }, ''),
+ h(SvgIcon, { name: 'py', color: '#108ee9' }, ''),
h('span', { class: 'pl-10px' }, 'Python Flink'),
]),
value: String(JobTypeEnum.PYFLINK),
},
- {
- label: h('div', {}, [
- h(SvgIcon, { name: 'cdc', color: '#108ee9', size: 16 }, ''),
- h('span', { class: 'pl-10px' }, 'Flink CDC'),
- ]),
- value: String(JobTypeEnum.CDC),
- },
];
};
@@ -115,7 +108,7 @@ export const useCreateSchema = (dependencyRef: Ref) => {
placeholder: t('flink.app.addAppTips.jobTypePlaceholder'),
options: getJobTypeOptions(),
onChange: (value) => {
- if (value != JobTypeEnum.SQL && value != JobTypeEnum.CDC) {
+ if (value != JobTypeEnum.SQL) {
formModel.resourceFrom = String(ResourceFromEnum.PROJECT);
}
},
@@ -133,7 +126,7 @@ export const useCreateSchema = (dependencyRef: Ref) => {
component: 'Select',
render: ({ model }) => renderResourceFrom(model),
rules: [{ required: true, message: t('flink.app.addAppTips.resourceFromMessage') }],
- show: ({ values }) => values?.jobType == JobTypeEnum.JAR,
+ show: ({ values }) => values?.jobType != JobTypeEnum.SQL,
},
{
field: 'uploadJobJar',
@@ -141,7 +134,7 @@ export const useCreateSchema = (dependencyRef: Ref) => {
component: 'Select',
render: ({ model }) => renderStreamParkJarApp({ model, resources: unref(teamResource) }),
ifShow: ({ values }) =>
- values?.jobType == JobTypeEnum.JAR && values?.resourceFrom == ResourceFromEnum.UPLOAD,
+ values?.jobType != JobTypeEnum.SQL && values?.resourceFrom == ResourceFromEnum.UPLOAD,
},
{
field: 'mainClass',
@@ -149,7 +142,7 @@ export const useCreateSchema = (dependencyRef: Ref) => {
component: 'Input',
componentProps: { placeholder: t('flink.app.addAppTips.mainClassPlaceholder') },
ifShow: ({ values }) =>
- values?.jobType == JobTypeEnum.JAR && values?.resourceFrom == ResourceFromEnum.UPLOAD,
+ values?.jobType != JobTypeEnum.SQL && values?.resourceFrom == ResourceFromEnum.UPLOAD,
rules: [{ required: true, message: t('flink.app.addAppTips.mainClassIsRequiredMessage') }],
},
{
@@ -177,7 +170,7 @@ export const useCreateSchema = (dependencyRef: Ref) => {
},
},
ifShow: ({ values }) =>
- values?.jobType == JobTypeEnum.JAR && values.resourceFrom != ResourceFromEnum.UPLOAD,
+ values?.jobType != JobTypeEnum.SQL && values.resourceFrom != ResourceFromEnum.UPLOAD,
rules: [{ required: true, message: t('flink.app.addAppTips.projectIsRequiredMessage') }],
},
{
@@ -201,7 +194,7 @@ export const useCreateSchema = (dependencyRef: Ref) => {
};
},
ifShow: ({ values }) =>
- values?.jobType == JobTypeEnum.JAR && values?.resourceFrom != ResourceFromEnum.UPLOAD,
+ values?.jobType != JobTypeEnum.SQL && values?.resourceFrom != ResourceFromEnum.UPLOAD,
rules: [{ required: true, message: t('flink.app.addAppTips.projectIsRequiredMessage') }],
},
{
@@ -232,7 +225,7 @@ export const useCreateSchema = (dependencyRef: Ref) => {
};
},
ifShow: ({ values }) =>
- values?.jobType == JobTypeEnum.JAR && values?.resourceFrom != ResourceFromEnum.UPLOAD,
+ values?.jobType != JobTypeEnum.SQL && values?.resourceFrom != ResourceFromEnum.UPLOAD,
dynamicRules: () => [
{ required: true, message: t('flink.app.addAppTips.appTypeIsRequiredMessage') },
],
@@ -257,7 +250,7 @@ export const useCreateSchema = (dependencyRef: Ref) => {
};
},
ifShow: ({ values }) =>
- values?.jobType == JobTypeEnum.JAR &&
+ values?.jobType != JobTypeEnum.SQL &&
values?.resourceFrom != ResourceFromEnum.UPLOAD &&
values.appType == String(AppTypeEnum.APACHE_FLINK),
rules: [{ required: true, message: t('flink.app.addAppTips.programJarIsRequiredMessage') }],
@@ -268,7 +261,7 @@ export const useCreateSchema = (dependencyRef: Ref) => {
component: 'Input',
componentProps: { placeholder: t('flink.app.addAppTips.mainClassPlaceholder') },
ifShow: ({ values }) =>
- (values?.jobType == JobTypeEnum.JAR || values?.jobType == JobTypeEnum.PYFLINK) &&
+ values?.jobType != JobTypeEnum.SQL &&
values?.resourceFrom != ResourceFromEnum.UPLOAD &&
values.appType == String(AppTypeEnum.APACHE_FLINK),
rules: [{ required: true, message: t('flink.app.addAppTips.mainClassIsRequiredMessage') }],
@@ -295,7 +288,7 @@ export const useCreateSchema = (dependencyRef: Ref) => {
};
},
ifShow: ({ values }) =>
- values?.jobType == JobTypeEnum.JAR &&
+ values?.jobType != JobTypeEnum.SQL &&
values?.resourceFrom != ResourceFromEnum.UPLOAD &&
values.appType == String(AppTypeEnum.STREAMPARK_FLINK),
dynamicRules: () => [{ required: true, validator: handleCheckConfig }],
diff --git a/streampark-flink/pom.xml b/streampark-flink/pom.xml
index 44f95abaa..41f938c3c 100644
--- a/streampark-flink/pom.xml
+++ b/streampark-flink/pom.xml
@@ -37,7 +37,6 @@
<module>streampark-flink-proxy</module>
<module>streampark-flink-packer</module>
<module>streampark-flink-kubernetes</module>
- <module>streampark-flink-cdcclient</module>
</modules>
<dependencies>
diff --git a/streampark-flink/streampark-flink-cdcclient/pom.xml b/streampark-flink/streampark-flink-cdcclient/pom.xml
deleted file mode 100644
index ca7db4b0e..000000000
--- a/streampark-flink/streampark-flink-cdcclient/pom.xml
+++ /dev/null
@@ -1,140 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
- ~ Licensed to the Apache Software Foundation (ASF) under one or more
- ~ contributor license agreements. See the NOTICE file distributed with
- ~ this work for additional information regarding copyright ownership.
- ~ The ASF licenses this file to You under the Apache License, Version 2.0
- ~ (the "License"); you may not use this file except in compliance with
- ~ the License. You may obtain a copy of the License at
- ~
- ~ http://www.apache.org/licenses/LICENSE-2.0
- ~
- ~ Unless required by applicable law or agreed to in writing, software
- ~ distributed under the License is distributed on an "AS IS" BASIS,
- ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- ~ See the License for the specific language governing permissions and
- ~ limitations under the License.
- -->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <modelVersion>4.0.0</modelVersion>
- <parent>
- <groupId>org.apache.streampark</groupId>
- <artifactId>streampark-flink</artifactId>
- <version>2.2.0-SNAPSHOT</version>
- </parent>
-
- <artifactId>streampark-flink-cdcclient_${scala.binary.version}</artifactId>
- <version>2.2.0-SNAPSHOT</version>
- <name>StreamPark : Flink CDC Client</name>
-
- <properties>
- <flink.cdc.version>3.2.1</flink.cdc.version>
- </properties>
-
- <dependencies>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-cdc-common</artifactId>
- <version>${flink.cdc.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-cdc-runtime</artifactId>
- <version>${flink.cdc.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-cdc-cli</artifactId>
- <version>${flink.cdc.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-cdc-composer</artifactId>
- <version>${flink.cdc.version}</version>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-java</artifactId>
- <version>${flink.version}</version>
- <scope>provided</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-runtime</artifactId>
- <version>${flink.version}</version>
- <scope>provided</scope>
- </dependency>
-
- <dependency>
- <groupId>org.apache.streampark</groupId>
- <artifactId>streampark-common_${scala.binary.version}</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-yarn</artifactId>
- <version>1.18.1</version>
- <scope>provided</scope>
- </dependency>
- <dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-api</artifactId>
- </dependency>
-
- </dependencies>
-
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-shade-plugin</artifactId>
- <executions>
- <execution>
- <id>shade-dist</id>
- <goals>
- <goal>shade</goal>
- </goals>
- <phase>package</phase>
- <configuration>
- <filters>
- <filter>
- <artifact>*:*</artifact>
- <excludes>
- <exclude>META-INF/*.SF</exclude>
- <exclude>META-INF/*.DSA</exclude>
- <exclude>META-INF/*.RSA</exclude>
- </excludes>
- </filter>
- </filters>
- <shadeTestJar>false</shadeTestJar>
- <createDependencyReducedPom>false</createDependencyReducedPom>
- <shadedArtifactAttached>false</shadedArtifactAttached>
- <artifactSet>
- <includes>
- <include>*</include>
- </includes>
- </artifactSet>
- <relocations>
- <relocation>
- <pattern>org.apache.commons.cli</pattern>
- <shadedPattern>org.apache.flink.cdc.shaded.com.apache.commons.cli</shadedPattern>
- </relocation>
- <relocation>
- <pattern>org.apache.calcite</pattern>
- <shadedPattern>org.apache.flink.cdc.shaded.org.apache.calcite</shadedPattern>
- </relocation>
- </relocations>
- <transformers>
- <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
- <mainClass>org.apache.streampark.flink.cdc.cli.CDCClient</mainClass>
- </transformer>
- </transformers>
- </configuration>
- </execution>
- </executions>
- </plugin>
- </plugins>
- </build>
-
-</project>
diff --git a/streampark-flink/streampark-flink-cdcclient/src/main/java/org/apache/streampark/flink/cdc/cli/CDCClient.java b/streampark-flink/streampark-flink-cdcclient/src/main/java/org/apache/streampark/flink/cdc/cli/CDCClient.java
deleted file mode 100644
index 379359548..000000000
--- a/streampark-flink/streampark-flink-cdcclient/src/main/java/org/apache/streampark/flink/cdc/cli/CDCClient.java
+++ /dev/null
@@ -1,80 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.streampark.flink.cdc.cli;
-
-import org.apache.streampark.common.conf.ConfigKeys;
-import org.apache.streampark.common.util.DeflaterUtils;
-import org.apache.streampark.common.util.PropertiesUtils;
-
-import org.apache.flink.api.java.utils.ParameterTool;
-import org.apache.flink.cdc.common.configuration.Configuration;
-import org.apache.flink.cdc.composer.PipelineExecution;
-import org.apache.flink.configuration.CoreOptions;
-import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
-import org.apache.flink.util.StringUtils;
-import org.apache.flink.yarn.configuration.YarnConfigOptions;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.Map;
-
-/**
- * cdc client
- */
-public class CDCClient {
-
- private static final Logger LOG = LoggerFactory.getLogger(CDCClient.class);
-
- public static void main(String[] args) throws Exception {
- ParameterTool parameter = ParameterTool.fromArgs(args);
- Map<String, String> configMap = new HashMap<>();
- String cdcYamlDecode = parameter.get(ConfigKeys.KEY_FLINK_SQL(null));
- String appNameDecode = parameter.get(ConfigKeys.KEY_APP_NAME(null));
- String flinkConfigDecode = parameter.get(ConfigKeys.KEY_FLINK_CONF(null));
- String parallelism = parameter.get(ConfigKeys.KEY_FLINK_PARALLELISM(null));
- if (StringUtils.isNullOrWhitespaceOnly(cdcYamlDecode)
- || StringUtils.isNullOrWhitespaceOnly(appNameDecode)
- || StringUtils.isNullOrWhitespaceOnly(flinkConfigDecode)) {
- LOG.error("--flink.conf or --app.name or `cdc yaml` must not be null.");
- return;
- }
-
- String cdcYaml = DeflaterUtils.unzipString(cdcYamlDecode);
- String appName = DeflaterUtils.unzipString(appNameDecode);
- String flinkConfigString = DeflaterUtils.unzipString(flinkConfigDecode);
- configMap.putAll(PropertiesUtils.fromYamlTextAsJava(flinkConfigString));
- configMap.put(YarnConfigOptions.APPLICATION_NAME.key(), appName);
- configMap.put(CoreOptions.DEFAULT_PARALLELISM.key(), parallelism);
- Configuration flinkConfig = Configuration.fromMap(configMap);
- LOG.debug("Flink cdc config {}", flinkConfig);
- LOG.debug("Flink cdc yaml {}", cdcYaml);
- PipelineExecution.ExecutionInfo result =
- new CDCExecutor(cdcYaml, flinkConfig, new ArrayList<>(), SavepointRestoreSettings.none()).run();
- printExecutionInfo(result);
-
- }
-
- private static void printExecutionInfo(PipelineExecution.ExecutionInfo info) {
- System.out.println("Pipeline has been submitted to cluster.");
- System.out.printf("Job ID: %s\n", info.getId());
- System.out.printf("Job Description: %s\n", info.getDescription());
- }
-}
diff --git a/streampark-flink/streampark-flink-cdcclient/src/main/java/org/apache/streampark/flink/cdc/cli/CDCExecutor.java b/streampark-flink/streampark-flink-cdcclient/src/main/java/org/apache/streampark/flink/cdc/cli/CDCExecutor.java
deleted file mode 100644
index 78a46d7a8..000000000
--- a/streampark-flink/streampark-flink-cdcclient/src/main/java/org/apache/streampark/flink/cdc/cli/CDCExecutor.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.streampark.flink.cdc.cli;
-
-import org.apache.flink.cdc.cli.parser.PipelineDefinitionParser;
-import org.apache.flink.cdc.cli.parser.YamlPipelineDefinitionParser;
-import org.apache.flink.cdc.cli.utils.FlinkEnvironmentUtils;
-import org.apache.flink.cdc.common.configuration.Configuration;
-import org.apache.flink.cdc.composer.PipelineComposer;
-import org.apache.flink.cdc.composer.PipelineExecution;
-import org.apache.flink.cdc.composer.definition.PipelineDef;
-import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
-
-import java.nio.file.Path;
-import java.util.List;
-
-/**
- * cdc executor
- */
-public class CDCExecutor {
-
- private final String pipelineString;
- private final Configuration configuration;
- private final SavepointRestoreSettings savePointSettings;
- private final List<Path> additionalJar;
-
- private PipelineComposer composer;
-
- public CDCExecutor(String pipelineString,
- Configuration flinkConfig,
- List<Path> additionalJar,
- SavepointRestoreSettings savePointRestoreSettings) {
- this.pipelineString = pipelineString;
- this.configuration = flinkConfig;
- this.additionalJar = additionalJar;
- this.savePointSettings = savePointRestoreSettings;
- }
-
- public PipelineExecution.ExecutionInfo run() throws Exception {
- PipelineDefinitionParser pipelineDefinitionParser = new YamlPipelineDefinitionParser();
- PipelineDef pipelineDef = pipelineDefinitionParser.parse(pipelineString, configuration);
- PipelineComposer composer = getComposer();
- PipelineExecution execution = composer.compose(pipelineDef);
- return execution.execute();
- }
-
- private PipelineComposer getComposer() throws Exception {
- if (composer == null) {
- return FlinkEnvironmentUtils.createComposer(
- true, configuration, additionalJar, savePointSettings);
- }
- return composer;
- }
-}
diff --git a/streampark-flink/streampark-flink-cdcclient/src/test/java/org/apache/streampark/flink/cdc/cli/package-info.java b/streampark-flink/streampark-flink-cdcclient/src/test/java/org/apache/streampark/flink/cdc/cli/package-info.java
deleted file mode 100644
index b85a06e75..000000000
--- a/streampark-flink/streampark-flink-cdcclient/src/test/java/org/apache/streampark/flink/cdc/cli/package-info.java
+++ /dev/null
@@ -1,18 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.streampark.flink.cdc.cli;
diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/SubmitRequest.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/SubmitRequest.scala
index afc99b928..007aa217b 100644
--- a/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/SubmitRequest.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/SubmitRequest.scala
@@ -68,7 +68,6 @@ case class SubmitRequest(
case FlinkJobType.FLINK_SQL =>
Constants.STREAMPARK_FLINKSQL_CLIENT_CLASS
case FlinkJobType.PYFLINK => Constants.PYTHON_FLINK_DRIVER_CLASS_NAME
- case FlinkJobType.FLINK_CDC => Constants.STREAMPARK_FLINKCDC_CLIENT_CLASS
case _ => appProperties(KEY_FLINK_APPLICATION_MAIN_CLASS)
}
diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnApplicationClient.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnApplicationClient.scala
index 0dc6cb813..966325282 100644
--- a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnApplicationClient.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnApplicationClient.scala
@@ -52,7 +52,7 @@ object YarnApplicationClient extends YarnClientTrait {
submitRequest.hdfsWorkspace.flinkPlugins,
submitRequest.hdfsWorkspace.appJars)
submitRequest.jobType match {
- case FlinkJobType.FLINK_SQL | FlinkJobType.FLINK_CDC =>
+ case FlinkJobType.FLINK_SQL =>
array += s"${workspace.APP_SHIMS}/flink-${submitRequest.flinkVersion.majorVersion}"
val jobLib = s"${workspace.APP_WORKSPACE}/${submitRequest.id}/lib"
if (HdfsUtils.exists(jobLib)) {
diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/FlinkClientTrait.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/FlinkClientTrait.scala
index ca1327927..0c0b7c84c 100644
--- a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/FlinkClientTrait.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/FlinkClientTrait.scala
@@ -478,7 +478,7 @@ trait FlinkClientTrait extends Logger {
programArgs += PARAM_KEY_FLINK_PARALLELISM += getParallelism(submitRequest).toString
submitRequest.jobType match {
- case FlinkJobType.FLINK_SQL | FlinkJobType.FLINK_CDC =>
+ case FlinkJobType.FLINK_SQL =>
programArgs += PARAM_KEY_FLINK_SQL += submitRequest.flinkSQL
if (submitRequest.appConf != null) {
programArgs += PARAM_KEY_APP_CONF += submitRequest.appConf
diff --git a/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/impl/FlinkYarnApplicationBuildPipeline.scala b/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/impl/FlinkYarnApplicationBuildPipeline.scala
index 3f5fc9f7d..4fdd0766e 100644
--- a/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/impl/FlinkYarnApplicationBuildPipeline.scala
+++ b/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/impl/FlinkYarnApplicationBuildPipeline.scala
@@ -46,7 +46,7 @@ class FlinkYarnApplicationBuildPipeline(request: FlinkYarnApplicationBuildReques
override protected def buildProcess(): SimpleBuildResponse = {
execStep(1) {
request.flinkJobType match {
- case FlinkJobType.FLINK_SQL | FlinkJobType.FLINK_CDC | FlinkJobType.PYFLINK =>
+ case FlinkJobType.FLINK_SQL | FlinkJobType.PYFLINK =>
LfsOperator.mkCleanDirs(request.localWorkspace)
HdfsOperator.mkCleanDirs(request.yarnProvidedPath)
case _ =>
diff --git a/tools/dependencies/known-dependencies.txt b/tools/dependencies/known-dependencies.txt
index 9d4e93b54..a51440d69 100644
--- a/tools/dependencies/known-dependencies.txt
+++ b/tools/dependencies/known-dependencies.txt
@@ -12,7 +12,6 @@ aether-transport-http-1.1.0.jar
aether-util-1.1.0.jar
aircompressor-0.10.jar
antlr4-runtime-4.8-1.jar
-apiguardian-api-1.1.2.jar
aopalliance-repackaged-2.6.1.jar
arrow-format-2.0.0.jar
arrow-memory-core-2.0.0.jar
@@ -24,7 +23,7 @@ asm-commons-8.0.jar
asm-tree-8.0.jar
asm-util-8.0.jar
aspectjweaver-1.9.7.jar
-avatica-core-1.22.0.jar
+async-profiler-2.9.jar
audience-annotations-0.5.0.jar
avro-1.8.2.jar
avro-ipc-1.8.2.jar
@@ -33,8 +32,6 @@ bcpkix-jdk18on-1.76.jar
bcprov-jdk18on-1.76.jar
bcutil-jdk18on-1.76.jar
buji-pac4j-5.0.1.jar
-calcite-core-1.32.0.jar
-calcite-linq4j-1.32.0.jar
caffeine-2.8.6.jar
checker-qual-3.5.0.jar
chill-java-0.9.5.jar
@@ -72,10 +69,6 @@ enumeratum_2.12-1.6.1.jar
failureaccess-1.0.1.jar
flatbuffers-java-1.9.0.jar
flink-annotations-1.20.1.jar
-flink-cdc-cli-3.2.1.jar
-flink-cdc-common-3.2.1.jar
-flink-cdc-composer-3.2.1.jar
-flink-cdc-runtime-3.2.1.jar
flink-clients-1.20.1.jar
flink-connector-datagen-1.20.1.jar
flink-core-1.20.1.jar
@@ -171,7 +164,6 @@ jna-5.13.0.jar
jsr305-1.3.9.jar
jsch-0.2.11.jar
json-smart-2.4.10.jar
-json-path-2.7.0.jar
json4s-ast_2.12-3.7.0-M11.jar
json4s-core_2.12-3.7.0-M11.jar
json4s-jackson_2.12-3.7.0-M11.jar
@@ -179,7 +171,6 @@ json4s-scalap_2.12-3.7.0-M11.jar
jsp-api-2.1.jar
jsqlparser-4.4.jar
jul-to-slf4j-1.7.36.jar
-jts-core-1.19.0.jar
kafka-clients-3.1.2.jar
kerb-admin-1.0.1.jar
kerb-client-1.0.1.jar
@@ -264,6 +255,7 @@ netty-resolver-dns-native-macos-4.1.91.Final-osx-x86_64.jar
netty-transport-4.1.91.Final.jar
netty-transport-classes-epoll-4.1.91.Final.jar
netty-transport-classes-kqueue-4.1.91.Final.jar
+netty-transport-native-epoll-4.1.91.Final.jar
netty-transport-native-epoll-4.1.91.Final-linux-aarch_64.jar
netty-transport-native-epoll-4.1.91.Final-linux-x86_64.jar
netty-transport-native-kqueue-4.1.91.Final-osx-aarch_64.jar
@@ -393,7 +385,5 @@ xnio-api-3.8.7.Final.jar
xnio-nio-3.8.7.Final.jar
xz-1.5.jar
zookeeper-3.6.3.jar
-icu4j-67.1.jar
zookeeper-jute-3.6.3.jar
-async-profiler-2.9.jar
-netty-transport-native-epoll-4.1.91.Final.jar
+icu4j-67.1.jar