You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by ni...@apache.org on 2024/07/10 11:25:48 UTC
(paimon-webui) 02/11: [Improvement] Support SessionController and JobController obtaining cluster through deployment mode (#484)
This is an automated email from the ASF dual-hosted git repository.
nicholasjiang pushed a commit to branch branch-0.1
in repository https://gitbox.apache.org/repos/asf/paimon-webui.git
commit 280419e333405e242143ade9ec2783187d247404
Author: s7monk <34...@users.noreply.github.com>
AuthorDate: Tue Jul 9 19:46:58 2024 +0800
[Improvement] Support SessionController and JobController obtaining cluster through deployment mode (#484)
---
paimon-web-server/pom.xml | 8 ++++++++
.../apache/paimon/web/server/controller/SessionController.java | 5 +++--
.../org/apache/paimon/web/server/service/impl/JobServiceImpl.java | 5 +++--
.../apache/paimon/web/server/controller/JobControllerTest.java | 1 +
.../paimon/web/server/controller/SessionControllerTest.java | 1 +
5 files changed, 16 insertions(+), 4 deletions(-)
diff --git a/paimon-web-server/pom.xml b/paimon-web-server/pom.xml
index ac352377..2d85244d 100644
--- a/paimon-web-server/pom.xml
+++ b/paimon-web-server/pom.xml
@@ -37,6 +37,7 @@ under the License.
<properties>
<hadoop.version>2.8.5</hadoop.version>
<flink.version>1.18.1</flink.version>
+ <scala.version>2.12</scala.version>
</properties>
<dependencies>
@@ -247,6 +248,13 @@ under the License.
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-planner_${scala.version}</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
+
<dependency>
<groupId>commons-cli</groupId>
<artifactId>commons-cli</artifactId>
diff --git a/paimon-web-server/src/main/java/org/apache/paimon/web/server/controller/SessionController.java b/paimon-web-server/src/main/java/org/apache/paimon/web/server/controller/SessionController.java
index cee6f61a..fc761f3e 100644
--- a/paimon-web-server/src/main/java/org/apache/paimon/web/server/controller/SessionController.java
+++ b/paimon-web-server/src/main/java/org/apache/paimon/web/server/controller/SessionController.java
@@ -18,6 +18,7 @@
package org.apache.paimon.web.server.controller;
+import org.apache.paimon.web.gateway.enums.DeploymentMode;
import org.apache.paimon.web.server.data.dto.SessionDTO;
import org.apache.paimon.web.server.data.model.ClusterInfo;
import org.apache.paimon.web.server.data.result.R;
@@ -59,7 +60,7 @@ public class SessionController {
}
int uid = StpUtil.getLoginIdAsInt();
QueryWrapper<ClusterInfo> queryWrapper = new QueryWrapper<>();
- queryWrapper.eq("type", "Flink");
+ queryWrapper.eq("deployment_mode", DeploymentMode.FLINK_SQL_GATEWAY.getType());
List<ClusterInfo> clusterInfos = clusterService.list(queryWrapper);
for (ClusterInfo cluster : clusterInfos) {
SessionDTO sessionDTO = new SessionDTO();
@@ -85,7 +86,7 @@ public class SessionController {
}
int uid = StpUtil.getLoginIdAsInt();
QueryWrapper<ClusterInfo> queryWrapper = new QueryWrapper<>();
- queryWrapper.eq("type", "Flink");
+ queryWrapper.eq("deployment_mode", DeploymentMode.FLINK_SQL_GATEWAY.getType());
List<ClusterInfo> clusterInfos = clusterService.list(queryWrapper);
for (ClusterInfo cluster : clusterInfos) {
SessionDTO sessionDTO = new SessionDTO();
diff --git a/paimon-web-server/src/main/java/org/apache/paimon/web/server/service/impl/JobServiceImpl.java b/paimon-web-server/src/main/java/org/apache/paimon/web/server/service/impl/JobServiceImpl.java
index e69c4518..ceb46c73 100644
--- a/paimon-web-server/src/main/java/org/apache/paimon/web/server/service/impl/JobServiceImpl.java
+++ b/paimon-web-server/src/main/java/org/apache/paimon/web/server/service/impl/JobServiceImpl.java
@@ -25,6 +25,7 @@ import org.apache.paimon.web.engine.flink.common.result.FetchResultParams;
import org.apache.paimon.web.engine.flink.common.status.JobStatus;
import org.apache.paimon.web.engine.flink.sql.gateway.model.SessionEntity;
import org.apache.paimon.web.gateway.config.ExecutionConfig;
+import org.apache.paimon.web.gateway.enums.DeploymentMode;
import org.apache.paimon.web.gateway.enums.EngineType;
import org.apache.paimon.web.gateway.provider.ExecutorFactoryProvider;
import org.apache.paimon.web.server.context.LogContextHolder;
@@ -302,8 +303,8 @@ public class JobServiceImpl extends ServiceImpl<JobMapper, JobInfo> implements J
if (taskType.equals("Flink")) {
QueryWrapper<ClusterInfo> queryWrapper = new QueryWrapper<>();
- queryWrapper.eq("type", "Flink");
- List<ClusterInfo> clusters = clusterService.list();
+ queryWrapper.eq("deployment_mode", DeploymentMode.FLINK_SQL_GATEWAY.getType());
+ List<ClusterInfo> clusters = clusterService.list(queryWrapper);
for (ClusterInfo cluster : clusters) {
try {
SessionEntity session =
diff --git a/paimon-web-server/src/test/java/org/apache/paimon/web/server/controller/JobControllerTest.java b/paimon-web-server/src/test/java/org/apache/paimon/web/server/controller/JobControllerTest.java
index eeeafdc4..42155fa6 100644
--- a/paimon-web-server/src/test/java/org/apache/paimon/web/server/controller/JobControllerTest.java
+++ b/paimon-web-server/src/test/java/org/apache/paimon/web/server/controller/JobControllerTest.java
@@ -113,6 +113,7 @@ public class JobControllerTest extends FlinkSQLGatewayTestBase {
.port(port)
.enabled(true)
.type("Flink")
+ .deploymentMode("flink-sql-gateway")
.build();
boolean res = clusterService.save(cluster);
assertTrue(res);
diff --git a/paimon-web-server/src/test/java/org/apache/paimon/web/server/controller/SessionControllerTest.java b/paimon-web-server/src/test/java/org/apache/paimon/web/server/controller/SessionControllerTest.java
index e49c0e6f..9833e728 100644
--- a/paimon-web-server/src/test/java/org/apache/paimon/web/server/controller/SessionControllerTest.java
+++ b/paimon-web-server/src/test/java/org/apache/paimon/web/server/controller/SessionControllerTest.java
@@ -123,6 +123,7 @@ public class SessionControllerTest extends FlinkSQLGatewayTestBase {
.port(port)
.enabled(true)
.type("Flink")
+ .deploymentMode("flink-sql-gateway")
.build();
boolean res = clusterService.save(cluster);
assertTrue(res);