You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sh...@apache.org on 2023/02/07 02:32:01 UTC
[flink] branch master updated: [FLINK-30811][sql-gateway] Fix "STOP JOB WITH SAVEPOINT" cannot read savepoint dir from session configuration
This is an automated email from the ASF dual-hosted git repository.
shengkai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new bbd5a7876eb [FLINK-30811][sql-gateway] Fix "STOP JOB WITH SAVEPOINT" cannot read savepoint dir from session configuration
bbd5a7876eb is described below
commit bbd5a7876eb1542ff89f05f5f5d82bb8bd41b7bd
Author: Jane Chan <qi...@gmail.com>
AuthorDate: Tue Jan 31 15:54:55 2023 +0800
[FLINK-30811][sql-gateway] Fix "STOP JOB WITH SAVEPOINT" cannot read savepoint dir from session configuration
This closes #21802
---
.../table/client/gateway/ExecutorImplITCase.java | 57 ++++++--
.../gateway/service/context/SessionContext.java | 122 ++--------------
.../service/operation/OperationExecutor.java | 158 ++++++++++++++++++---
.../table/gateway/service/session/Session.java | 2 +-
.../service/context/SessionContextTest.java | 55 +++----
5 files changed, 218 insertions(+), 176 deletions(-)
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/ExecutorImplITCase.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/ExecutorImplITCase.java
index 47634410839..770c4ea6f61 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/ExecutorImplITCase.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/ExecutorImplITCase.java
@@ -19,6 +19,7 @@
package org.apache.flink.table.client.gateway;
import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.configuration.CheckpointingOptions;
@@ -39,7 +40,6 @@ import org.apache.flink.table.client.config.ResultMode;
import org.apache.flink.table.client.gateway.result.ChangelogCollectResult;
import org.apache.flink.table.client.gateway.result.MaterializedResult;
import org.apache.flink.table.data.RowData;
-import org.apache.flink.table.data.StringData;
import org.apache.flink.table.functions.AggregateFunction;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.gateway.api.operation.OperationHandle;
@@ -74,6 +74,8 @@ import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import javax.annotation.Nullable;
@@ -184,7 +186,6 @@ class ExecutorImplITCase {
config.set(StateBackendOptions.STATE_BACKEND, "hashmap");
config.set(CheckpointingOptions.CHECKPOINT_STORAGE, "filesystem");
config.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, tempFolder.toURI().toString());
- config.set(CheckpointingOptions.SAVEPOINT_DIRECTORY, tempFolder.toURI().toString());
return config;
}
@@ -434,8 +435,9 @@ class ExecutorImplITCase {
}
}
- @Test
- void testStopJob() throws Exception {
+ @ValueSource(booleans = {true, false})
+ @ParameterizedTest
+ void testStopJob(boolean withSavepoint) throws Exception {
final Map<String, String> configMap = new HashMap<>();
configMap.put(EXECUTION_RESULT_MODE.key(), ResultMode.TABLE.name());
configMap.put(RUNTIME_MODE.key(), RuntimeExecutionMode.STREAMING.name());
@@ -449,6 +451,7 @@ class ExecutorImplITCase {
createRestServiceExecutor(
Collections.singletonList(udfDependency),
Configuration.fromMap(configMap))) {
+
executor.configureSession(srcDdl);
executor.configureSession(snkDdl);
StatementResult result = executor.executeStatement(insert);
@@ -456,17 +459,41 @@ class ExecutorImplITCase {
// wait till the job turns into running status or the test times out
TestUtils.waitUntilAllTasksAreRunning(clusterClient, jobID);
- StringData savepointPath =
- CollectionUtil.iteratorToList(
- executor.executeStatement(
- String.format("STOP JOB '%s' WITH SAVEPOINT", jobID)))
- .get(0)
- .getString(0);
- assertThat(savepointPath)
- .isNotNull()
- .matches(
- stringData ->
- Files.exists(Paths.get(URI.create(stringData.toString()))));
+
+ JobStatus expectedJobStatus;
+ if (withSavepoint) {
+ executor.configureSession(
+ String.format(
+ "SET '%s' = '%s'",
+ CheckpointingOptions.SAVEPOINT_DIRECTORY.key(),
+ tempFolder.toURI()));
+ StatementResult stopResult =
+ executor.executeStatement(
+ String.format("STOP JOB '%s' WITH SAVEPOINT", jobID));
+ String savepointPath =
+ CollectionUtil.iteratorToList(stopResult).get(0).getString(0).toString();
+ assertThat(savepointPath)
+ .matches(path -> Files.exists(Paths.get(URI.create(path))));
+ expectedJobStatus = JobStatus.FINISHED;
+ } else {
+ executor.executeStatement(String.format("STOP JOB '%s'", jobID));
+ TestUtils.waitUntilJobCanceled(jobID, clusterClient);
+ expectedJobStatus = JobStatus.CANCELED;
+ }
+ assertThat(jobID).isNotNull();
+ assertThat(
+ CollectionUtil.iteratorToList(executor.executeStatement("SHOW JOBS"))
+ .stream()
+ .filter(
+ row ->
+ row.getString(0)
+ .toString()
+ .equals(jobID.toHexString()))
+ .findAny())
+ .isPresent()
+ .map(row -> row.getString(2).toString())
+ .get()
+ .isEqualTo(expectedJobStatus.name());
}
}
diff --git a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/context/SessionContext.java b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/context/SessionContext.java
index 7ce60d8fda4..d9dddf20a36 100644
--- a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/context/SessionContext.java
+++ b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/context/SessionContext.java
@@ -22,24 +22,13 @@ import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ReadableConfig;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.configuration.UnmodifiableConfiguration;
import org.apache.flink.table.api.EnvironmentSettings;
-import org.apache.flink.table.api.SqlDialect;
-import org.apache.flink.table.api.TableConfig;
-import org.apache.flink.table.api.TableException;
-import org.apache.flink.table.api.ValidationException;
-import org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl;
import org.apache.flink.table.api.config.TableConfigOptions;
-import org.apache.flink.table.api.internal.TableEnvironmentInternal;
import org.apache.flink.table.catalog.Catalog;
import org.apache.flink.table.catalog.CatalogManager;
import org.apache.flink.table.catalog.FunctionCatalog;
import org.apache.flink.table.catalog.GenericInMemoryCatalog;
-import org.apache.flink.table.delegation.Executor;
-import org.apache.flink.table.delegation.ExecutorFactory;
-import org.apache.flink.table.delegation.Planner;
-import org.apache.flink.table.factories.FactoryUtil;
-import org.apache.flink.table.factories.PlannerFactoryUtil;
import org.apache.flink.table.gateway.api.endpoint.EndpointVersion;
import org.apache.flink.table.gateway.api.session.SessionEnvironment;
import org.apache.flink.table.gateway.api.session.SessionHandle;
@@ -51,7 +40,6 @@ import org.apache.flink.table.module.Module;
import org.apache.flink.table.module.ModuleManager;
import org.apache.flink.table.operations.ModifyOperation;
import org.apache.flink.table.resource.ResourceManager;
-import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkUserCodeClassLoaders;
import org.apache.flink.util.MutableURLClassLoader;
@@ -59,7 +47,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
-import java.lang.reflect.Method;
import java.net.URL;
import java.net.URLClassLoader;
import java.nio.file.Path;
@@ -69,7 +56,6 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.Deque;
import java.util.List;
-import java.util.Map;
import java.util.concurrent.ExecutorService;
/**
@@ -122,8 +108,8 @@ public class SessionContext {
return this.sessionId;
}
- public Map<String, String> getConfigMap() {
- return sessionConf.toMap();
+ public Configuration getSessionConf() {
+ return new UnmodifiableConfiguration(sessionConf);
}
public OperationManager getOperationManager() {
@@ -138,6 +124,14 @@ public class SessionContext {
return sessionState;
}
+ public DefaultContext getDefaultContext() {
+ return defaultContext;
+ }
+
+ public URLClassLoader getUserClassloader() {
+ return userClassloader;
+ }
+
public void set(String key, String value) {
try {
// Test whether the key value will influence the creation of the Executor.
@@ -266,100 +260,6 @@ public class SessionContext {
// Helpers
// ------------------------------------------------------------------------------------------------------------------
- public TableEnvironmentInternal createTableEnvironment() {
- // checks the value of RUNTIME_MODE
- final EnvironmentSettings settings =
- EnvironmentSettings.newInstance().withConfiguration(sessionConf).build();
-
- StreamExecutionEnvironment streamExecEnv = createStreamExecutionEnvironment();
-
- TableConfig tableConfig = TableConfig.getDefault();
- tableConfig.setRootConfiguration(defaultContext.getFlinkConfig());
- tableConfig.addConfiguration(sessionConf);
-
- final Executor executor = lookupExecutor(streamExecEnv, userClassloader);
- return createStreamTableEnvironment(
- streamExecEnv,
- settings,
- tableConfig,
- executor,
- sessionState.catalogManager,
- sessionState.moduleManager,
- sessionState.resourceManager,
- sessionState.functionCatalog);
- }
-
- private TableEnvironmentInternal createStreamTableEnvironment(
- StreamExecutionEnvironment env,
- EnvironmentSettings settings,
- TableConfig tableConfig,
- Executor executor,
- CatalogManager catalogManager,
- ModuleManager moduleManager,
- ResourceManager resourceManager,
- FunctionCatalog functionCatalog) {
-
- final Planner planner =
- PlannerFactoryUtil.createPlanner(
- executor,
- tableConfig,
- resourceManager.getUserClassLoader(),
- moduleManager,
- catalogManager,
- functionCatalog);
-
- try {
- return new StreamTableEnvironmentImpl(
- catalogManager,
- moduleManager,
- resourceManager,
- functionCatalog,
- tableConfig,
- env,
- planner,
- executor,
- settings.isStreamingMode());
- } catch (ValidationException e) {
- if (tableConfig.getSqlDialect() == SqlDialect.HIVE) {
- String additionErrorMsg =
- "Note: if you want to use Hive dialect, "
- + "please first move the jar `flink-table-planner_2.12` located in `FLINK_HOME/opt` "
- + "to `FLINK_HOME/lib` and then move out the jar `flink-table-planner-loader` from `FLINK_HOME/lib`.";
- ExceptionUtils.updateDetailMessage(e, t -> t.getMessage() + additionErrorMsg);
- }
- throw e;
- }
- }
-
- private static Executor lookupExecutor(
- StreamExecutionEnvironment executionEnvironment, ClassLoader userClassLoader) {
- try {
- final ExecutorFactory executorFactory =
- FactoryUtil.discoverFactory(
- userClassLoader,
- ExecutorFactory.class,
- ExecutorFactory.DEFAULT_IDENTIFIER);
- final Method createMethod =
- executorFactory
- .getClass()
- .getMethod("create", StreamExecutionEnvironment.class);
-
- return (Executor) createMethod.invoke(executorFactory, executionEnvironment);
- } catch (Exception e) {
- throw new TableException(
- "Could not instantiate the executor. Make sure a planner module is on the classpath",
- e);
- }
- }
-
- private StreamExecutionEnvironment createStreamExecutionEnvironment() {
- // We need not different StreamExecutionEnvironments to build and submit flink job,
- // instead we just use StreamExecutionEnvironment#executeAsync(StreamGraph) method
- // to execute existing StreamGraph.
- // This requires StreamExecutionEnvironment to have a full flink configuration.
- return new StreamExecutionEnvironment(new Configuration(sessionConf), userClassloader);
- }
-
protected static Configuration initializeConfiguration(
DefaultContext defaultContext,
SessionEnvironment environment,
diff --git a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationExecutor.java b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationExecutor.java
index 9c11f3a5d55..f705536442e 100644
--- a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationExecutor.java
+++ b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationExecutor.java
@@ -30,13 +30,21 @@ import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.execution.SavepointFormatType;
import org.apache.flink.runtime.client.JobStatusMessage;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.CatalogNotExistException;
import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.SqlDialect;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl;
import org.apache.flink.table.api.internal.TableEnvironmentInternal;
import org.apache.flink.table.api.internal.TableResultInternal;
import org.apache.flink.table.catalog.CatalogBaseTable.TableKind;
import org.apache.flink.table.catalog.CatalogManager;
import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.FunctionCatalog;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ResolvedCatalogBaseTable;
import org.apache.flink.table.catalog.ResolvedSchema;
@@ -44,6 +52,11 @@ import org.apache.flink.table.catalog.UnresolvedIdentifier;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.delegation.Executor;
+import org.apache.flink.table.delegation.ExecutorFactory;
+import org.apache.flink.table.delegation.Planner;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.factories.PlannerFactoryUtil;
import org.apache.flink.table.functions.FunctionDefinition;
import org.apache.flink.table.functions.FunctionIdentifier;
import org.apache.flink.table.gateway.api.operation.OperationHandle;
@@ -52,6 +65,7 @@ import org.apache.flink.table.gateway.api.results.TableInfo;
import org.apache.flink.table.gateway.service.context.SessionContext;
import org.apache.flink.table.gateway.service.result.ResultFetcher;
import org.apache.flink.table.gateway.service.utils.SqlExecutionException;
+import org.apache.flink.table.module.ModuleManager;
import org.apache.flink.table.operations.BeginStatementSetOperation;
import org.apache.flink.table.operations.EndStatementSetOperation;
import org.apache.flink.table.operations.LoadModuleOperation;
@@ -70,8 +84,10 @@ import org.apache.flink.table.operations.command.StopJobOperation;
import org.apache.flink.table.operations.ddl.AlterOperation;
import org.apache.flink.table.operations.ddl.CreateOperation;
import org.apache.flink.table.operations.ddl.DropOperation;
+import org.apache.flink.table.resource.ResourceManager;
import org.apache.flink.table.utils.DateTimeUtils;
import org.apache.flink.util.CollectionUtil;
+import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.TemporaryClassLoaderContext;
@@ -79,6 +95,7 @@ import org.apache.flink.util.TemporaryClassLoaderContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.lang.reflect.Method;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
@@ -109,7 +126,8 @@ public class OperationExecutor {
private static final Logger LOG = LoggerFactory.getLogger(OperationExecutor.class);
protected final SessionContext sessionContext;
- protected final Configuration executionConfig;
+
+ private final Configuration executionConfig;
private final ClusterClientServiceLoader clusterClientServiceLoader;
@@ -291,9 +309,98 @@ public class OperationExecutor {
@VisibleForTesting
public TableEnvironmentInternal getTableEnvironment() {
- TableEnvironmentInternal tableEnv = sessionContext.createTableEnvironment();
- tableEnv.getConfig().getConfiguration().addAll(executionConfig);
- return tableEnv;
+ // checks the value of RUNTIME_MODE
+ Configuration operationConfig = sessionContext.getSessionConf().clone();
+ operationConfig.addAll(executionConfig);
+ final EnvironmentSettings settings =
+ EnvironmentSettings.newInstance().withConfiguration(operationConfig).build();
+
+ // We need not different StreamExecutionEnvironments to build and submit flink job,
+ // instead we just use StreamExecutionEnvironment#executeAsync(StreamGraph) method
+ // to execute existing StreamGraph.
+ // This requires StreamExecutionEnvironment to have a full flink configuration.
+ StreamExecutionEnvironment streamExecEnv =
+ new StreamExecutionEnvironment(
+ operationConfig, sessionContext.getUserClassloader());
+
+ TableConfig tableConfig = TableConfig.getDefault();
+ tableConfig.setRootConfiguration(sessionContext.getDefaultContext().getFlinkConfig());
+ tableConfig.addConfiguration(operationConfig);
+
+ final Executor executor =
+ lookupExecutor(streamExecEnv, sessionContext.getUserClassloader());
+ return createStreamTableEnvironment(
+ streamExecEnv,
+ settings,
+ tableConfig,
+ executor,
+ sessionContext.getSessionState().catalogManager,
+ sessionContext.getSessionState().moduleManager,
+ sessionContext.getSessionState().resourceManager,
+ sessionContext.getSessionState().functionCatalog);
+ }
+
+ private static Executor lookupExecutor(
+ StreamExecutionEnvironment executionEnvironment, ClassLoader userClassLoader) {
+ try {
+ final ExecutorFactory executorFactory =
+ FactoryUtil.discoverFactory(
+ userClassLoader,
+ ExecutorFactory.class,
+ ExecutorFactory.DEFAULT_IDENTIFIER);
+ final Method createMethod =
+ executorFactory
+ .getClass()
+ .getMethod("create", StreamExecutionEnvironment.class);
+
+ return (Executor) createMethod.invoke(executorFactory, executionEnvironment);
+ } catch (Exception e) {
+ throw new TableException(
+ "Could not instantiate the executor. Make sure a planner module is on the classpath",
+ e);
+ }
+ }
+
+ private TableEnvironmentInternal createStreamTableEnvironment(
+ StreamExecutionEnvironment env,
+ EnvironmentSettings settings,
+ TableConfig tableConfig,
+ Executor executor,
+ CatalogManager catalogManager,
+ ModuleManager moduleManager,
+ ResourceManager resourceManager,
+ FunctionCatalog functionCatalog) {
+
+ final Planner planner =
+ PlannerFactoryUtil.createPlanner(
+ executor,
+ tableConfig,
+ resourceManager.getUserClassLoader(),
+ moduleManager,
+ catalogManager,
+ functionCatalog);
+
+ try {
+ return new StreamTableEnvironmentImpl(
+ catalogManager,
+ moduleManager,
+ resourceManager,
+ functionCatalog,
+ tableConfig,
+ env,
+ planner,
+ executor,
+ settings.isStreamingMode());
+ } catch (ValidationException e) {
+ if (tableConfig.getSqlDialect() == SqlDialect.HIVE) {
+ String additionErrorMsg =
+ "Note: if you want to use Hive dialect, "
+ + "please first move the jar `flink-table-planner_2.12` located in `FLINK_HOME/opt` "
+ + "to `FLINK_HOME/lib` and then move out the jar `flink-table-planner-loader` from `FLINK_HOME/lib`.";
+ ExceptionUtils.updateDetailMessage(e, t -> t.getMessage() + additionErrorMsg);
+ }
+ throw e;
+ }
}
private ResultFetcher executeOperationInStatementSetState(
@@ -330,9 +437,9 @@ public class OperationExecutor {
TableResultInternal result = tableEnv.executeInternal(op);
return ResultFetcher.fromTableResult(handle, result, true);
} else if (op instanceof StopJobOperation) {
- return callStopJobOperation(handle, (StopJobOperation) op);
+ return callStopJobOperation(tableEnv, handle, (StopJobOperation) op);
} else if (op instanceof ShowJobsOperation) {
- return callShowJobsOperation(handle, (ShowJobsOperation) op);
+ return callShowJobsOperation(tableEnv, handle, (ShowJobsOperation) op);
} else if (op instanceof RemoveJarOperation) {
return callRemoveJar(handle, ((RemoveJarOperation) op).getPath());
} else {
@@ -481,16 +588,18 @@ public class OperationExecutor {
}
public ResultFetcher callStopJobOperation(
- OperationHandle handle, StopJobOperation stopJobOperation)
+ TableEnvironmentInternal tableEnv,
+ OperationHandle handle,
+ StopJobOperation stopJobOperation)
throws SqlExecutionException {
String jobId = stopJobOperation.getJobId();
boolean isWithSavepoint = stopJobOperation.isWithSavepoint();
boolean isWithDrain = stopJobOperation.isWithDrain();
- Duration clientTimeout =
- Configuration.fromMap(sessionContext.getConfigMap())
- .get(ClientOptions.CLIENT_TIMEOUT);
+ Configuration configuration = tableEnv.getConfig().getConfiguration();
+ Duration clientTimeout = configuration.get(ClientOptions.CLIENT_TIMEOUT);
Optional<String> savepoint =
runClusterAction(
+ configuration,
handle,
clusterClient -> {
try {
@@ -501,7 +610,7 @@ public class OperationExecutor {
.stopWithSavepoint(
JobID.fromHexString(jobId),
isWithDrain,
- executionConfig.get(
+ configuration.get(
CheckpointingOptions
.SAVEPOINT_DIRECTORY),
SavepointFormatType.DEFAULT)
@@ -516,11 +625,11 @@ public class OperationExecutor {
}
} catch (Exception e) {
throw new SqlExecutionException(
- "Could not stop job "
- + jobId
- + " for operation "
- + handle.getIdentifier()
- + ".",
+ String.format(
+ "Could not stop job %s %s for operation %s.",
+ jobId,
+ isWithSavepoint ? "with savepoint" : "",
+ handle.getIdentifier()),
e);
}
});
@@ -536,13 +645,15 @@ public class OperationExecutor {
}
public ResultFetcher callShowJobsOperation(
- OperationHandle operationHandle, ShowJobsOperation showJobsOperation)
+ TableEnvironmentInternal tableEnv,
+ OperationHandle operationHandle,
+ ShowJobsOperation showJobsOperation)
throws SqlExecutionException {
- Duration clientTimeout =
- Configuration.fromMap(sessionContext.getConfigMap())
- .get(ClientOptions.CLIENT_TIMEOUT);
+ Configuration configuration = tableEnv.getConfig().getConfiguration();
+ Duration clientTimeout = configuration.get(ClientOptions.CLIENT_TIMEOUT);
Collection<JobStatusMessage> jobs =
runClusterAction(
+ configuration,
operationHandle,
clusterClient -> {
try {
@@ -579,6 +690,8 @@ public class OperationExecutor {
* Retrieves the {@link ClusterClient} from the session and runs the given {@link ClusterAction}
* against it.
*
+ * @param configuration the combined configuration of {@code sessionConf} and {@code
+ * executionConfig}.
* @param handle the specified operation handle
* @param clusterAction the cluster action to run against the retrieved {@link ClusterClient}.
* @param <ClusterID> type of the cluster id
@@ -586,9 +699,10 @@ public class OperationExecutor {
* @throws SqlExecutionException if something goes wrong
*/
private <ClusterID, Result> Result runClusterAction(
- OperationHandle handle, ClusterAction<ClusterID, Result> clusterAction)
+ Configuration configuration,
+ OperationHandle handle,
+ ClusterAction<ClusterID, Result> clusterAction)
throws SqlExecutionException {
- final Configuration configuration = Configuration.fromMap(sessionContext.getConfigMap());
final ClusterClientFactory<ClusterID> clusterClientFactory =
clusterClientServiceLoader.getClusterClientFactory(configuration);
diff --git a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/session/Session.java b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/session/Session.java
index 87d77a87c36..51aba2c4bfc 100644
--- a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/session/Session.java
+++ b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/session/Session.java
@@ -55,7 +55,7 @@ public class Session implements Closeable {
}
public Map<String, String> getSessionConfig() {
- return sessionContext.getConfigMap();
+ return sessionContext.getSessionConf().toMap();
}
public EndpointVersion getEndpointVersion() {
diff --git a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/context/SessionContextTest.java b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/context/SessionContextTest.java
index 1651db70802..bc6976b887c 100644
--- a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/context/SessionContextTest.java
+++ b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/context/SessionContextTest.java
@@ -18,8 +18,9 @@
package org.apache.flink.table.gateway.service.context;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.gateway.api.session.SessionEnvironment;
import org.apache.flink.table.gateway.api.session.SessionHandle;
import org.apache.flink.table.gateway.api.utils.MockedEndpointVersion;
@@ -71,19 +72,19 @@ class SessionContextTest {
sessionContext.set(NAME.key(), "test");
// runtime config from flink-conf
sessionContext.set(OBJECT_REUSE.key(), "false");
- assertThat(getConfiguration().get(TABLE_SQL_DIALECT)).isEqualTo("hive");
- assertThat(getConfiguration().get(MAX_PARALLELISM)).isEqualTo(128);
- assertThat(getConfiguration().get(NAME)).isEqualTo("test");
- assertThat(getConfiguration().get(OBJECT_REUSE)).isFalse();
+ assertThat(sessionContext.getSessionConf().get(TABLE_SQL_DIALECT)).isEqualTo("hive");
+ assertThat(sessionContext.getSessionConf().get(MAX_PARALLELISM)).isEqualTo(128);
+ assertThat(sessionContext.getSessionConf().get(NAME)).isEqualTo("test");
+ assertThat(sessionContext.getSessionConf().get(OBJECT_REUSE)).isFalse();
sessionContext.reset();
- assertThat(getConfiguration().get(TABLE_SQL_DIALECT)).isEqualTo("default");
- assertThat(getConfiguration().get(NAME)).isNull();
+ assertThat(sessionContext.getSessionConf().get(TABLE_SQL_DIALECT)).isEqualTo("default");
+ assertThat(sessionContext.getSessionConf().get(NAME)).isNull();
// The value of MAX_PARALLELISM in DEFAULTS_ENVIRONMENT_FILE is 16
- assertThat(getConfiguration().get(MAX_PARALLELISM)).isEqualTo(16);
- assertThat(getConfiguration().getOptional(NAME)).isEmpty();
+ assertThat(sessionContext.getSessionConf().get(MAX_PARALLELISM)).isEqualTo(16);
+ assertThat(sessionContext.getSessionConf().getOptional(NAME)).isEmpty();
// The value of OBJECT_REUSE in origin configuration is true
- assertThat(getConfiguration().get(OBJECT_REUSE)).isTrue();
+ assertThat(sessionContext.getSessionConf().get(OBJECT_REUSE)).isTrue();
}
@Test
@@ -97,22 +98,22 @@ class SessionContextTest {
// runtime config from flink-conf
sessionContext.set(OBJECT_REUSE.key(), "false");
- assertThat(getConfiguration().get(TABLE_SQL_DIALECT)).isEqualTo("hive");
- assertThat(getConfiguration().get(MAX_PARALLELISM)).isEqualTo(128);
- assertThat(getConfiguration().get(NAME)).isEqualTo("test");
- assertThat(getConfiguration().get(OBJECT_REUSE)).isFalse();
+ assertThat(sessionContext.getSessionConf().get(TABLE_SQL_DIALECT)).isEqualTo("hive");
+ assertThat(sessionContext.getSessionConf().get(MAX_PARALLELISM)).isEqualTo(128);
+ assertThat(sessionContext.getSessionConf().get(NAME)).isEqualTo("test");
+ assertThat(sessionContext.getSessionConf().get(OBJECT_REUSE)).isFalse();
sessionContext.reset(TABLE_SQL_DIALECT.key());
- assertThat(getConfiguration().get(TABLE_SQL_DIALECT)).isEqualTo("default");
+ assertThat(sessionContext.getSessionConf().get(TABLE_SQL_DIALECT)).isEqualTo("default");
sessionContext.reset(MAX_PARALLELISM.key());
- assertThat(getConfiguration().get(MAX_PARALLELISM)).isEqualTo(16);
+ assertThat(sessionContext.getSessionConf().get(MAX_PARALLELISM)).isEqualTo(16);
sessionContext.reset(NAME.key());
- assertThat(getConfiguration().get(NAME)).isNull();
+ assertThat(sessionContext.getSessionConf().get(NAME)).isNull();
sessionContext.reset(OBJECT_REUSE.key());
- assertThat(getConfiguration().get(OBJECT_REUSE)).isTrue();
+ assertThat(sessionContext.getSessionConf().get(OBJECT_REUSE)).isTrue();
}
@Test
@@ -121,15 +122,19 @@ class SessionContextTest {
sessionContext.set("aa", "11");
sessionContext.set("bb", "22");
- assertThat(sessionContext.getConfigMap().get("aa")).isEqualTo("11");
- assertThat(sessionContext.getConfigMap().get("bb")).isEqualTo("22");
+ ConfigOption<String> aa = ConfigOptions.key("aa").stringType().defaultValue("11");
+ ConfigOption<String> bb = ConfigOptions.key("bb").stringType().defaultValue("22");
+
+ assertThat(sessionContext.getSessionConf())
+ .matches((conf) -> conf.contains(aa) && conf.contains(bb));
sessionContext.reset("aa");
- assertThat(sessionContext.getConfigMap().get("aa")).isNull();
- assertThat(sessionContext.getConfigMap().get("bb")).isEqualTo("22");
+ assertThat(sessionContext.getSessionConf())
+ .matches((conf) -> !conf.containsKey("aa") && conf.contains(bb));
sessionContext.reset("bb");
- assertThat(sessionContext.getConfigMap().get("bb")).isNull();
+ assertThat(sessionContext.getSessionConf())
+ .matches((conf) -> !conf.containsKey("aa") && !conf.containsKey("bb"));
}
// --------------------------------------------------------------------------------------------
@@ -147,8 +152,4 @@ class SessionContextTest {
return SessionContext.create(
defaultContext, SessionHandle.create(), environment, EXECUTOR_SERVICE);
}
-
- private ReadableConfig getConfiguration() {
- return Configuration.fromMap(sessionContext.getConfigMap());
- }
}