Skip to content

Commit 0e59da2

Browse files
reswqareswqa
committed
[FLINK-31223][sqlgateway] Pass all flink parameters to the SqlGatewayRestEndpoint.
1 parent 3553bce commit 0e59da2

File tree

6 files changed

+124
-16
lines changed

6 files changed

+124
-16
lines changed

‎flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/SqlClientSSLTest.java

Lines changed: 107 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,107 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
package org.apache.flink.table.client;
20+
21+
import org.apache.flink.configuration.Configuration;
22+
import org.apache.flink.configuration.SecurityOptions;
23+
import org.apache.flink.table.gateway.rest.util.SqlGatewayRestEndpointExtension;
24+
import org.apache.flink.table.gateway.service.utils.SqlGatewayServiceExtension;
25+
26+
import org.junit.jupiter.api.Order;
27+
import org.junit.jupiter.api.Test;
28+
import org.junit.jupiter.api.extension.RegisterExtension;
29+
30+
import java.io.IOException;
31+
import java.net.InetSocketAddress;
32+
import java.net.URL;
33+
import java.nio.file.Files;
34+
import java.nio.file.Path;
35+
import java.util.stream.Collectors;
36+
37+
import static org.assertj.core.api.Assertions.assertThat;
38+
39+
/** Test that {@link SqlClient} works normally when SSL is enabled. */
40+
class SqlClientSSLTest extends SqlClientTestBase {
41+
@RegisterExtension
42+
@Order(1)
43+
public static final SqlGatewayServiceExtension SQL_GATEWAY_SERVICE_EXTENSION =
44+
new SqlGatewayServiceExtension(Configuration::new);
45+
46+
@RegisterExtension
47+
@Order(2)
48+
private static final SqlGatewayRestEndpointExtension SQL_GATEWAY_REST_ENDPOINT_EXTENSION =
49+
new SqlGatewayRestEndpointExtension(
50+
SQL_GATEWAY_SERVICE_EXTENSION::getService, SqlClientSSLTest::withSSL);
51+
52+
private static final String truststorePath = getTestResource("ssl/local127.truststore");
53+
54+
private static final String keystorePath = getTestResource("ssl/local127.keystore");
55+
56+
@Test
57+
void testEmbeddedMode() throws Exception {
58+
String[] args = new String[] {"embedded"};
59+
String actual = runSqlClient(args, String.join("\n", "SET;", "QUIT;"), false);
60+
assertThat(actual).contains(SecurityOptions.SSL_REST_ENABLED.key(), "true");
61+
}
62+
63+
@Test
64+
void testGatewayMode() throws Exception {
65+
String[] args =
66+
new String[] {
67+
"gateway",
68+
"-e",
69+
InetSocketAddress.createUnresolved(
70+
SQL_GATEWAY_REST_ENDPOINT_EXTENSION.getTargetAddress(),
71+
SQL_GATEWAY_REST_ENDPOINT_EXTENSION.getTargetPort())
72+
.toString()
73+
};
74+
String actual = runSqlClient(args, String.join("\n", "SET;", "QUIT;"), false);
75+
assertThat(actual).contains(SecurityOptions.SSL_REST_ENABLED.key(), "true");
76+
}
77+
78+
private static void withSSL(Configuration configuration) {
79+
configuration.set(SecurityOptions.SSL_REST_ENABLED, true);
80+
configuration.set(SecurityOptions.SSL_REST_TRUSTSTORE, truststorePath);
81+
configuration.set(SecurityOptions.SSL_REST_TRUSTSTORE_PASSWORD, "password");
82+
configuration.set(SecurityOptions.SSL_REST_KEYSTORE, keystorePath);
83+
configuration.set(SecurityOptions.SSL_REST_KEYSTORE_PASSWORD, "password");
84+
configuration.set(SecurityOptions.SSL_REST_KEY_PASSWORD, "password");
85+
}
86+
87+
@Override
88+
protected void writeConfigOptionsToConfYaml(Path confYamlPath) throws IOException {
89+
Configuration configuration = new Configuration();
90+
withSSL(configuration);
91+
Files.write(
92+
confYamlPath,
93+
configuration.toMap().entrySet().stream()
94+
.map(entry -> entry.getKey() + ": " + entry.getValue())
95+
.collect(Collectors.toList()));
96+
}
97+
98+
private static String getTestResource(final String fileName) {
99+
final ClassLoader classLoader = ClassLoader.getSystemClassLoader();
100+
final URL resource = classLoader.getResource(fileName);
101+
if (resource == null) {
102+
throw new IllegalArgumentException(
103+
String.format("Test resource %s does not exist", fileName));
104+
}
105+
return resource.getFile();
106+
}
107+
}

