Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r162715070
--- Diff:
flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
---
@@ -376,4 +430,99 @@ public GetClusterStatusResponse getClusterStatus() {
public int getMaxSlots() {
return 0;
}
+
+
//-------------------------------------------------------------------------
+ // RestClient Helper
+
//-------------------------------------------------------------------------
+
+ private <M extends MessageHeaders<EmptyRequestBody, P, U>, U extends
MessageParameters, P extends ResponseBody> CompletableFuture<P>
+ sendRequest(M messageHeaders, U messageParameters)
throws IOException, LeaderNotAvailableException {
+ return sendRequest(messageHeaders, messageParameters,
EmptyRequestBody.getInstance());
+ }
+
+ private <M extends MessageHeaders<R, P, EmptyMessageParameters>, R
extends RequestBody, P extends ResponseBody> CompletableFuture<P>
+ sendRequest(M messageHeaders, R request) throws
IOException, LeaderNotAvailableException {
+ return sendRequest(messageHeaders,
EmptyMessageParameters.getInstance(), request);
+ }
+
+ private <M extends MessageHeaders<EmptyRequestBody, P,
EmptyMessageParameters>, P extends ResponseBody> CompletableFuture<P>
+ sendRequest(M messageHeaders) throws IOException,
LeaderNotAvailableException {
+ return sendRequest(messageHeaders,
EmptyMessageParameters.getInstance(), EmptyRequestBody.getInstance());
+ }
+
+ private <M extends MessageHeaders<R, P, U>, U extends
MessageParameters, R extends RequestBody, P extends ResponseBody>
CompletableFuture<P>
+ sendRequest(M messageHeaders, U messageParameters, R
request) throws IOException, LeaderNotAvailableException {
+ final URL restServerBaseUrl =
restServerLeaderHolder.getLeaderAddress();
+ return restClient.sendRequest(restServerBaseUrl.getHost(),
restServerBaseUrl.getPort(), messageHeaders, messageParameters, request);
+ }
+
+ private <M extends MessageHeaders<R, P, U>, U extends
MessageParameters, R extends RequestBody, P extends ResponseBody>
CompletableFuture<P>
+ sendRetryableRequest(M messageHeaders, U
messageParameters, R request, Predicate<Throwable> retryPredicate) {
+ return retry(() -> {
+ final URL restServerBaseUrl =
restServerLeaderHolder.getLeaderAddress();
--- End diff --
Good point.
---
