This is an automated email from the ASF dual-hosted git repository.
nicholasjiang pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-webui.git
The following commit(s) were added to refs/heads/main by this push:
new 098e6f3 [Feature] Introduce Session Interface (#209)
098e6f3 is described below
commit 098e6f3421556b7f5f6f9236300cdbbc7804a17a
Author: s7monk <[email protected]>
AuthorDate: Mon May 13 11:45:13 2024 +0800
[Feature] Introduce Session Interface (#209)
---
paimon-web-server/pom.xml | 56 ++++++
.../paimon/web/server/configrue/RetryConfig.java | 27 +++
.../web/server/controller/SessionController.java | 90 +++++++++
.../paimon/web/server/data/dto/SessionDTO.java | 34 ++++
.../paimon/web/server/service/SessionService.java | 57 ++++++
.../web/server/service/UserSessionManager.java | 50 +++++
.../server/service/impl/SessionServiceImpl.java | 105 +++++++++++
.../server/controller/FlinkSQLGatewayTestBase.java | 201 +++++++++++++++++++++
.../server/controller/SessionControllerTest.java | 175 ++++++++++++++++++
9 files changed, 795 insertions(+)
diff --git a/paimon-web-server/pom.xml b/paimon-web-server/pom.xml
index 5aee2a6..ebd3c7f 100644
--- a/paimon-web-server/pom.xml
+++ b/paimon-web-server/pom.xml
@@ -36,6 +36,7 @@ under the License.
<properties>
<hadoop.version>2.8.5</hadoop.version>
+ <flink.version>1.18.1</flink.version>
</properties>
<dependencies>
@@ -51,6 +52,12 @@ under the License.
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-web-gateway</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
<dependency>
<groupId>cn.hutool</groupId>
<artifactId>hutool-all</artifactId>
@@ -154,6 +161,16 @@ under the License.
<artifactId>mybatis-plus-boot-starter</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.springframework.retry</groupId>
+ <artifactId>spring-retry</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-aspects</artifactId>
+ </dependency>
+
<dependency>
<groupId>cn.dev33</groupId>
<artifactId>sa-token-spring-boot-starter</artifactId>
@@ -172,20 +189,59 @@ under the License.
<artifactId>hadoop-client</artifactId>
<version>${hadoop.version}</version>
<scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>commons-cli</groupId>
+ <artifactId>commons-cli</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
</dependency>
+
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
+
<dependency>
<groupId>org.springdoc</groupId>
<artifactId>springdoc-openapi-ui</artifactId>
</dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-test-utils</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <artifactId>log4j-slf4j-impl</artifactId>
+ <groupId>org.apache.logging.log4j</groupId>
+ </exclusion>
+ <exclusion>
+ <artifactId>log4j-core</artifactId>
+ <groupId>org.apache.logging.log4j</groupId>
+ </exclusion>
+ <exclusion>
+ <artifactId>log4j-api</artifactId>
+ <groupId>org.apache.logging.log4j</groupId>
+ </exclusion>
+ <exclusion>
+ <groupId>commons-cli</groupId>
+ <artifactId>commons-cli</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>commons-cli</groupId>
+ <artifactId>commons-cli</artifactId>
+ <version>1.4</version>
+ </dependency>
</dependencies>
<build>
diff --git
a/paimon-web-server/src/main/java/org/apache/paimon/web/server/configrue/RetryConfig.java
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/configrue/RetryConfig.java
new file mode 100644
index 0000000..82ab4a6
--- /dev/null
+++
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/configrue/RetryConfig.java
@@ -0,0 +1,27 @@
+/*
+ * 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.paimon.web.server.configrue;
+
+import org.springframework.context.annotation.Configuration;
+import org.springframework.retry.annotation.EnableRetry;
+
+/** RetryConfig. */
+@EnableRetry
+@Configuration
+public class RetryConfig {}
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
new file mode 100644
index 0000000..ff2f641
--- /dev/null
+++
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/controller/SessionController.java
@@ -0,0 +1,90 @@
+/*
+ * 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.paimon.web.server.controller;
+
+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;
+import org.apache.paimon.web.server.service.ClusterService;
+import org.apache.paimon.web.server.service.SessionService;
+
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.retry.annotation.Backoff;
+import org.springframework.retry.annotation.Recover;
+import org.springframework.retry.annotation.Retryable;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.List;
+
+/** Session api controller. */
+@Slf4j
+@RestController
+@RequestMapping("/api/session")
+public class SessionController {
+
+ @Autowired private SessionService sessionService;
+
+ @Autowired private ClusterService clusterService;
+
+ @PostMapping("/create")
+ @Retryable(
+ value = {Exception.class},
+ maxAttempts = 3,
+ backoff = @Backoff(delay = 3000))
+ public R<Void> createSession(Integer uid) {
+ QueryWrapper<ClusterInfo> queryWrapper = new QueryWrapper<>();
+ queryWrapper.eq("type", "Flink");
+ List<ClusterInfo> clusterInfos = clusterService.list(queryWrapper);
+ for (ClusterInfo cluster : clusterInfos) {
+ SessionDTO sessionDTO = new SessionDTO();
+ sessionDTO.setHost(cluster.getHost());
+ sessionDTO.setPort(cluster.getPort());
+ sessionDTO.setClusterId(cluster.getId());
+ sessionDTO.setUid(uid);
+ sessionService.createSession(sessionDTO);
+ }
+ return R.succeed();
+ }
+
+ @Recover
+ public R<Void> recover(Exception e, Integer uid) {
+ log.error("After retries failed to create session for UID: {}", uid,
e);
+ return R.failed();
+ }
+
+ @PostMapping("/drop")
+ public R<Void> dropSession(Integer uid) {
+ QueryWrapper<ClusterInfo> queryWrapper = new QueryWrapper<>();
+ queryWrapper.eq("type", "Flink");
+ List<ClusterInfo> clusterInfos = clusterService.list(queryWrapper);
+ for (ClusterInfo cluster : clusterInfos) {
+ SessionDTO sessionDTO = new SessionDTO();
+ sessionDTO.setHost(cluster.getHost());
+ sessionDTO.setPort(cluster.getPort());
+ sessionDTO.setClusterId(cluster.getId());
+ sessionDTO.setUid(uid);
+ sessionService.closeSession(sessionDTO);
+ }
+ return R.succeed();
+ }
+}
diff --git
a/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/dto/SessionDTO.java
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/dto/SessionDTO.java
new file mode 100644
index 0000000..1399c60
--- /dev/null
+++
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/data/dto/SessionDTO.java
@@ -0,0 +1,34 @@
+/*
+ * 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.paimon.web.server.data.dto;
+
+import lombok.Data;
+
+/** DTO of session. */
+@Data
+public class SessionDTO {
+
+ private Integer uid;
+
+ private Integer clusterId;
+
+ private String host;
+
+ private Integer port;
+}
diff --git
a/paimon-web-server/src/main/java/org/apache/paimon/web/server/service/SessionService.java
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/service/SessionService.java
new file mode 100644
index 0000000..4204e87
--- /dev/null
+++
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/service/SessionService.java
@@ -0,0 +1,57 @@
+/*
+ * 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.paimon.web.server.service;
+
+import org.apache.paimon.web.engine.flink.sql.gateway.model.SessionEntity;
+import org.apache.paimon.web.server.data.dto.SessionDTO;
+
+/** Session Service. */
+public interface SessionService {
+
+ /**
+ * Creates a new session.
+ *
+ * @param sessionDTO the data transfer object containing session details
+ */
+ void createSession(SessionDTO sessionDTO);
+
+ /**
+ * Closes an existing session.
+ *
+ * @param sessionDTO the data transfer object containing session details
+ */
+ void closeSession(SessionDTO sessionDTO);
+
+ /**
+ * Triggers a heartbeat update for a session.
+ *
+ * @param sessionDTO the data transfer object containing session details
+ * @return the status code after triggering the heartbeat
+ */
+ int triggerSessionHeartbeat(SessionDTO sessionDTO);
+
+ /**
+ * Retrieves the session for a given user ID within a specified cluster.
+ *
+ * @param uid the unique identifier of the user
+ * @param clusterId the identifier of the cluster
+ * @return the SessionEntity for the specified user and cluster, or null
if no session is found
+ */
+ SessionEntity getSession(Integer uid, Integer clusterId);
+}
diff --git
a/paimon-web-server/src/main/java/org/apache/paimon/web/server/service/UserSessionManager.java
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/service/UserSessionManager.java
new file mode 100644
index 0000000..8fd5dc9
--- /dev/null
+++
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/service/UserSessionManager.java
@@ -0,0 +1,50 @@
+/*
+ * 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.paimon.web.server.service;
+
+import org.apache.paimon.web.engine.flink.sql.gateway.model.SessionEntity;
+
+import org.springframework.stereotype.Service;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+
+/** Manages user sessions for Flink SQL Gateway. */
+@Service
+public class UserSessionManager {
+
+ private final ConcurrentHashMap<String, SessionEntity> sessions = new
ConcurrentHashMap<>();
+
+ public SessionEntity getSession(String id) {
+ return sessions.get(id);
+ }
+
+ public void addSession(String id, SessionEntity session) {
+ sessions.put(id, session);
+ }
+
+ public void removeSession(String id) {
+ sessions.remove(id);
+ }
+
+ public List<SessionEntity> getAllSessions() {
+ return new ArrayList<>(sessions.values());
+ }
+}
diff --git
a/paimon-web-server/src/main/java/org/apache/paimon/web/server/service/impl/SessionServiceImpl.java
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/service/impl/SessionServiceImpl.java
new file mode 100644
index 0000000..4848f5a
--- /dev/null
+++
b/paimon-web-server/src/main/java/org/apache/paimon/web/server/service/impl/SessionServiceImpl.java
@@ -0,0 +1,105 @@
+/*
+ * 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.paimon.web.server.service.impl;
+
+import org.apache.paimon.web.engine.flink.sql.gateway.client.SqlGatewayClient;
+import org.apache.paimon.web.engine.flink.sql.gateway.model.SessionEntity;
+import org.apache.paimon.web.server.data.dto.SessionDTO;
+import org.apache.paimon.web.server.service.SessionService;
+import org.apache.paimon.web.server.service.UserService;
+import org.apache.paimon.web.server.service.UserSessionManager;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.util.UUID;
+
+/** The implementation of {@link SessionService}. */
+@Service
+public class SessionServiceImpl implements SessionService {
+
+ private static final Integer ACTIVE_STATUS = 1;
+ private static final Integer INACTIVE_STATUS = 0;
+
+ @Autowired private UserSessionManager sessionManager;
+
+ @Autowired private UserService userService;
+
+ @Override
+ public void createSession(SessionDTO sessionDTO) {
+ try {
+ SqlGatewayClient client =
+ new SqlGatewayClient(sessionDTO.getHost(),
sessionDTO.getPort());
+ if (sessionDTO.getUid() != null) {
+ String username =
userService.getUserById(sessionDTO.getUid()).getUsername();
+ String sessionName = username + "_" + UUID.randomUUID();
+ if (getSession(sessionDTO.getUid(), sessionDTO.getClusterId())
== null
+ || triggerSessionHeartbeat(sessionDTO) < 1) {
+ SessionEntity sessionEntity =
client.openSession(sessionName);
+ sessionManager.addSession(
+ sessionDTO.getUid() + "_" +
sessionDTO.getClusterId(), sessionEntity);
+ }
+ }
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to create session", e);
+ }
+ }
+
+ @Override
+ public void closeSession(SessionDTO sessionDTO) {
+ try {
+ SqlGatewayClient client =
+ new SqlGatewayClient(sessionDTO.getHost(),
sessionDTO.getPort());
+ if (sessionDTO.getUid() != null) {
+ SessionEntity session =
+ sessionManager.getSession(
+ sessionDTO.getUid() + "_" +
sessionDTO.getClusterId());
+ if (session != null) {
+ client.closeSession(session.getSessionId());
+ sessionManager.removeSession(
+ sessionDTO.getUid() + "_" +
sessionDTO.getClusterId());
+ }
+ }
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to close session", e);
+ }
+ }
+
+ @Override
+ public int triggerSessionHeartbeat(SessionDTO sessionDTO) {
+ try {
+ if (sessionDTO.getUid() != null) {
+ SqlGatewayClient client =
+ new SqlGatewayClient(sessionDTO.getHost(),
sessionDTO.getPort());
+ SessionEntity session =
+ sessionManager.getSession(
+ sessionDTO.getUid() + "_" +
sessionDTO.getClusterId());
+ client.triggerSessionHeartbeat(session.getSessionId());
+ }
+ } catch (Exception e) {
+ return INACTIVE_STATUS;
+ }
+ return ACTIVE_STATUS;
+ }
+
+ @Override
+ public SessionEntity getSession(Integer uid, Integer clusterId) {
+ return sessionManager.getSession(uid + "_" + clusterId);
+ }
+}
diff --git
a/paimon-web-server/src/test/java/org/apache/paimon/web/server/controller/FlinkSQLGatewayTestBase.java
b/paimon-web-server/src/test/java/org/apache/paimon/web/server/controller/FlinkSQLGatewayTestBase.java
new file mode 100644
index 0000000..59f7312
--- /dev/null
+++
b/paimon-web-server/src/test/java/org/apache/paimon/web/server/controller/FlinkSQLGatewayTestBase.java
@@ -0,0 +1,201 @@
+/*
+ * 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.paimon.web.server.controller;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.testutils.CommonTestUtils;
+import org.apache.flink.table.gateway.api.SqlGatewayService;
+import
org.apache.flink.table.gateway.api.endpoint.SqlGatewayEndpointFactoryUtils;
+import org.apache.flink.table.gateway.rest.SqlGatewayRestEndpoint;
+import org.apache.flink.table.gateway.rest.util.SqlGatewayRestOptions;
+import org.apache.flink.table.gateway.service.SqlGatewayServiceImpl;
+import org.apache.flink.table.gateway.service.context.DefaultContext;
+import org.apache.flink.table.gateway.service.session.SessionManager;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.extension.AfterAllCallback;
+import org.junit.jupiter.api.extension.BeforeAllCallback;
+import org.junit.jupiter.api.extension.Extension;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.rules.TemporaryFolder;
+
+import javax.annotation.Nullable;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+import static
org.apache.flink.configuration.ConfigConstants.ENV_FLINK_CONF_DIR;
+import static
org.apache.flink.table.gateway.api.endpoint.SqlGatewayEndpointFactoryUtils.getEndpointConfig;
+import static
org.apache.flink.table.gateway.api.endpoint.SqlGatewayEndpointFactoryUtils.getSqlGatewayOptionPrefix;
+import static
org.apache.flink.table.gateway.rest.SqlGatewayRestEndpointFactory.IDENTIFIER;
+import static
org.apache.flink.table.gateway.rest.SqlGatewayRestEndpointFactory.rebuildRestEndpointOptions;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/** The base class for sql gateway test. */
+public class FlinkSQLGatewayTestBase {
+
+ @RegisterExtension
+ @Order(1)
+ private static final MiniClusterExtension MINI_CLUSTER = new
MiniClusterExtension();
+
+ @RegisterExtension
+ @Order(2)
+ protected static final SqlGatewayServiceExtension
SQL_GATEWAY_SERVICE_EXTENSION =
+ new
SqlGatewayServiceExtension(MINI_CLUSTER::getClientConfiguration);
+
+ @Nullable protected static String targetAddress = null;
+ @Nullable private static SqlGatewayRestEndpoint sqlGatewayRestEndpoint =
null;
+
+ protected static int port = 0;
+
+ @BeforeAll
+ static void start() throws Exception {
+ final String address =
InetAddress.getLoopbackAddress().getHostAddress();
+ Configuration config = getBaseConfig(getFlinkConfig(address, address,
"0"));
+ sqlGatewayRestEndpoint =
+ new SqlGatewayRestEndpoint(config,
SQL_GATEWAY_SERVICE_EXTENSION.getService());
+ sqlGatewayRestEndpoint.start();
+ InetSocketAddress serverAddress =
checkNotNull(sqlGatewayRestEndpoint.getServerAddress());
+ targetAddress = serverAddress.getHostName();
+ port = serverAddress.getPort();
+ }
+
+ @AfterAll
+ static void stop() throws Exception {
+ checkNotNull(sqlGatewayRestEndpoint);
+ sqlGatewayRestEndpoint.close();
+ }
+
+ private static Configuration getBaseConfig(Configuration flinkConf) {
+ SqlGatewayEndpointFactoryUtils.DefaultEndpointFactoryContext context =
+ new
SqlGatewayEndpointFactoryUtils.DefaultEndpointFactoryContext(
+ null, flinkConf, getEndpointConfig(flinkConf,
IDENTIFIER));
+
+ return rebuildRestEndpointOptions(context.getEndpointOptions());
+ }
+
+ private static Configuration getFlinkConfig(
+ String address, String bindAddress, String portRange) {
+ final Configuration config = new Configuration();
+ if (address != null) {
+ config.setString(
+
getSqlGatewayRestOptionFullName(SqlGatewayRestOptions.ADDRESS.key()), address);
+ }
+ if (bindAddress != null) {
+ config.setString(
+
getSqlGatewayRestOptionFullName(SqlGatewayRestOptions.BIND_ADDRESS.key()),
+ bindAddress);
+ }
+ if (portRange != null) {
+ config.setString(
+
getSqlGatewayRestOptionFullName(SqlGatewayRestOptions.PORT.key()), portRange);
+ }
+ return config;
+ }
+
+ private static String getSqlGatewayRestOptionFullName(String key) {
+ return getSqlGatewayOptionPrefix(IDENTIFIER) + key;
+ }
+
+ /** A simple {@link Extension} to be used by tests that require a {@link
SqlGatewayService}. */
+ static class SqlGatewayServiceExtension implements BeforeAllCallback,
AfterAllCallback {
+
+ private SqlGatewayService service;
+ private SessionManager sessionManager;
+ private TemporaryFolder temporaryFolder;
+ private final Supplier<Configuration> configSupplier;
+ private final Function<DefaultContext, SessionManager>
sessionManagerCreator;
+
+ public SqlGatewayServiceExtension(Supplier<Configuration>
configSupplier) {
+ this(configSupplier, SessionManager::create);
+ }
+
+ public SqlGatewayServiceExtension(
+ Supplier<Configuration> configSupplier,
+ Function<DefaultContext, SessionManager>
sessionManagerCreator) {
+ this.configSupplier = configSupplier;
+ this.sessionManagerCreator = sessionManagerCreator;
+ }
+
+ @Override
+ public void beforeAll(ExtensionContext context) throws Exception {
+ final Map<String, String> originalEnv = System.getenv();
+ try {
+ // prepare conf dir
+ temporaryFolder = new TemporaryFolder();
+ temporaryFolder.create();
+ File confFolder = temporaryFolder.newFolder("conf");
+ File confYaml = new File(confFolder, "flink-conf.yaml");
+ if (!confYaml.createNewFile()) {
+ throw new IOException("Can't create testing
flink-conf.yaml file.");
+ }
+
+ FileUtils.write(
+ confYaml,
+ getFlinkConfContent(configSupplier.get().toMap()),
+ StandardCharsets.UTF_8);
+
+ // adjust the test environment for the purposes of this test
+ Map<String, String> map = new HashMap<>(System.getenv());
+ map.put(ENV_FLINK_CONF_DIR, confFolder.getAbsolutePath());
+ CommonTestUtils.setEnv(map);
+
+ sessionManager =
+ sessionManagerCreator.apply(
+ DefaultContext.load(
+ new Configuration(),
Collections.emptyList(), true, false));
+ } finally {
+ CommonTestUtils.setEnv(originalEnv);
+ }
+
+ service = new SqlGatewayServiceImpl(sessionManager);
+ sessionManager.start();
+ }
+
+ @Override
+ public void afterAll(ExtensionContext context) throws Exception {
+ if (sessionManager != null) {
+ sessionManager.stop();
+ }
+ temporaryFolder.delete();
+ }
+
+ public SqlGatewayService getService() {
+ return service;
+ }
+
+ private String getFlinkConfContent(Map<String, String> flinkConf) {
+ StringBuilder sb = new StringBuilder();
+ flinkConf.forEach((k, v) -> sb.append(k).append(":
").append(v).append("\n"));
+ return sb.toString();
+ }
+ }
+}
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
new file mode 100644
index 0000000..c2bcda4
--- /dev/null
+++
b/paimon-web-server/src/test/java/org/apache/paimon/web/server/controller/SessionControllerTest.java
@@ -0,0 +1,175 @@
+/*
+ * 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.paimon.web.server.controller;
+
+import org.apache.paimon.web.engine.flink.sql.gateway.model.SessionEntity;
+import org.apache.paimon.web.server.data.dto.LoginDTO;
+import org.apache.paimon.web.server.data.model.ClusterInfo;
+import org.apache.paimon.web.server.data.result.R;
+import org.apache.paimon.web.server.service.ClusterService;
+import org.apache.paimon.web.server.service.UserSessionManager;
+import org.apache.paimon.web.server.util.ObjectMapperUtils;
+import org.apache.paimon.web.server.util.StringUtils;
+
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.fasterxml.jackson.core.type.TypeReference;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.MethodOrderer;
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestMethodOrder;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import
org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.http.MediaType;
+import org.springframework.mock.web.MockCookie;
+import org.springframework.mock.web.MockHttpServletResponse;
+import org.springframework.test.web.servlet.MockMvc;
+import org.springframework.test.web.servlet.request.MockMvcRequestBuilders;
+import org.springframework.test.web.servlet.result.MockMvcResultHandlers;
+import org.springframework.test.web.servlet.result.MockMvcResultMatchers;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** Test for {@link SessionController}. */
+@SpringBootTest
+@AutoConfigureMockMvc
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+public class SessionControllerTest extends FlinkSQLGatewayTestBase {
+
+ private static final String loginPath = "/api/login";
+ private static final String logoutPath = "/api/logout";
+ private static final String sessionPath = "/api/session";
+
+ @Value("${spring.application.name}")
+ private String tokenName;
+
+ @Autowired public MockMvc mockMvc;
+
+ public static MockCookie cookie;
+
+ @Autowired private ClusterService clusterService;
+
+ @Autowired private UserSessionManager sessionManager;
+
+ @BeforeEach
+ public void before() throws Exception {
+ LoginDTO login = new LoginDTO();
+ login.setUsername("admin");
+ login.setPassword("admin");
+ MockHttpServletResponse response =
+ mockMvc.perform(
+ MockMvcRequestBuilders.post(loginPath)
+
.content(ObjectMapperUtils.toJSON(login))
+
.contentType(MediaType.APPLICATION_JSON_VALUE)
+
.accept(MediaType.APPLICATION_JSON_VALUE))
+ .andExpect(MockMvcResultMatchers.status().isOk())
+ .andDo(MockMvcResultHandlers.print())
+ .andReturn()
+ .getResponse();
+ String result = response.getContentAsString();
+ R<?> r = ObjectMapperUtils.fromJSON(result, R.class);
+ assertEquals(200, r.getCode());
+
+ assertTrue(StringUtils.isNotBlank(r.getData().toString()));
+
+ cookie = (MockCookie) response.getCookie(tokenName);
+ }
+
+ @AfterEach
+ public void after() throws Exception {
+ String result =
+ mockMvc.perform(
+ MockMvcRequestBuilders.post(logoutPath)
+ .cookie(cookie)
+
.contentType(MediaType.APPLICATION_JSON_VALUE)
+
.accept(MediaType.APPLICATION_JSON_VALUE))
+ .andExpect(MockMvcResultMatchers.status().isOk())
+ .andDo(MockMvcResultHandlers.print())
+ .andReturn()
+ .getResponse()
+ .getContentAsString();
+ R<?> r = ObjectMapperUtils.fromJSON(result, R.class);
+ assertEquals(200, r.getCode());
+ }
+
+ @Test
+ @Order(1)
+ public void testCreateSession() throws Exception {
+ ClusterInfo cluster =
+ ClusterInfo.builder()
+ .clusterName("test_cluster")
+ .host(targetAddress)
+ .port(port)
+ .enabled(true)
+ .type("Flink")
+ .build();
+ boolean res = clusterService.save(cluster);
+ assertTrue(res);
+
+ QueryWrapper<ClusterInfo> queryWrapper = new QueryWrapper<>();
+ queryWrapper.eq("cluster_name", "test_cluster");
+ ClusterInfo one = clusterService.getOne(queryWrapper);
+ String responseString =
+ mockMvc.perform(
+ MockMvcRequestBuilders.post(sessionPath +
"/create")
+ .cookie(cookie)
+ .param("uid", "1")
+
.contentType(MediaType.APPLICATION_JSON_VALUE)
+
.accept(MediaType.APPLICATION_JSON_VALUE))
+ .andExpect(MockMvcResultMatchers.status().isOk())
+ .andDo(MockMvcResultHandlers.print())
+ .andReturn()
+ .getResponse()
+ .getContentAsString();
+ R<Void> r = ObjectMapperUtils.fromJSON(responseString, new
TypeReference<R<Void>>() {});
+ assertEquals(200, r.getCode());
+ SessionEntity session = sessionManager.getSession("1" + "_" +
one.getId());
+ assertEquals(session.getHost(), targetAddress);
+ assertEquals(session.getPort(), port);
+ }
+
+ @Test
+ @Order(2)
+ public void testDropSession() throws Exception {
+ QueryWrapper<ClusterInfo> queryWrapper = new QueryWrapper<>();
+ queryWrapper.eq("cluster_name", "test_cluster");
+ ClusterInfo one = clusterService.getOne(queryWrapper);
+ String responseString =
+ mockMvc.perform(
+ MockMvcRequestBuilders.post(sessionPath +
"/drop")
+ .cookie(cookie)
+ .param("uid", "1")
+
.contentType(MediaType.APPLICATION_JSON_VALUE)
+
.accept(MediaType.APPLICATION_JSON_VALUE))
+ .andExpect(MockMvcResultMatchers.status().isOk())
+ .andDo(MockMvcResultHandlers.print())
+ .andReturn()
+ .getResponse()
+ .getContentAsString();
+ R<Void> r = ObjectMapperUtils.fromJSON(responseString, new
TypeReference<R<Void>>() {});
+ assertEquals(200, r.getCode());
+ SessionEntity session = sessionManager.getSession("1" + "_" +
one.getId());
+ assertNull(session);
+ }
+}