‎flink-table/flink-sql-client/src/test/resources/ssl/local127.keystore

4.21 KB
Binary file not shown.

‎flink-table/flink-sql-client/src/test/resources/ssl/local127.truststore

1.47 KB
Binary file not shown.

‎flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/endpoint/SqlGatewayEndpointFactory.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,7 @@ interface Context {
5353
/**
5454
* Get a map contains all flink configurations.
5555
*
56-
* @return The copy of flink configurations in the form of map, modify this map will not
56+
* @return The copy of flink configurations in the form of map, modifying this map will not
5757
* influence the original configuration object.
5858
*/
5959
Map<String, String> getFlinkConfigurationOptions();

‎flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/SqlGatewayRestEndpointFactory.java

Lines changed: 14 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,6 @@
2626
import org.apache.flink.table.gateway.api.endpoint.SqlGatewayEndpointFactoryUtils;
2727
import org.apache.flink.table.gateway.api.utils.SqlGatewayException;
2828

29-
import java.util.HashMap;
3029
import java.util.HashSet;
3130
import java.util.Map;
3231
import java.util.Set;
@@ -48,34 +47,35 @@ public SqlGatewayEndpoint createSqlGatewayEndpoint(Context context) {
4847
SqlGatewayEndpointFactoryUtils.createEndpointFactoryHelper(this, context);
4948
// Check that ADDRESS must be set
5049
endpointFactoryHelper.validate();
51-
Configuration config = rebuildRestEndpointOptions(context.getEndpointOptions());
50+
Configuration config =
51+
rebuildRestEndpointOptions(
52+
context.getEndpointOptions(), context.getFlinkConfigurationOptions());
5253
try {
5354
return new SqlGatewayRestEndpoint(config, context.getSqlGatewayService());
5455
} catch (Exception e) {
5556
throw new SqlGatewayException("Cannot start the rest endpoint.", e);
5657
}
5758
}
5859

59-
public static Configuration rebuildRestEndpointOptions(Map<String, String> configMap) {
60-
Map<String, String> effectiveConfigMap = new HashMap<>(configMap);
60+
public static Configuration rebuildRestEndpointOptions(
61+
Map<String, String> endpointConfigMap, Map<String, String> flinkConfigMap) {
62+
flinkConfigMap.put(RestOptions.ADDRESS.key(), endpointConfigMap.get(ADDRESS.key()));
6163

62-
effectiveConfigMap.put(RestOptions.ADDRESS.key(), configMap.get(ADDRESS.key()));
63-
64-
if (configMap.containsKey(BIND_ADDRESS.key())) {
65-
effectiveConfigMap.put(
66-
RestOptions.BIND_ADDRESS.key(), configMap.get(BIND_ADDRESS.key()));
64+
if (endpointConfigMap.containsKey(BIND_ADDRESS.key())) {
65+
flinkConfigMap.put(
66+
RestOptions.BIND_ADDRESS.key(), endpointConfigMap.get(BIND_ADDRESS.key()));
6767
}
6868

6969
// we need to override RestOptions.PORT anyway, to use a different default value
70-
effectiveConfigMap.put(
70+
flinkConfigMap.put(
7171
RestOptions.PORT.key(),
72-
configMap.getOrDefault(PORT.key(), PORT.defaultValue().toString()));
72+
endpointConfigMap.getOrDefault(PORT.key(), PORT.defaultValue().toString()));
7373

74-
if (configMap.containsKey(BIND_PORT.key())) {
75-
effectiveConfigMap.put(RestOptions.BIND_PORT.key(), configMap.get(BIND_PORT.key()));
74+
if (endpointConfigMap.containsKey(BIND_PORT.key())) {
75+
flinkConfigMap.put(RestOptions.BIND_PORT.key(), endpointConfigMap.get(BIND_PORT.key()));
7676
}
7777

78-
return Configuration.fromMap(effectiveConfigMap);
78+
return Configuration.fromMap(flinkConfigMap);
7979
}
8080

8181
@Override

‎flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/rest/util/SqlGatewayRestEndpointTestUtils.java

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,8 @@ public static Configuration getBaseConfig(Configuration flinkConf) {
4040
new SqlGatewayEndpointFactoryUtils.DefaultEndpointFactoryContext(
4141
null, flinkConf, getEndpointConfig(flinkConf, IDENTIFIER));
4242

43-
return rebuildRestEndpointOptions(context.getEndpointOptions());
43+
return rebuildRestEndpointOptions(
44+
context.getEndpointOptions(), context.getFlinkConfigurationOptions());
4445
}
4546

4647
/** Create the configuration generated from config.yaml. */

0 commit comments

Comments
 (0)