yuzelin commented on code in PR #21525:
URL: https://github.com/apache/flink/pull/21525#discussion_r1057502548
##########
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");
}
/** Test that requests of different version are routed to correct
handlers. */
@Test
void testVersionSelection() throws Exception {
- CompletableFuture<TestResponse> version1Response =
- restClient.sendRequest(
- serverAddress.getHostName(),
- serverAddress.getPort(),
- header1,
- EmptyMessageParameters.getInstance(),
- EmptyRequestBody.getInstance(),
- Collections.emptyList(),
- SqlGatewayRestAPIVersion.V0);
-
- TestResponse testResponse = version1Response.get(5, TimeUnit.SECONDS);
- assertThat(testResponse.getStatus()).isEqualTo("V0");
-
- CompletableFuture<TestResponse> version2Response =
- restClient.sendRequest(
- serverAddress.getHostName(),
- serverAddress.getPort(),
- header2,
- EmptyMessageParameters.getInstance(),
- EmptyRequestBody.getInstance(),
- Collections.emptyList(),
- SqlGatewayRestAPIVersion.V1);
- TestResponse testResponse2 = version2Response.get(5, TimeUnit.SECONDS);
- assertThat(testResponse2.getStatus()).isEqualTo("V1");
+ validateVersionSelection(header0, SqlGatewayRestAPIVersion.V0);
+ validateVersionSelection(header12, SqlGatewayRestAPIVersion.V1);
+ validateVersionSelection(header12, SqlGatewayRestAPIVersion.V2);
Review Comment:
Now I think this test is very similar to the test
'testSqlGatewayMessageHeaders', because both test the message is routed
correctly. So I think we can delete this test.
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
