You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gu...@apache.org on 2024/04/24 16:12:05 UTC
(flink) branch FLINK-31223-rest-ssl-fix created (now 05cfbb8b308)
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a change to branch FLINK-31223-rest-ssl-fix
in repository https://gitbox.apache.org/repos/asf/flink.git
at 05cfbb8b308 [FLINK-31223][sqlgateway] Pass all flink parameters to the SqlGatewayRestEndpoint.
This branch includes the following new commits:
new d3123fd43c6 [FLINK-31223][sqlgateway] Introduce getFlinkConfigurationOptions to get all flink configurations copy in the form of map.
new 43e1ee5061a [FLINK-31223][test] Abstract SqlClientTestBase.
new 05cfbb8b308 [FLINK-31223][sqlgateway] Pass all flink parameters to the SqlGatewayRestEndpoint.
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(flink) 03/03: [FLINK-31223][sqlgateway] Pass all flink parameters to the SqlGatewayRestEndpoint.
Posted by gu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch FLINK-31223-rest-ssl-fix
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 05cfbb8b308ed0250077ed90bbd06a6661982f9a
Author: Weijie Guo <re...@163.com>
AuthorDate: Sun Feb 26 22:25:37 2023 +0800
[FLINK-31223][sqlgateway] Pass all flink parameters to the SqlGatewayRestEndpoint.
---
.../flink/table/client/SqlClientSSLTest.java | 107 +++++++++++++++++++++
.../src/test/resources/ssl/local127.keystore | Bin 0 -> 4311 bytes
.../src/test/resources/ssl/local127.truststore | Bin 0 -> 1508 bytes
.../rest/SqlGatewayRestEndpointFactory.java | 28 +++---
.../rest/util/SqlGatewayRestEndpointTestUtils.java | 3 +-
5 files changed, 123 insertions(+), 15 deletions(-)
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/SqlClientSSLTest.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/SqlClientSSLTest.java
new file mode 100644
index 00000000000..ccb4832f076
--- /dev/null
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/SqlClientSSLTest.java
@@ -0,0 +1,107 @@
+/*
+ * 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.flink.table.client;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.SecurityOptions;
+import org.apache.flink.table.gateway.rest.util.SqlGatewayRestEndpointExtension;
+import org.apache.flink.table.gateway.service.utils.SqlGatewayServiceExtension;
+
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.URL;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test that {@link SqlClient} works normally when SSL is enabled. */
+class SqlClientSSLTest extends SqlClientTestBase {
+ @RegisterExtension
+ @Order(1)
+ public static final SqlGatewayServiceExtension SQL_GATEWAY_SERVICE_EXTENSION =
+ new SqlGatewayServiceExtension(Configuration::new);
+
+ @RegisterExtension
+ @Order(2)
+ private static final SqlGatewayRestEndpointExtension SQL_GATEWAY_REST_ENDPOINT_EXTENSION =
+ new SqlGatewayRestEndpointExtension(
+ SQL_GATEWAY_SERVICE_EXTENSION::getService, SqlClientSSLTest::withSSL);
+
+ private static final String truststorePath = getTestResource("ssl/local127.truststore");
+
+ private static final String keystorePath = getTestResource("ssl/local127.keystore");
+
+ @Test
+ void testEmbeddedMode() throws Exception {
+ String[] args = new String[] {"embedded"};
+ String actual = runSqlClient(args, String.join("\n", "SET;", "QUIT;"), false);
+ assertThat(actual).contains(SecurityOptions.SSL_REST_ENABLED.key(), "true");
+ }
+
+ @Test
+ void testGatewayMode() throws Exception {
+ String[] args =
+ new String[] {
+ "gateway",
+ "-e",
+ InetSocketAddress.createUnresolved(
+ SQL_GATEWAY_REST_ENDPOINT_EXTENSION.getTargetAddress(),
+ SQL_GATEWAY_REST_ENDPOINT_EXTENSION.getTargetPort())
+ .toString()
+ };
+ String actual = runSqlClient(args, String.join("\n", "SET;", "QUIT;"), false);
+ assertThat(actual).contains(SecurityOptions.SSL_REST_ENABLED.key(), "true");
+ }
+
+ private static void withSSL(Configuration configuration) {
+ configuration.set(SecurityOptions.SSL_REST_ENABLED, true);
+ configuration.set(SecurityOptions.SSL_REST_TRUSTSTORE, truststorePath);
+ configuration.set(SecurityOptions.SSL_REST_TRUSTSTORE_PASSWORD, "password");
+ configuration.set(SecurityOptions.SSL_REST_KEYSTORE, keystorePath);
+ configuration.set(SecurityOptions.SSL_REST_KEYSTORE_PASSWORD, "password");
+ configuration.set(SecurityOptions.SSL_REST_KEY_PASSWORD, "password");
+ }
+
+ @Override
+ protected void writeConfigOptionsToConfYaml(Path confYamlPath) throws IOException {
+ Configuration configuration = new Configuration();
+ withSSL(configuration);
+ Files.write(
+ confYamlPath,
+ configuration.toMap().entrySet().stream()
+ .map(entry -> entry.getKey() + ": " + entry.getValue())
+ .collect(Collectors.toList()));
+ }
+
+ private static String getTestResource(final String fileName) {
+ final ClassLoader classLoader = ClassLoader.getSystemClassLoader();
+ final URL resource = classLoader.getResource(fileName);
+ if (resource == null) {
+ throw new IllegalArgumentException(
+ String.format("Test resource %s does not exist", fileName));
+ }
+ return resource.getFile();
+ }
+}
diff --git a/flink-table/flink-sql-client/src/test/resources/ssl/local127.keystore b/flink-table/flink-sql-client/src/test/resources/ssl/local127.keystore
new file mode 100644
index 00000000000..4992ac45137
Binary files /dev/null and b/flink-table/flink-sql-client/src/test/resources/ssl/local127.keystore differ
diff --git a/flink-table/flink-sql-client/src/test/resources/ssl/local127.truststore b/flink-table/flink-sql-client/src/test/resources/ssl/local127.truststore
new file mode 100644
index 00000000000..df4acf86b0d
Binary files /dev/null and b/flink-table/flink-sql-client/src/test/resources/ssl/local127.truststore differ
diff --git a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/SqlGatewayRestEndpointFactory.java b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/SqlGatewayRestEndpointFactory.java
index e0e0956f81f..de2622b80c0 100644
--- a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/SqlGatewayRestEndpointFactory.java
+++ b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/SqlGatewayRestEndpointFactory.java
@@ -26,7 +26,6 @@ import org.apache.flink.table.gateway.api.endpoint.SqlGatewayEndpointFactory;
import org.apache.flink.table.gateway.api.endpoint.SqlGatewayEndpointFactoryUtils;
import org.apache.flink.table.gateway.api.utils.SqlGatewayException;
-import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
@@ -48,7 +47,9 @@ public class SqlGatewayRestEndpointFactory implements SqlGatewayEndpointFactory
SqlGatewayEndpointFactoryUtils.createEndpointFactoryHelper(this, context);
// Check that ADDRESS must be set
endpointFactoryHelper.validate();
- Configuration config = rebuildRestEndpointOptions(context.getEndpointOptions());
+ Configuration config =
+ rebuildRestEndpointOptions(
+ context.getEndpointOptions(), context.getFlinkConfigurationOptions());
try {
return new SqlGatewayRestEndpoint(config, context.getSqlGatewayService());
} catch (Exception e) {
@@ -56,26 +57,25 @@ public class SqlGatewayRestEndpointFactory implements SqlGatewayEndpointFactory
}
}
- public static Configuration rebuildRestEndpointOptions(Map<String, String> configMap) {
- Map<String, String> effectiveConfigMap = new HashMap<>(configMap);
+ public static Configuration rebuildRestEndpointOptions(
+ Map<String, String> endpointConfigMap, Map<String, String> flinkConfigMap) {
+ flinkConfigMap.put(RestOptions.ADDRESS.key(), endpointConfigMap.get(ADDRESS.key()));
- effectiveConfigMap.put(RestOptions.ADDRESS.key(), configMap.get(ADDRESS.key()));
-
- if (configMap.containsKey(BIND_ADDRESS.key())) {
- effectiveConfigMap.put(
- RestOptions.BIND_ADDRESS.key(), configMap.get(BIND_ADDRESS.key()));
+ if (endpointConfigMap.containsKey(BIND_ADDRESS.key())) {
+ flinkConfigMap.put(
+ RestOptions.BIND_ADDRESS.key(), endpointConfigMap.get(BIND_ADDRESS.key()));
}
// we need to override RestOptions.PORT anyway, to use a different default value
- effectiveConfigMap.put(
+ flinkConfigMap.put(
RestOptions.PORT.key(),
- configMap.getOrDefault(PORT.key(), PORT.defaultValue().toString()));
+ endpointConfigMap.getOrDefault(PORT.key(), PORT.defaultValue().toString()));
- if (configMap.containsKey(BIND_PORT.key())) {
- effectiveConfigMap.put(RestOptions.BIND_PORT.key(), configMap.get(BIND_PORT.key()));
+ if (endpointConfigMap.containsKey(BIND_PORT.key())) {
+ flinkConfigMap.put(RestOptions.BIND_PORT.key(), endpointConfigMap.get(BIND_PORT.key()));
}
- return Configuration.fromMap(effectiveConfigMap);
+ return Configuration.fromMap(flinkConfigMap);
}
@Override
diff --git a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/rest/util/SqlGatewayRestEndpointTestUtils.java b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/rest/util/SqlGatewayRestEndpointTestUtils.java
index 65daeb4a3a8..2d246aeee57 100644
--- a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/rest/util/SqlGatewayRestEndpointTestUtils.java
+++ b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/rest/util/SqlGatewayRestEndpointTestUtils.java
@@ -40,7 +40,8 @@ public class SqlGatewayRestEndpointTestUtils {
new SqlGatewayEndpointFactoryUtils.DefaultEndpointFactoryContext(
null, flinkConf, getEndpointConfig(flinkConf, IDENTIFIER));
- return rebuildRestEndpointOptions(context.getEndpointOptions());
+ return rebuildRestEndpointOptions(
+ context.getEndpointOptions(), context.getFlinkConfigurationOptions());
}
/** Create the configuration generated from config.yaml. */
(flink) 02/03: [FLINK-31223][test] Abstract SqlClientTestBase.
Posted by gu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch FLINK-31223-rest-ssl-fix
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 43e1ee5061a2fe8660220930f31b11055801c97f
Author: Weijie Guo <re...@163.com>
AuthorDate: Sun Feb 26 23:13:45 2023 +0800
[FLINK-31223][test] Abstract SqlClientTestBase.
---
.../apache/flink/table/client/SqlClientTest.java | 86 +--------------
.../flink/table/client/SqlClientTestBase.java | 119 +++++++++++++++++++++
.../rest/util/SqlGatewayRestEndpointExtension.java | 14 ++-
3 files changed, 134 insertions(+), 85 deletions(-)
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/SqlClientTest.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/SqlClientTest.java
index be1ac347dd9..87f5e449340 100644
--- a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/SqlClientTest.java
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/SqlClientTest.java
@@ -19,49 +19,33 @@
package org.apache.flink.table.client;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.core.testutils.CommonTestUtils;
-import org.apache.flink.table.client.cli.TerminalUtils;
import org.apache.flink.table.gateway.rest.util.SqlGatewayRestEndpointExtension;
import org.apache.flink.table.gateway.service.utils.SqlGatewayServiceExtension;
import org.apache.flink.util.FileUtils;
import org.apache.flink.util.Preconditions;
-import org.jline.terminal.Size;
-import org.jline.terminal.Terminal;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
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 java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
import java.io.File;
+import java.net.InetSocketAddress;
import java.io.IOException;
import java.io.OutputStream;
import java.net.URL;
-import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
-import java.nio.file.Path;
import java.nio.file.Paths;
-import java.nio.file.StandardOpenOption;
import java.util.Arrays;
import java.util.Collections;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
-import static org.apache.flink.configuration.ConfigConstants.ENV_FLINK_CONF_DIR;
import static org.apache.flink.configuration.DeploymentOptions.TARGET;
import static org.apache.flink.core.testutils.CommonTestUtils.assertThrows;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link SqlClient}. */
-class SqlClientTest {
-
- @TempDir private Path tempFolder;
+class SqlClientTest extends SqlClientTestBase {
@RegisterExtension
@Order(1)
@@ -78,34 +62,6 @@ class SqlClientTest {
private static final SqlGatewayRestEndpointExtension SQL_GATEWAY_REST_ENDPOINT_EXTENSION =
new SqlGatewayRestEndpointExtension(SQL_GATEWAY_SERVICE_EXTENSION::getService);
- private Map<String, String> originalEnv;
-
- private String historyPath;
-
- @BeforeEach
- void before() throws IOException {
- originalEnv = System.getenv();
-
- // prepare conf dir
- File confFolder = Files.createTempDirectory(tempFolder, "conf").toFile();
- File confYaml = new File(confFolder, "config.yaml");
- if (!confYaml.createNewFile()) {
- throw new IOException("Can't create testing config.yaml file.");
- }
-
- // 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);
-
- historyPath = Files.createTempFile(tempFolder, "history", "").toFile().getPath();
- }
-
- @AfterEach
- void after() {
- CommonTestUtils.setEnv(originalEnv);
- }
-
@Test
void testEmbeddedWithOptions() throws Exception {
String[] args = new String[] {"embedded", "-hist", historyPath};
@@ -332,42 +288,4 @@ class SqlClientTest {
.toURI())));
assertThat(runSqlClient(args)).isEqualTo(actual);
}
-
- private String runSqlClient(String[] args) throws Exception {
- return runSqlClient(args, "QUIT;\n", false);
- }
-
- private String runSqlClient(String[] args, String statements, boolean printInput)
- throws Exception {
- try (OutputStream out = new ByteArrayOutputStream();
- Terminal terminal =
- TerminalUtils.createDumbTerminal(
- new ByteArrayInputStream(
- statements.getBytes(StandardCharsets.UTF_8)),
- out)) {
- if (printInput) {
- // The default terminal has an empty size. Here increase the terminal to allow
- // the line reader print the input string.
- terminal.setSize(new Size(160, 80));
- }
- SqlClient.startClient(args, () -> terminal);
- return out.toString().replace("\r\n", System.lineSeparator());
- }
- }
-
- private String createSqlFile(List<String> statements, String name) throws IOException {
- // create sql file
- File sqlFileFolder = Files.createTempDirectory(tempFolder, "sql-file").toFile();
- File sqlFile = new File(sqlFileFolder, name);
- if (!sqlFile.createNewFile()) {
- throw new IOException(String.format("Can't create testing %s.", name));
- }
- String sqlFilePath = sqlFile.getPath();
- Files.write(
- Paths.get(sqlFilePath),
- statements,
- StandardCharsets.UTF_8,
- StandardOpenOption.APPEND);
- return sqlFilePath;
- }
}
diff --git a/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/SqlClientTestBase.java b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/SqlClientTestBase.java
new file mode 100644
index 00000000000..d778524ecad
--- /dev/null
+++ b/flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/SqlClientTestBase.java
@@ -0,0 +1,119 @@
+/*
+ * 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.flink.table.client;
+
+import org.apache.flink.core.testutils.CommonTestUtils;
+import org.apache.flink.table.client.cli.TerminalUtils;
+
+import org.jline.terminal.Size;
+import org.jline.terminal.Terminal;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardOpenOption;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.flink.configuration.ConfigConstants.ENV_FLINK_CONF_DIR;
+
+/** Base class for test {@link SqlClient}. */
+class SqlClientTestBase {
+ @TempDir private Path tempFolder;
+
+ protected String historyPath;
+
+ protected Map<String, String> originalEnv;
+
+ @BeforeEach
+ void before() throws IOException {
+ originalEnv = System.getenv();
+
+ // prepare conf dir
+ File confFolder = Files.createTempDirectory(tempFolder, "conf").toFile();
+ File confYaml = new File(confFolder, "config.yaml");
+ if (!confYaml.createNewFile()) {
+ throw new IOException("Can't create testing config.yaml file.");
+ }
+ writeConfigOptionsToConfYaml(confYaml.toPath());
+ // 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);
+
+ historyPath = Files.createTempFile(tempFolder, "history", "").toFile().getPath();
+ }
+
+ @AfterEach
+ void after() {
+ CommonTestUtils.setEnv(originalEnv);
+ }
+
+ protected String createSqlFile(List<String> statements, String name) throws IOException {
+ // create sql file
+ File sqlFileFolder = Files.createTempDirectory(tempFolder, "sql-file").toFile();
+ File sqlFile = new File(sqlFileFolder, name);
+ if (!sqlFile.createNewFile()) {
+ throw new IOException(String.format("Can't create testing %s.", name));
+ }
+ String sqlFilePath = sqlFile.getPath();
+ Files.write(
+ Paths.get(sqlFilePath),
+ statements,
+ StandardCharsets.UTF_8,
+ StandardOpenOption.APPEND);
+ return sqlFilePath;
+ }
+
+ public static String runSqlClient(String[] args) throws Exception {
+ return runSqlClient(args, "QUIT;\n", false);
+ }
+
+ public static String runSqlClient(String[] args, String statements, boolean printInput)
+ throws Exception {
+ try (OutputStream out = new ByteArrayOutputStream();
+ Terminal terminal =
+ TerminalUtils.createDumbTerminal(
+ new ByteArrayInputStream(
+ statements.getBytes(StandardCharsets.UTF_8)),
+ out)) {
+ if (printInput) {
+ // The default terminal has an empty size. Here increase the terminal to allow
+ // the line reader print the input string.
+ terminal.setSize(new Size(160, 80));
+ }
+ SqlClient.startClient(args, () -> terminal);
+ return out.toString().replace("\r\n", System.lineSeparator());
+ }
+ }
+
+ protected void writeConfigOptionsToConfYaml(Path confYamlPath) throws IOException {
+ // no-op for default.
+ }
+}
diff --git a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/rest/util/SqlGatewayRestEndpointExtension.java b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/rest/util/SqlGatewayRestEndpointExtension.java
index e694a0057b6..80a78068e9c 100644
--- a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/rest/util/SqlGatewayRestEndpointExtension.java
+++ b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/rest/util/SqlGatewayRestEndpointExtension.java
@@ -30,6 +30,7 @@ import org.junit.jupiter.api.extension.ExtensionContext;
import java.net.InetAddress;
import java.net.InetSocketAddress;
+import java.util.function.Consumer;
import java.util.function.Supplier;
import static org.apache.flink.table.gateway.rest.util.SqlGatewayRestEndpointTestUtils.getBaseConfig;
@@ -41,6 +42,8 @@ public class SqlGatewayRestEndpointExtension implements BeforeAllCallback, After
private final Supplier<SqlGatewayService> serviceSupplier;
+ private final Consumer<Configuration> flinkConfConsumer;
+
private SqlGatewayRestEndpoint sqlGatewayRestEndpoint;
private SqlGatewayService sqlGatewayService;
private String targetAddress;
@@ -59,13 +62,22 @@ public class SqlGatewayRestEndpointExtension implements BeforeAllCallback, After
}
public SqlGatewayRestEndpointExtension(Supplier<SqlGatewayService> serviceSupplier) {
+ this(serviceSupplier, (conf) -> {});
+ }
+
+ public SqlGatewayRestEndpointExtension(
+ Supplier<SqlGatewayService> serviceSupplier,
+ Consumer<Configuration> flinkConfConsumer) {
this.serviceSupplier = serviceSupplier;
+ this.flinkConfConsumer = flinkConfConsumer;
}
@Override
public void beforeAll(ExtensionContext context) {
String address = InetAddress.getLoopbackAddress().getHostAddress();
- Configuration config = getBaseConfig(getFlinkConfig(address, address, "0"));
+ Configuration flinkConfig = getFlinkConfig(address, address, "0");
+ flinkConfConsumer.accept(flinkConfig);
+ Configuration config = getBaseConfig(flinkConfig);
try {
sqlGatewayService = serviceSupplier.get();
(flink) 01/03: [FLINK-31223][sqlgateway] Introduce getFlinkConfigurationOptions to get all flink configurations copy in the form of map.
Posted by gu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch FLINK-31223-rest-ssl-fix
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d3123fd43c6c97d6061cb4f9fc38be452d5e7f8f
Author: Weijie Guo <re...@163.com>
AuthorDate: Sun Feb 26 22:23:45 2023 +0800
[FLINK-31223][sqlgateway] Introduce getFlinkConfigurationOptions to get all flink configurations copy in the form of map.
---
.../table/gateway/api/endpoint/SqlGatewayEndpointFactory.java | 8 ++++++++
.../gateway/api/endpoint/SqlGatewayEndpointFactoryUtils.java | 5 +++++
2 files changed, 13 insertions(+)
diff --git a/flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/endpoint/SqlGatewayEndpointFactory.java b/flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/endpoint/SqlGatewayEndpointFactory.java
index 03d7c1e44dd..e48fe153451 100644
--- a/flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/endpoint/SqlGatewayEndpointFactory.java
+++ b/flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/endpoint/SqlGatewayEndpointFactory.java
@@ -50,6 +50,14 @@ public interface SqlGatewayEndpointFactory extends Factory {
/** Gives read-only access to the configuration of the current session. */
ReadableConfig getFlinkConfiguration();
+ /**
+ * Get a map contains all flink configurations.
+ *
+ * @return The copy of flink configurations in the form of map, modify this map will not
+ * influence the original configuration object.
+ */
+ Map<String, String> getFlinkConfigurationOptions();
+
/**
* Returns the options with which the endpoint is created. All options that are prefixed
* with the endpoint identifier are included in the map.
diff --git a/flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/endpoint/SqlGatewayEndpointFactoryUtils.java b/flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/endpoint/SqlGatewayEndpointFactoryUtils.java
index 7ba59f609e0..3063cf89b3c 100644
--- a/flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/endpoint/SqlGatewayEndpointFactoryUtils.java
+++ b/flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/endpoint/SqlGatewayEndpointFactoryUtils.java
@@ -141,6 +141,11 @@ public class SqlGatewayEndpointFactoryUtils {
return flinkConfiguration;
}
+ @Override
+ public Map<String, String> getFlinkConfigurationOptions() {
+ return flinkConfiguration.toMap();
+ }
+
@Override
public Map<String, String> getEndpointOptions() {
return endpointConfig;