yuzelin commented on code in PR #21525:
URL: https://github.com/apache/flink/pull/21525#discussion_r1057486097
##########
flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/rest/SqlGatewayRestEndpointITCase.java:
##########
@@ -58,154 +58,139 @@
import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
+import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
-import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
import java.util.function.Function;
+import java.util.stream.Collectors;
import static
org.apache.flink.table.gateway.rest.util.RestConfigUtils.getBaseConfig;
import static
org.apache.flink.table.gateway.rest.util.RestConfigUtils.getFlinkConfig;
+import static
org.apache.flink.table.gateway.rest.util.SqlGatewayRestClientAndEndpointUtils.TestRestClient.getTestRestClient;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** IT cases for {@link SqlGatewayRestEndpoint}. */
class SqlGatewayRestEndpointITCase {
- private static final SqlGatewayService service = null;
-
- private static RestServerEndpoint serverEndpoint;
- private static RestClient restClient;
+ private static SqlGatewayRestEndpoint serverEndpoint;
+ private static TestRestClient restClient;
private static InetSocketAddress serverAddress;
- private static TestBadCaseHandler testHandler;
- private static TestVersionSelectionHeaders1 header1;
- private static TestVersionSelectionHeaders2 header2;
private static TestBadCaseHeaders badCaseHeader;
- private static TestVersionHandler testVersionHandler1;
- private static TestVersionHandler testVersionHandler2;
+ private static TestBadCaseHandler testHandler;
+
+ private static TestVersionSelectionHeaders0 header0;
+ private static TestVersionSelectionHeaders12 header12;
+
+ private static TestVersionHandler testVersionHandler0;
+ private static TestVersionHandler testVersionHandler12;
private static Configuration config;
private static final Time timeout = Time.seconds(10L);
@BeforeEach
void setup() throws Exception {
// Test version cases
- header1 = new TestVersionSelectionHeaders1();
- header2 = new TestVersionSelectionHeaders2();
- testVersionHandler1 = new TestVersionHandler(service, header1);
- testVersionHandler2 = new TestVersionHandler(service, header2);
+ header0 = new TestVersionSelectionHeaders0();
+ header12 = new TestVersionSelectionHeaders12();
+ testVersionHandler0 = new TestVersionHandler(header0);
+ testVersionHandler12 = new TestVersionHandler(header12);
// Test exception cases
badCaseHeader = new TestBadCaseHeaders();
- testHandler = new TestBadCaseHandler(service);
+ testHandler = new TestBadCaseHandler();
// Init
final String address =
InetAddress.getLoopbackAddress().getHostAddress();
config = getBaseConfig(getFlinkConfig(address, address, "0"));
serverEndpoint =
- TestingSqlGatewayRestEndpoint.builder(config, service)
+ TestSqlGatewayRestEndpoint.builder(config)
.withHandler(badCaseHeader, testHandler)
- .withHandler(header1, testVersionHandler1)
- .withHandler(header2, testVersionHandler2)
+ .withHandler(header0, testVersionHandler0)
+ .withHandler(header12, testVersionHandler12)
.buildAndStart();
- restClient =
- new RestClient(
- config,
- Executors.newFixedThreadPool(
- 1, new
ExecutorThreadFactory("rest-client-thread-pool")));
+ restClient = getTestRestClient();
serverAddress = serverEndpoint.getServerAddress();
}
@AfterEach
void stop() throws Exception {
-
if (restClient != null) {
- restClient.shutdown(timeout);
+ restClient.shutdown();
restClient = null;
}
if (serverEndpoint != null) {
- serverEndpoint.closeAsync().get(timeout.getSize(),
timeout.getUnit());
+ serverEndpoint.stop();
serverEndpoint = null;
}
}
/** Test that {@link SqlGatewayMessageHeaders} can identify the version
correctly. */
@Test
void testSqlGatewayMessageHeaders() throws Exception {
- // The header only support V1, but send request by V0
+ // The header can't support V0, but sends request by V0
assertThatThrownBy(
() ->
restClient.sendRequest(
serverAddress.getHostName(),
serverAddress.getPort(),
- header2,
+ header12,
EmptyMessageParameters.getInstance(),
EmptyRequestBody.getInstance(),
Collections.emptyList(),
SqlGatewayRestAPIVersion.V0))
- .isInstanceOf(IllegalArgumentException.class);
-
- // The header only support V1, send request by V1
+ .satisfies(
+ FlinkAssertions.anyCauseMatches(
+ IllegalArgumentException.class,
+ String.format(
+ "The requested version V0 is not
supported by the request (method=%s URL=%s). Supported versions are: %s.",
+ header12.getHttpMethod(),
+ header12.getTargetRestEndpointURL(),
+
header12.getSupportedAPIVersions().stream()
+
.map(RestAPIVersion::getURLVersionPrefix)
+
.collect(Collectors.joining(",")))));
+
+ // The header only supports V0, sends request by V0
CompletableFuture<TestResponse> specifiedVersionResponse =
restClient.sendRequest(
serverAddress.getHostName(),
serverAddress.getPort(),
- header2,
+ header0,
EmptyMessageParameters.getInstance(),
EmptyRequestBody.getInstance(),
Collections.emptyList(),
- SqlGatewayRestAPIVersion.V1);
+ SqlGatewayRestAPIVersion.V0);
- TestResponse testResponse1 = specifiedVersionResponse.get(5,
TimeUnit.SECONDS);
- assertThat(testResponse1.getStatus()).isEqualTo("V1");
+ TestResponse testResponse1 =
+ specifiedVersionResponse.get(timeout.getSize(),
timeout.getUnit());
+ assertThat(testResponse1.getStatus()).isEqualTo("V0");
- // The header only support V1, send request by latest version V1
+ // The header supports V1 and V2, lets the client get the latest
version as default
CompletableFuture<TestResponse> unspecifiedVersionResponse =
restClient.sendRequest(
serverAddress.getHostName(),
serverAddress.getPort(),
- header2,
+ header12,
EmptyMessageParameters.getInstance(),
EmptyRequestBody.getInstance(),
Collections.emptyList());
- TestResponse testResponse2 = unspecifiedVersionResponse.get(5,
TimeUnit.SECONDS);
- assertThat(testResponse2.getStatus()).isEqualTo("V1");
+ TestResponse testResponse2 =
+ unspecifiedVersionResponse.get(timeout.getSize(),
timeout.getUnit());
+ assertThat(testResponse2.getStatus()).isEqualTo("V2");
Review Comment:
> This part is same to `testDefaultVersionRouting`? If so, I think we can
remove here.
No, it is to test getting the latest version of all the supported versions.
In `RestClient` line 315.
Default version routing is needed when the message doesn't provide version.
The message header here will provide the version.
> nit: BTW, it's better to introduce a case that API is only available in
the high version, which is the actual case. With the default version routing,
the request works.
Not needed because the RestClient can get the supported version, so we can
use the V0 case instead of introducing a new header.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
