You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GJL <gi...@git.apache.org> on 2018/01/18 17:18:39 UTC
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
GitHub user GJL opened a pull request:
https://github.com/apache/flink/pull/5312
[FLINK-8344][WIP] Add support for HA to RestClusterClient
WIP
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/GJL/flink rest-cluster-client-ha
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/5312.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #5312
----
commit ef720bd792dd8c3dc23e713c9d59761fd4278325
Author: gyao <ga...@...>
Date: 2018-01-15T11:13:25Z
[hotfix] Fix typo in AbstractCustomCommandLine Javadoc
commit 44d2c47259cbb7ab716045bdc420f6565c736787
Author: gyao <ga...@...>
Date: 2018-01-15T11:13:48Z
[hotfix] Fix checkstyle violations in ZooKeeperUtils
commit 02dc08b7939d696f32398acb1420b137dbe33e81
Author: gyao <ga...@...>
Date: 2018-01-15T11:26:47Z
[hotfix] Fix checkstyle violations in ZooKeeperLeaderElectionService
commit da087dd6d793aece2f0e622da3d3d4aa8c9fd1fa
Author: gyao <ga...@...>
Date: 2018-01-15T12:03:23Z
[hotfix] Simplify log message format
commit 95afc71ecee630b75f1c53e9e7449338a293eb00
Author: gyao <ga...@...>
Date: 2018-01-15T12:55:05Z
[hotfix] Unify argument validation in RestClusterClient
commit b6b0c97b0e364dfb093e0ad0ac756f1b51e684ca
Author: gyao <ga...@...>
Date: 2018-01-15T12:55:52Z
[hotfix] Fix wrong Javadoc in RestServerEndpoint
getRestAddress() returns a String instead of a CompletableFuture
commit 3567c6b6f5391aa47a214f3e1c7ca99d2f21d77c
Author: gyao <ga...@...>
Date: 2018-01-18T17:17:40Z
[FLINK-8344][flip6] WIP
----
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r162687806
--- Diff: flink-docs/src/main/java/org/apache/flink/docs/rest/RestAPIDocGenerator.java ---
@@ -285,13 +289,45 @@ private static String createMessageHtmlEntry(Class<?> messageClass, Class<?> emp
}
private DocumentingDispatcherRestEndpoint() {
- super(restConfig, dispatcherGatewayRetriever, config, handlerConfig, resourceManagerGatewayRetriever, executor, metricQueryServiceRetriever);
+ super(restConfig, dispatcherGatewayRetriever, config, handlerConfig, resourceManagerGatewayRetriever, executor, metricQueryServiceRetriever, NoOpElectionService.INSTANCE, NoOpFatalErrorHandler.INSTANCE);
}
@Override
public List<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> initializeHandlers(CompletableFuture<String> restAddressFuture) {
return super.initializeHandlers(restAddressFuture);
}
+
+ private enum NoOpElectionService implements LeaderElectionService {
--- End diff --
Nice, didn't know about this technique :-)
---
[GitHub] flink pull request #5312: [FLINK-8344][flip6] Add support for HA to RestClus...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r163776886
--- Diff: flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClientConfiguration.java ---
@@ -19,60 +19,73 @@
package org.apache.flink.client.program.rest;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.runtime.rest.RestClientConfiguration;
import org.apache.flink.util.ConfigurationException;
import org.apache.flink.util.Preconditions;
+import static org.apache.flink.util.Preconditions.checkArgument;
+
/**
* A configuration object for {@link RestClusterClient}s.
*/
public final class RestClusterClientConfiguration {
- private final String blobServerAddress;
-
private final RestClientConfiguration restClientConfiguration;
- private final String restServerAddress;
+ private final int awaitLeaderTimeout;
+
+ private final int retryMaxAttempts;
- private final int restServerPort;
+ private final int retryDelay;
private RestClusterClientConfiguration(
- String blobServerAddress,
- RestClientConfiguration endpointConfiguration,
- String restServerAddress,
- int restServerPort) {
- this.blobServerAddress = Preconditions.checkNotNull(blobServerAddress);
+ final RestClientConfiguration endpointConfiguration,
+ final int awaitLeaderTimeout,
--- End diff --
this could be a `long`. So far we always give ms timeouts the `long` type.
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r162689030
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java ---
@@ -251,6 +245,12 @@ protected void channelRead0(ChannelHandlerContext ctx, Object msg) {
ctx.close();
}
+ @Override
+ public void exceptionCaught(final ChannelHandlerContext ctx, final Throwable cause) throws Exception {
+ jsonFuture.completeExceptionally(cause);
--- End diff --
Nice catch :-)
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r163160849
--- Diff: flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java ---
@@ -327,13 +376,14 @@ public T getClusterId() {
}
private <R, A extends AsynchronouslyCreatedResource<R>> R waitForResource(
- final SupplierWithException<CompletableFuture<A>, IOException> resourceFutureSupplier)
+ final Supplier<CompletableFuture<A>> resourceFutureSupplier)
throws IOException, InterruptedException, ExecutionException, TimeoutException {
--- End diff --
removed.. also made the method non-blocking
---
[GitHub] flink pull request #5312: [FLINK-8344][flip6] Add support for HA to RestClus...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r164074612
--- Diff: flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClientConfiguration.java ---
@@ -19,60 +19,73 @@
package org.apache.flink.client.program.rest;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.runtime.rest.RestClientConfiguration;
import org.apache.flink.util.ConfigurationException;
import org.apache.flink.util.Preconditions;
+import static org.apache.flink.util.Preconditions.checkArgument;
+
/**
* A configuration object for {@link RestClusterClient}s.
*/
public final class RestClusterClientConfiguration {
- private final String blobServerAddress;
-
private final RestClientConfiguration restClientConfiguration;
- private final String restServerAddress;
+ private final int awaitLeaderTimeout;
+
+ private final int retryMaxAttempts;
- private final int restServerPort;
+ private final int retryDelay;
private RestClusterClientConfiguration(
- String blobServerAddress,
- RestClientConfiguration endpointConfiguration,
- String restServerAddress,
- int restServerPort) {
- this.blobServerAddress = Preconditions.checkNotNull(blobServerAddress);
+ final RestClientConfiguration endpointConfiguration,
+ final int awaitLeaderTimeout,
--- End diff --
changed all delays and timeouts to `long`
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r163160802
--- 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 --
done
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r163161066
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/standalone/StandaloneHaServices.java ---
@@ -132,4 +137,23 @@ public LeaderElectionService getJobManagerLeaderElectionService(JobID jobID) {
return new StandaloneLeaderElectionService();
}
}
+
+ @Override
+ public LeaderRetrievalService getRestServerLeaderRetriever() {
--- End diff --
renamed
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r163160672
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java ---
@@ -144,6 +154,15 @@ public WebMonitorEndpoint(
metricQueryServiceRetriever,
executor,
restConfiguration.getTimeout());
+
+ this.leaderElectionService = Preconditions.checkNotNull(leaderElectionService);
+ this.fatalErrorHandler = Preconditions.checkNotNull(fatalErrorHandler);
+ }
+
+ @Override
+ public void start() throws Exception {
+ super.start();
+ leaderElectionService.start(this);
--- End diff --
Added
```
try {
leaderElectionService.stop();
} catch (Exception e) {
log.warn("Error while stopping leaderElectionService", e);
}
```
to `shutdown` hook
---
[GitHub] flink pull request #5312: [FLINK-8344][flip6] Add support for HA to RestClus...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r163783813
--- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestClientTest.java ---
@@ -0,0 +1,102 @@
+/*
+ * 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.runtime.rest;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.RestOptions;
+import org.apache.flink.runtime.concurrent.Executors;
+import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.EmptyResponseBody;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.util.ExceptionUtils;
+
+import org.apache.flink.shaded.netty4.io.netty.channel.ConnectTimeoutException;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import org.junit.Test;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Tests for {@link RestClient}.
+ */
+public class RestClientTest {
--- End diff --
`extends TestLogger` missing
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by satybald <gi...@git.apache.org>.
Github user satybald commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r162814676
--- Diff: flink-clients/src/main/java/org/apache/flink/client/program/rest/LeaderHolder.java ---
@@ -0,0 +1,74 @@
+/*
+ * 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.client.program.rest;
+
+import javax.annotation.Nullable;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+class LeaderHolder<T> {
+
+ private final String leaderType;
+
+ private final long awaitLeaderTimeoutMs;
+
+ private volatile T leaderAddress;
+
+ LeaderHolder(final String leaderType, final long awaitLeaderTimeoutMs) {
+ checkArgument(awaitLeaderTimeoutMs >= 0, "awaitLeaderTimeoutMs must be greater than or equal to 0");
+ this.leaderType = leaderType;
+ this.awaitLeaderTimeoutMs = awaitLeaderTimeoutMs;
+ }
+
+ public T getLeaderAddress() throws LeaderNotAvailableException {
+ final long startTime = System.nanoTime();
+ T result;
+ synchronized (this) {
+ long sleptTimeMs = 0;
+ while (leaderAddress == null && sleptTimeMs < awaitLeaderTimeoutMs) {
+ try {
+ wait(awaitLeaderTimeoutMs - sleptTimeMs);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ sleptTimeMs = (System.nanoTime() - startTime) / 1000000;
--- End diff --
what do you think about adding underscores for readability? i.e. 1_000_000
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by GJL <gi...@git.apache.org>.
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.
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r163160938
--- 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();
+ return restClient.sendRequest(restServerBaseUrl.getHost(), restServerBaseUrl.getPort(), messageHeaders, messageParameters, request);
+ }, retryPredicate);
+ }
+
+ private <C> CompletableFuture<C> retry(
+ CheckedSupplier<CompletableFuture<C>> operation,
+ Predicate<Throwable> retryPredicate) {
+ return FutureUtils.retryWithDelay(
+ CheckedSupplier.unchecked(operation),
+ MAX_RETRIES,
+ RETRY_DELAY,
+ retryPredicate,
+ new ScheduledExecutorServiceAdapter(retryExecutorService));
+ }
+
+ private static Predicate<Throwable> isTimeoutException() {
+ return (throwable) ->
+ ExceptionUtils.findThrowable(throwable, java.net.ConnectException.class).isPresent() ||
+ ExceptionUtils.findThrowable(throwable, java.net.SocketTimeoutException.class).isPresent() ||
+ ExceptionUtils.findThrowable(throwable, ConnectTimeoutException.class).isPresent() ||
+ ExceptionUtils.findThrowable(throwable, IOException.class).isPresent();
+ }
+
+ private static Predicate<Throwable> isHttpStatusUnsuccessfulException() {
+ return (throwable) -> ExceptionUtils.findThrowable(throwable, RestClientException.class)
+ .map(restClientException -> {
+ final int code = restClientException.getHttpResponseStatus().code();
+ return code < 200 || code > 299;
+ })
+ .orElse(false);
+ }
+
+ private abstract class RestClusterClientLeaderRetrievalListener implements LeaderRetrievalListener {
+ @Override
+ public final void handleError(final Exception exception) {
+ log.error("Exception in LeaderRetrievalListener", exception);
+ shutdown();
--- End diff --
Using `LeaderRetriever` now
---
[GitHub] flink pull request #5312: [FLINK-8344][flip6] Add support for HA to RestClus...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r163776982
--- Diff: flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientConfigurationTest.java ---
@@ -0,0 +1,51 @@
+/*
+ * 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.client.program.rest;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.RestOptions;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests for {@link RestClusterClientConfiguration}.
+ */
+public class RestClusterClientConfigurationTest {
--- End diff --
`extends TestLogger` missing
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r162694907
--- 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();
+ return restClient.sendRequest(restServerBaseUrl.getHost(), restServerBaseUrl.getPort(), messageHeaders, messageParameters, request);
+ }, retryPredicate);
+ }
+
+ private <C> CompletableFuture<C> retry(
+ CheckedSupplier<CompletableFuture<C>> operation,
+ Predicate<Throwable> retryPredicate) {
+ return FutureUtils.retryWithDelay(
+ CheckedSupplier.unchecked(operation),
+ MAX_RETRIES,
+ RETRY_DELAY,
+ retryPredicate,
+ new ScheduledExecutorServiceAdapter(retryExecutorService));
+ }
+
+ private static Predicate<Throwable> isTimeoutException() {
+ return (throwable) ->
+ ExceptionUtils.findThrowable(throwable, java.net.ConnectException.class).isPresent() ||
+ ExceptionUtils.findThrowable(throwable, java.net.SocketTimeoutException.class).isPresent() ||
+ ExceptionUtils.findThrowable(throwable, ConnectTimeoutException.class).isPresent() ||
+ ExceptionUtils.findThrowable(throwable, IOException.class).isPresent();
+ }
+
+ private static Predicate<Throwable> isHttpStatusUnsuccessfulException() {
+ return (throwable) -> ExceptionUtils.findThrowable(throwable, RestClientException.class)
+ .map(restClientException -> {
+ final int code = restClientException.getHttpResponseStatus().code();
+ return code < 200 || code > 299;
+ })
+ .orElse(false);
+ }
+
+ private abstract class RestClusterClientLeaderRetrievalListener implements LeaderRetrievalListener {
+ @Override
+ public final void handleError(final Exception exception) {
+ log.error("Exception in LeaderRetrievalListener", exception);
+ shutdown();
--- End diff --
Yes, we have to fail all ongoing operations in order to unblock the `RestClusterClient` caller.
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r162696312
--- 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();
+ return restClient.sendRequest(restServerBaseUrl.getHost(), restServerBaseUrl.getPort(), messageHeaders, messageParameters, request);
+ }, retryPredicate);
+ }
+
+ private <C> CompletableFuture<C> retry(
+ CheckedSupplier<CompletableFuture<C>> operation,
+ Predicate<Throwable> retryPredicate) {
+ return FutureUtils.retryWithDelay(
+ CheckedSupplier.unchecked(operation),
+ MAX_RETRIES,
+ RETRY_DELAY,
+ retryPredicate,
+ new ScheduledExecutorServiceAdapter(retryExecutorService));
+ }
+
+ private static Predicate<Throwable> isTimeoutException() {
+ return (throwable) ->
+ ExceptionUtils.findThrowable(throwable, java.net.ConnectException.class).isPresent() ||
+ ExceptionUtils.findThrowable(throwable, java.net.SocketTimeoutException.class).isPresent() ||
+ ExceptionUtils.findThrowable(throwable, ConnectTimeoutException.class).isPresent() ||
+ ExceptionUtils.findThrowable(throwable, IOException.class).isPresent();
+ }
+
+ private static Predicate<Throwable> isHttpStatusUnsuccessfulException() {
+ return (throwable) -> ExceptionUtils.findThrowable(throwable, RestClientException.class)
+ .map(restClientException -> {
+ final int code = restClientException.getHttpResponseStatus().code();
+ return code < 200 || code > 299;
+ })
+ .orElse(false);
+ }
+
+ private abstract class RestClusterClientLeaderRetrievalListener implements LeaderRetrievalListener {
+ @Override
+ public final void handleError(final Exception exception) {
+ log.error("Exception in LeaderRetrievalListener", exception);
+ shutdown();
--- End diff --
Maybe we could store the latest created future in `waitForResource` which we complete exceptionally here.
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r162893190
--- Diff: flink-clients/src/main/java/org/apache/flink/client/program/rest/LeaderHolder.java ---
@@ -0,0 +1,74 @@
+/*
+ * 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.client.program.rest;
+
+import javax.annotation.Nullable;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+class LeaderHolder<T> {
+
+ private final String leaderType;
+
+ private final long awaitLeaderTimeoutMs;
+
+ private volatile T leaderAddress;
+
+ LeaderHolder(final String leaderType, final long awaitLeaderTimeoutMs) {
+ checkArgument(awaitLeaderTimeoutMs >= 0, "awaitLeaderTimeoutMs must be greater than or equal to 0");
+ this.leaderType = leaderType;
+ this.awaitLeaderTimeoutMs = awaitLeaderTimeoutMs;
+ }
+
+ public T getLeaderAddress() throws LeaderNotAvailableException {
+ final long startTime = System.nanoTime();
+ T result;
+ synchronized (this) {
+ long sleptTimeMs = 0;
+ while (leaderAddress == null && sleptTimeMs < awaitLeaderTimeoutMs) {
+ try {
+ wait(awaitLeaderTimeoutMs - sleptTimeMs);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ sleptTimeMs = (System.nanoTime() - startTime) / 1000000;
--- End diff --
That's better but I am going to remove this class and use `LeaderRetriever` instead.
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r162581631
--- 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();
+ return restClient.sendRequest(restServerBaseUrl.getHost(), restServerBaseUrl.getPort(), messageHeaders, messageParameters, request);
+ }, retryPredicate);
+ }
+
+ private <C> CompletableFuture<C> retry(
+ CheckedSupplier<CompletableFuture<C>> operation,
+ Predicate<Throwable> retryPredicate) {
+ return FutureUtils.retryWithDelay(
+ CheckedSupplier.unchecked(operation),
+ MAX_RETRIES,
+ RETRY_DELAY,
+ retryPredicate,
+ new ScheduledExecutorServiceAdapter(retryExecutorService));
+ }
+
+ private static Predicate<Throwable> isTimeoutException() {
+ return (throwable) ->
+ ExceptionUtils.findThrowable(throwable, java.net.ConnectException.class).isPresent() ||
+ ExceptionUtils.findThrowable(throwable, java.net.SocketTimeoutException.class).isPresent() ||
+ ExceptionUtils.findThrowable(throwable, ConnectTimeoutException.class).isPresent() ||
+ ExceptionUtils.findThrowable(throwable, IOException.class).isPresent();
+ }
+
+ private static Predicate<Throwable> isHttpStatusUnsuccessfulException() {
+ return (throwable) -> ExceptionUtils.findThrowable(throwable, RestClientException.class)
+ .map(restClientException -> {
+ final int code = restClientException.getHttpResponseStatus().code();
+ return code < 200 || code > 299;
+ })
+ .orElse(false);
+ }
+
+ private abstract class RestClusterClientLeaderRetrievalListener implements LeaderRetrievalListener {
+ @Override
+ public final void handleError(final Exception exception) {
+ log.error("Exception in LeaderRetrievalListener", exception);
+ shutdown();
--- End diff --
It's not good to call `shutdown()` here because some code in `CliFrontend` may wait forever due to some threadpools being shutdown. One could just null the leader address and let the rest client fail.
---
[GitHub] flink pull request #5312: [FLINK-8344][flip6] Add support for HA to RestClus...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r163227450
--- Diff: flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java ---
@@ -596,11 +596,11 @@ public int run(String[] args) throws CliArgsException, FlinkException {
//------------------ ClusterClient deployed, handle connection details
yarnApplicationId = clusterClient.getClusterId();
- String jobManagerAddress =
- clusterClient.getJobManagerAddress().getAddress().getHostName() +
- ':' + clusterClient.getJobManagerAddress().getPort();
+// String jobManagerAddress =
+// clusterClient.getJobManagerAddress().getAddress().getHostName() +
--- End diff --
had to add it back because `YARNSessionCapacitySchedulerITCase` is relying on it
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r162412749
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/standalone/StandaloneHaServices.java ---
@@ -132,4 +137,23 @@ public LeaderElectionService getJobManagerLeaderElectionService(JobID jobID) {
return new StandaloneLeaderElectionService();
}
}
+
+ @Override
+ public LeaderRetrievalService getRestServerLeaderRetriever() {
--- End diff --
Since `WebMonitorEndpoint` is the instance taking part in leader election currently, this should be called `getWebmonitorLeaderRetriever`.
---
[GitHub] flink pull request #5312: [FLINK-8344][flip6] Add support for HA to RestClus...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r164074562
--- Diff: flink-core/src/main/java/org/apache/flink/configuration/RestOptions.java ---
@@ -41,4 +41,34 @@
public static final ConfigOption<Integer> REST_PORT =
key("rest.port")
.defaultValue(9067);
+
+ /**
+ * The time in ms to wait for the leader address, e.g., Dispatcher or WebMonitorEndpoint.
+ */
+ public static final ConfigOption<Integer> AWAIT_LEADER_TIMEOUT =
+ key("rest.await-leader-timeout")
+ .defaultValue(30_000);
+
+ /**
+ * The number of attempts to retry a retryable operation.
+ * @see #RETRY_DELAY
+ */
+ public static final ConfigOption<Integer> RETRY_MAX_ATTEMPTS =
+ key("rest.retry.max-attempts")
+ .defaultValue(20);
+
+ /**
+ * The time in ms to wait between retries.
+ * @see #RETRY_MAX_ATTEMPTS
+ */
+ public static final ConfigOption<Integer> RETRY_DELAY =
+ key("rest.retry.delay")
+ .defaultValue(3_000);
+
+ /**
+ * The maximum time in ms to establish a TCP connection.
+ */
+ public static final ConfigOption<Integer> CONNECTION_TIMEOUT =
+ key("rest.connection-timeout")
+ .defaultValue(15_000);
--- End diff --
Added for all `ConfigOptions` in `RestOptions`
---
[GitHub] flink pull request #5312: [FLINK-8344][flip6] Add support for HA to RestClus...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/5312
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r162413303
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestClient.java ---
@@ -251,6 +245,12 @@ protected void channelRead0(ChannelHandlerContext ctx, Object msg) {
ctx.close();
}
+ @Override
+ public void exceptionCaught(final ChannelHandlerContext ctx, final Throwable cause) throws Exception {
+ jsonFuture.completeExceptionally(cause);
--- End diff --
Without this, the `CliFrontend` can block indefintely for some operations. For example:
```
CompletableFuture<Collection<JobStatusMessage>> jobDetailsFuture = clusterClient.listJobs();
logAndSysout("Waiting for response...");
jobDetails = jobDetailsFuture.get();
```
This way, the exception gets propagated to the client.
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r162695596
--- 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();
+ return restClient.sendRequest(restServerBaseUrl.getHost(), restServerBaseUrl.getPort(), messageHeaders, messageParameters, request);
+ }, retryPredicate);
+ }
+
+ private <C> CompletableFuture<C> retry(
+ CheckedSupplier<CompletableFuture<C>> operation,
+ Predicate<Throwable> retryPredicate) {
+ return FutureUtils.retryWithDelay(
+ CheckedSupplier.unchecked(operation),
+ MAX_RETRIES,
+ RETRY_DELAY,
+ retryPredicate,
+ new ScheduledExecutorServiceAdapter(retryExecutorService));
+ }
+
+ private static Predicate<Throwable> isTimeoutException() {
+ return (throwable) ->
+ ExceptionUtils.findThrowable(throwable, java.net.ConnectException.class).isPresent() ||
+ ExceptionUtils.findThrowable(throwable, java.net.SocketTimeoutException.class).isPresent() ||
+ ExceptionUtils.findThrowable(throwable, ConnectTimeoutException.class).isPresent() ||
+ ExceptionUtils.findThrowable(throwable, IOException.class).isPresent();
+ }
+
+ private static Predicate<Throwable> isHttpStatusUnsuccessfulException() {
+ return (throwable) -> ExceptionUtils.findThrowable(throwable, RestClientException.class)
+ .map(restClientException -> {
+ final int code = restClientException.getHttpResponseStatus().code();
+ return code < 200 || code > 299;
+ })
+ .orElse(false);
+ }
+
+ private abstract class RestClusterClientLeaderRetrievalListener implements LeaderRetrievalListener {
+ @Override
+ public final void handleError(final Exception exception) {
+ log.error("Exception in LeaderRetrievalListener", exception);
+ shutdown();
--- End diff --
E.g. in `FutureUtils`, I think we should catch `RejectedExecutionExceptions` to fail the `resultFuture`.
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r162414419
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionService.java ---
@@ -412,4 +413,11 @@ protected void handleStateChange(ConnectionState newState) {
public void unhandledError(String message, Throwable e) {
leaderContender.handleError(new FlinkException("Unhandled error in ZooKeeperLeaderElectionService: " + message, e));
}
+
+ @Override
+ public String toString() {
--- End diff --
Added because the instance logs itself:
`LOG.info("Starting ZooKeeperLeaderElectionService {}.", this);`
---
[GitHub] flink pull request #5312: [FLINK-8344][flip6] Add support for HA to RestClus...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r164074500
--- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestClientTest.java ---
@@ -0,0 +1,102 @@
+/*
+ * 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.runtime.rest;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.RestOptions;
+import org.apache.flink.runtime.concurrent.Executors;
+import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.EmptyResponseBody;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.util.ExceptionUtils;
+
+import org.apache.flink.shaded.netty4.io.netty.channel.ConnectTimeoutException;
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import org.junit.Test;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Tests for {@link RestClient}.
+ */
+public class RestClientTest {
--- End diff --
added
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r162416324
--- Diff: flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java ---
@@ -327,13 +376,14 @@ public T getClusterId() {
}
private <R, A extends AsynchronouslyCreatedResource<R>> R waitForResource(
- final SupplierWithException<CompletableFuture<A>, IOException> resourceFutureSupplier)
+ final Supplier<CompletableFuture<A>> resourceFutureSupplier)
throws IOException, InterruptedException, ExecutionException, TimeoutException {
A asynchronouslyCreatedResource;
long attempt = 0;
while (true) {
final CompletableFuture<A> responseFuture = resourceFutureSupplier.get();
--- End diff --
Currently the supplied future in `resourceFutureSupplier` will be retried with `FutureUtils` but the retry could also simply move into the `while(true)` loop. However, this would not work well if we wanted to retry other operations such as list jobs because the method returns a `CompletableFuture`:
`CompletableFuture<Collection<JobStatusMessage>> listJobs()`
---
[GitHub] flink pull request #5312: [FLINK-8344][flip6] Add support for HA to RestClus...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r164074605
--- Diff: flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClientConfiguration.java ---
@@ -19,60 +19,73 @@
package org.apache.flink.client.program.rest;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.runtime.rest.RestClientConfiguration;
import org.apache.flink.util.ConfigurationException;
import org.apache.flink.util.Preconditions;
+import static org.apache.flink.util.Preconditions.checkArgument;
+
/**
* A configuration object for {@link RestClusterClient}s.
*/
public final class RestClusterClientConfiguration {
- private final String blobServerAddress;
-
private final RestClientConfiguration restClientConfiguration;
- private final String restServerAddress;
+ private final int awaitLeaderTimeout;
+
+ private final int retryMaxAttempts;
- private final int restServerPort;
+ private final int retryDelay;
private RestClusterClientConfiguration(
- String blobServerAddress,
- RestClientConfiguration endpointConfiguration,
- String restServerAddress,
- int restServerPort) {
- this.blobServerAddress = Preconditions.checkNotNull(blobServerAddress);
+ final RestClientConfiguration endpointConfiguration,
+ final int awaitLeaderTimeout,
+ final int retryMaxAttempts,
+ final int retryDelay) {
--- End diff --
changed all delays and timeouts to `long`
---
[GitHub] flink pull request #5312: [FLINK-8344][flip6] Add support for HA to RestClus...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r164074572
--- Diff: flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientConfigurationTest.java ---
@@ -0,0 +1,51 @@
+/*
+ * 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.client.program.rest;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.RestOptions;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests for {@link RestClusterClientConfiguration}.
+ */
+public class RestClusterClientConfigurationTest {
--- End diff --
added
---
[GitHub] flink issue #5312: [FLINK-8344][flip6] Add support for HA to RestClusterClie...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on the issue:
https://github.com/apache/flink/pull/5312
Thanks for the good work @GJL. Merging this PR.
---
[GitHub] flink issue #5312: [FLINK-8344][flip6] Add support for HA to RestClusterClie...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on the issue:
https://github.com/apache/flink/pull/5312
Thanks man 👍
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r162687082
--- Diff: flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClientConfiguration.java ---
@@ -30,49 +28,20 @@
*/
public final class RestClusterClientConfiguration {
--- End diff --
This class seems to be no longer really needed.
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r162693549
--- Diff: flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java ---
@@ -327,13 +376,14 @@ public T getClusterId() {
}
private <R, A extends AsynchronouslyCreatedResource<R>> R waitForResource(
- final SupplierWithException<CompletableFuture<A>, IOException> resourceFutureSupplier)
+ final Supplier<CompletableFuture<A>> resourceFutureSupplier)
throws IOException, InterruptedException, ExecutionException, TimeoutException {
--- End diff --
`IOException` and `TimeoutException` not thrown.
---
[GitHub] flink pull request #5312: [FLINK-8344][flip6] Add support for HA to RestClus...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r163775574
--- Diff: flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java ---
@@ -258,89 +312,96 @@ public String cancelWithSavepoint(JobID jobId, @Nullable String savepointDirecto
final CompletableFuture<SavepointTriggerResponseBody> responseFuture;
- try {
- responseFuture = restClient.sendRequest(
- restClusterClientConfiguration.getRestServerAddress(),
- restClusterClientConfiguration.getRestServerPort(),
- savepointTriggerHeaders,
- savepointTriggerMessageParameters,
- new SavepointTriggerRequestBody(savepointDirectory));
- } catch (IOException e) {
- throw new FlinkException("Could not send trigger savepoint request to Flink cluster.", e);
- }
+ responseFuture = sendRequest(
+ savepointTriggerHeaders,
+ savepointTriggerMessageParameters,
+ new SavepointTriggerRequestBody(savepointDirectory));
- return responseFuture.thenApply(savepointTriggerResponseBody -> {
+ return responseFuture.thenCompose(savepointTriggerResponseBody -> {
final SavepointTriggerId savepointTriggerId = savepointTriggerResponseBody.getSavepointTriggerId();
- final SavepointInfo savepointInfo;
- try {
- savepointInfo = waitForSavepointCompletion(jobId, savepointTriggerId);
- } catch (Exception e) {
- throw new CompletionException(e);
- }
+ return waitForSavepointCompletion(jobId, savepointTriggerId);
+ }).thenApply(savepointInfo -> {
if (savepointInfo.getFailureCause() != null) {
throw new CompletionException(savepointInfo.getFailureCause());
}
return savepointInfo.getLocation();
});
}
- private SavepointInfo waitForSavepointCompletion(
+ private CompletableFuture<SavepointInfo> waitForSavepointCompletion(
final JobID jobId,
- final SavepointTriggerId savepointTriggerId) throws Exception {
+ final SavepointTriggerId savepointTriggerId) {
return waitForResource(() -> {
final SavepointStatusHeaders savepointStatusHeaders = SavepointStatusHeaders.getInstance();
final SavepointStatusMessageParameters savepointStatusMessageParameters =
savepointStatusHeaders.getUnresolvedMessageParameters();
savepointStatusMessageParameters.jobIdPathParameter.resolve(jobId);
savepointStatusMessageParameters.savepointTriggerIdPathParameter.resolve(savepointTriggerId);
- return restClient.sendRequest(
- restClusterClientConfiguration.getRestServerAddress(),
- restClusterClientConfiguration.getRestServerPort(),
+ return sendRetryableRequest(
savepointStatusHeaders,
- savepointStatusMessageParameters
- );
+ savepointStatusMessageParameters,
+ EmptyRequestBody.getInstance(),
+ isConnectionProblemException());
});
}
@Override
public CompletableFuture<Collection<JobStatusMessage>> listJobs() throws Exception {
- JobsOverviewHeaders headers = JobsOverviewHeaders.getInstance();
- CompletableFuture<MultipleJobsDetails> jobDetailsFuture = restClient.sendRequest(
- restClusterClientConfiguration.getRestServerAddress(),
- restClusterClientConfiguration.getRestServerPort(),
- headers
- );
- return jobDetailsFuture
+ return sendRequest(JobsOverviewHeaders.getInstance())
.thenApply(
- (MultipleJobsDetails multipleJobsDetails) -> {
- final Collection<JobDetails> jobDetails = multipleJobsDetails.getJobs();
- Collection<JobStatusMessage> flattenedDetails = new ArrayList<>(jobDetails.size());
- jobDetails.forEach(detail -> flattenedDetails.add(new JobStatusMessage(detail.getJobId(), detail.getJobName(), detail.getStatus(), detail.getStartTime())));
-
- return flattenedDetails;
- });
+ (multipleJobsDetails) -> multipleJobsDetails
+ .getJobs()
+ .stream()
+ .map(detail -> new JobStatusMessage(
+ detail.getJobId(),
+ detail.getJobName(),
+ detail.getStatus(),
+ detail.getStartTime()))
+ .collect(Collectors.toList()));
}
@Override
public T getClusterId() {
return clusterId;
}
- private <R, A extends AsynchronouslyCreatedResource<R>> R waitForResource(
- final SupplierWithException<CompletableFuture<A>, IOException> resourceFutureSupplier)
- throws IOException, InterruptedException, ExecutionException, TimeoutException {
- A asynchronouslyCreatedResource;
- long attempt = 0;
- while (true) {
- final CompletableFuture<A> responseFuture = resourceFutureSupplier.get();
- asynchronouslyCreatedResource = responseFuture.get(timeout.toMillis(), TimeUnit.MILLISECONDS);
- if (asynchronouslyCreatedResource.queueStatus().getId() == QueueStatus.Id.COMPLETED) {
- break;
+ /**
+ * Polls a {@code AsynchronouslyCreatedResource} until its
+ * {@link AsynchronouslyCreatedResource#queueStatus() QueueStatus} becomes
+ * {@link QueueStatus.Id#COMPLETED COMPLETED}. This method returns a {@code CompletableFuture}
+ * which completes with {@link AsynchronouslyCreatedResource#resource()}.
+ *
+ * @param resourceFutureSupplier The operation which polls for the
+ * {@code AsynchronouslyCreatedResource}.
+ * @param <R> The type of the resource.
+ * @param <A> The type of the {@code AsynchronouslyCreatedResource}.
+ * @return A {@code CompletableFuture} delivering the resource.
+ */
+ private <R, A extends AsynchronouslyCreatedResource<R>> CompletableFuture<R> waitForResource(
--- End diff --
I think this method could be renamed into something like `createResourcePollingFuture` or so
---
[GitHub] flink pull request #5312: [FLINK-8344][flip6] Add support for HA to RestClus...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r163775424
--- Diff: flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java ---
@@ -258,89 +312,96 @@ public String cancelWithSavepoint(JobID jobId, @Nullable String savepointDirecto
final CompletableFuture<SavepointTriggerResponseBody> responseFuture;
- try {
- responseFuture = restClient.sendRequest(
- restClusterClientConfiguration.getRestServerAddress(),
- restClusterClientConfiguration.getRestServerPort(),
- savepointTriggerHeaders,
- savepointTriggerMessageParameters,
- new SavepointTriggerRequestBody(savepointDirectory));
- } catch (IOException e) {
- throw new FlinkException("Could not send trigger savepoint request to Flink cluster.", e);
- }
+ responseFuture = sendRequest(
+ savepointTriggerHeaders,
+ savepointTriggerMessageParameters,
+ new SavepointTriggerRequestBody(savepointDirectory));
- return responseFuture.thenApply(savepointTriggerResponseBody -> {
+ return responseFuture.thenCompose(savepointTriggerResponseBody -> {
final SavepointTriggerId savepointTriggerId = savepointTriggerResponseBody.getSavepointTriggerId();
- final SavepointInfo savepointInfo;
- try {
- savepointInfo = waitForSavepointCompletion(jobId, savepointTriggerId);
- } catch (Exception e) {
- throw new CompletionException(e);
- }
+ return waitForSavepointCompletion(jobId, savepointTriggerId);
+ }).thenApply(savepointInfo -> {
if (savepointInfo.getFailureCause() != null) {
throw new CompletionException(savepointInfo.getFailureCause());
}
return savepointInfo.getLocation();
});
}
- private SavepointInfo waitForSavepointCompletion(
+ private CompletableFuture<SavepointInfo> waitForSavepointCompletion(
--- End diff --
Isn't this more `getSavepointFuture` instead of waiting? Waiting indicates for me that I wait for the completion of a future.
---
[GitHub] flink issue #5312: [FLINK-8344][WIP] Add support for HA to RestClusterClient
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on the issue:
https://github.com/apache/flink/pull/5312
cc: @tillrohrmann
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r162699002
--- 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 --
This is a blocking operation which can take potentially quite long. I think it would be better to return a future (by using `LeaderRetriever` for example) and register a `thenCompose` which executes the `sendRequest` call.
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r162699368
--- 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();
+ return restClient.sendRequest(restServerBaseUrl.getHost(), restServerBaseUrl.getPort(), messageHeaders, messageParameters, request);
+ }, retryPredicate);
+ }
+
+ private <C> CompletableFuture<C> retry(
+ CheckedSupplier<CompletableFuture<C>> operation,
+ Predicate<Throwable> retryPredicate) {
+ return FutureUtils.retryWithDelay(
+ CheckedSupplier.unchecked(operation),
+ MAX_RETRIES,
+ RETRY_DELAY,
+ retryPredicate,
+ new ScheduledExecutorServiceAdapter(retryExecutorService));
+ }
+
+ private static Predicate<Throwable> isTimeoutException() {
+ return (throwable) ->
+ ExceptionUtils.findThrowable(throwable, java.net.ConnectException.class).isPresent() ||
+ ExceptionUtils.findThrowable(throwable, java.net.SocketTimeoutException.class).isPresent() ||
+ ExceptionUtils.findThrowable(throwable, ConnectTimeoutException.class).isPresent() ||
+ ExceptionUtils.findThrowable(throwable, IOException.class).isPresent();
+ }
+
+ private static Predicate<Throwable> isHttpStatusUnsuccessfulException() {
+ return (throwable) -> ExceptionUtils.findThrowable(throwable, RestClientException.class)
+ .map(restClientException -> {
+ final int code = restClientException.getHttpResponseStatus().code();
+ return code < 200 || code > 299;
+ })
+ .orElse(false);
+ }
+
+ private abstract class RestClusterClientLeaderRetrievalListener implements LeaderRetrievalListener {
+ @Override
+ public final void handleError(final Exception exception) {
+ log.error("Exception in LeaderRetrievalListener", exception);
+ shutdown();
--- End diff --
By using the `LeaderRetriever`, this should actually be covered.
---
[GitHub] flink pull request #5312: [FLINK-8344][flip6] Add support for HA to RestClus...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r164075001
--- Diff: flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java ---
@@ -258,89 +312,96 @@ public String cancelWithSavepoint(JobID jobId, @Nullable String savepointDirecto
final CompletableFuture<SavepointTriggerResponseBody> responseFuture;
- try {
- responseFuture = restClient.sendRequest(
- restClusterClientConfiguration.getRestServerAddress(),
- restClusterClientConfiguration.getRestServerPort(),
- savepointTriggerHeaders,
- savepointTriggerMessageParameters,
- new SavepointTriggerRequestBody(savepointDirectory));
- } catch (IOException e) {
- throw new FlinkException("Could not send trigger savepoint request to Flink cluster.", e);
- }
+ responseFuture = sendRequest(
+ savepointTriggerHeaders,
+ savepointTriggerMessageParameters,
+ new SavepointTriggerRequestBody(savepointDirectory));
- return responseFuture.thenApply(savepointTriggerResponseBody -> {
+ return responseFuture.thenCompose(savepointTriggerResponseBody -> {
final SavepointTriggerId savepointTriggerId = savepointTriggerResponseBody.getSavepointTriggerId();
- final SavepointInfo savepointInfo;
- try {
- savepointInfo = waitForSavepointCompletion(jobId, savepointTriggerId);
- } catch (Exception e) {
- throw new CompletionException(e);
- }
+ return waitForSavepointCompletion(jobId, savepointTriggerId);
+ }).thenApply(savepointInfo -> {
if (savepointInfo.getFailureCause() != null) {
throw new CompletionException(savepointInfo.getFailureCause());
}
return savepointInfo.getLocation();
});
}
- private SavepointInfo waitForSavepointCompletion(
+ private CompletableFuture<SavepointInfo> waitForSavepointCompletion(
--- End diff --
It's now `pollSavepointAsync`
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r163160900
--- Diff: flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java ---
@@ -61,46 +69,77 @@
import org.apache.flink.runtime.rest.messages.job.savepoints.SavepointTriggerResponseBody;
import org.apache.flink.runtime.rest.messages.queue.AsynchronouslyCreatedResource;
import org.apache.flink.runtime.rest.messages.queue.QueueStatus;
+import org.apache.flink.runtime.rest.util.RestClientException;
import org.apache.flink.runtime.util.ExecutorThreadFactory;
+import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.ExecutorUtils;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.SerializedThrowable;
-import org.apache.flink.util.function.SupplierWithException;
+import org.apache.flink.util.function.CheckedSupplier;
+
+import org.apache.flink.shaded.netty4.io.netty.channel.ConnectTimeoutException;
+
+import akka.actor.AddressFromURIString;
import javax.annotation.Nullable;
import java.io.IOException;
import java.net.InetSocketAddress;
+import java.net.MalformedURLException;
import java.net.URL;
-import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
-import static java.util.Objects.requireNonNull;
+import scala.Option;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
/**
* A {@link ClusterClient} implementation that communicates via HTTP REST requests.
*/
public class RestClusterClient<T> extends ClusterClient<T> {
+ private static final long AWAIT_LEADER_TIMEOUT = 10_000;
+
+ private static final int MAX_RETRIES = 20;
+
+ private static final Time RETRY_DELAY = Time.seconds(3);
--- End diff --
Moved to `RestOptions`
---
[GitHub] flink pull request #5312: [FLINK-8344][flip6] Add support for HA to RestClus...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r164074804
--- Diff: flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java ---
@@ -258,89 +312,96 @@ public String cancelWithSavepoint(JobID jobId, @Nullable String savepointDirecto
final CompletableFuture<SavepointTriggerResponseBody> responseFuture;
- try {
- responseFuture = restClient.sendRequest(
- restClusterClientConfiguration.getRestServerAddress(),
- restClusterClientConfiguration.getRestServerPort(),
- savepointTriggerHeaders,
- savepointTriggerMessageParameters,
- new SavepointTriggerRequestBody(savepointDirectory));
- } catch (IOException e) {
- throw new FlinkException("Could not send trigger savepoint request to Flink cluster.", e);
- }
+ responseFuture = sendRequest(
+ savepointTriggerHeaders,
+ savepointTriggerMessageParameters,
+ new SavepointTriggerRequestBody(savepointDirectory));
- return responseFuture.thenApply(savepointTriggerResponseBody -> {
+ return responseFuture.thenCompose(savepointTriggerResponseBody -> {
final SavepointTriggerId savepointTriggerId = savepointTriggerResponseBody.getSavepointTriggerId();
- final SavepointInfo savepointInfo;
- try {
- savepointInfo = waitForSavepointCompletion(jobId, savepointTriggerId);
- } catch (Exception e) {
- throw new CompletionException(e);
- }
+ return waitForSavepointCompletion(jobId, savepointTriggerId);
+ }).thenApply(savepointInfo -> {
if (savepointInfo.getFailureCause() != null) {
throw new CompletionException(savepointInfo.getFailureCause());
}
return savepointInfo.getLocation();
});
}
- private SavepointInfo waitForSavepointCompletion(
+ private CompletableFuture<SavepointInfo> waitForSavepointCompletion(
final JobID jobId,
- final SavepointTriggerId savepointTriggerId) throws Exception {
+ final SavepointTriggerId savepointTriggerId) {
return waitForResource(() -> {
final SavepointStatusHeaders savepointStatusHeaders = SavepointStatusHeaders.getInstance();
final SavepointStatusMessageParameters savepointStatusMessageParameters =
savepointStatusHeaders.getUnresolvedMessageParameters();
savepointStatusMessageParameters.jobIdPathParameter.resolve(jobId);
savepointStatusMessageParameters.savepointTriggerIdPathParameter.resolve(savepointTriggerId);
- return restClient.sendRequest(
- restClusterClientConfiguration.getRestServerAddress(),
- restClusterClientConfiguration.getRestServerPort(),
+ return sendRetryableRequest(
savepointStatusHeaders,
- savepointStatusMessageParameters
- );
+ savepointStatusMessageParameters,
+ EmptyRequestBody.getInstance(),
+ isConnectionProblemException());
});
}
@Override
public CompletableFuture<Collection<JobStatusMessage>> listJobs() throws Exception {
- JobsOverviewHeaders headers = JobsOverviewHeaders.getInstance();
- CompletableFuture<MultipleJobsDetails> jobDetailsFuture = restClient.sendRequest(
- restClusterClientConfiguration.getRestServerAddress(),
- restClusterClientConfiguration.getRestServerPort(),
- headers
- );
- return jobDetailsFuture
+ return sendRequest(JobsOverviewHeaders.getInstance())
.thenApply(
- (MultipleJobsDetails multipleJobsDetails) -> {
- final Collection<JobDetails> jobDetails = multipleJobsDetails.getJobs();
- Collection<JobStatusMessage> flattenedDetails = new ArrayList<>(jobDetails.size());
- jobDetails.forEach(detail -> flattenedDetails.add(new JobStatusMessage(detail.getJobId(), detail.getJobName(), detail.getStatus(), detail.getStartTime())));
-
- return flattenedDetails;
- });
+ (multipleJobsDetails) -> multipleJobsDetails
+ .getJobs()
+ .stream()
+ .map(detail -> new JobStatusMessage(
+ detail.getJobId(),
+ detail.getJobName(),
+ detail.getStatus(),
+ detail.getStartTime()))
+ .collect(Collectors.toList()));
}
@Override
public T getClusterId() {
return clusterId;
}
- private <R, A extends AsynchronouslyCreatedResource<R>> R waitForResource(
- final SupplierWithException<CompletableFuture<A>, IOException> resourceFutureSupplier)
- throws IOException, InterruptedException, ExecutionException, TimeoutException {
- A asynchronouslyCreatedResource;
- long attempt = 0;
- while (true) {
- final CompletableFuture<A> responseFuture = resourceFutureSupplier.get();
- asynchronouslyCreatedResource = responseFuture.get(timeout.toMillis(), TimeUnit.MILLISECONDS);
- if (asynchronouslyCreatedResource.queueStatus().getId() == QueueStatus.Id.COMPLETED) {
- break;
+ /**
+ * Polls a {@code AsynchronouslyCreatedResource} until its
+ * {@link AsynchronouslyCreatedResource#queueStatus() QueueStatus} becomes
+ * {@link QueueStatus.Id#COMPLETED COMPLETED}. This method returns a {@code CompletableFuture}
+ * which completes with {@link AsynchronouslyCreatedResource#resource()}.
+ *
+ * @param resourceFutureSupplier The operation which polls for the
+ * {@code AsynchronouslyCreatedResource}.
+ * @param <R> The type of the resource.
+ * @param <A> The type of the {@code AsynchronouslyCreatedResource}.
+ * @return A {@code CompletableFuture} delivering the resource.
+ */
+ private <R, A extends AsynchronouslyCreatedResource<R>> CompletableFuture<R> waitForResource(
--- End diff --
It's now `pollResourceAsync`. This is consistent with the methods in `CompletableFuture`.
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r162696429
--- 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();
+ return restClient.sendRequest(restServerBaseUrl.getHost(), restServerBaseUrl.getPort(), messageHeaders, messageParameters, request);
+ }, retryPredicate);
+ }
+
+ private <C> CompletableFuture<C> retry(
+ CheckedSupplier<CompletableFuture<C>> operation,
+ Predicate<Throwable> retryPredicate) {
+ return FutureUtils.retryWithDelay(
+ CheckedSupplier.unchecked(operation),
+ MAX_RETRIES,
+ RETRY_DELAY,
+ retryPredicate,
+ new ScheduledExecutorServiceAdapter(retryExecutorService));
+ }
+
+ private static Predicate<Throwable> isTimeoutException() {
+ return (throwable) ->
+ ExceptionUtils.findThrowable(throwable, java.net.ConnectException.class).isPresent() ||
+ ExceptionUtils.findThrowable(throwable, java.net.SocketTimeoutException.class).isPresent() ||
+ ExceptionUtils.findThrowable(throwable, ConnectTimeoutException.class).isPresent() ||
+ ExceptionUtils.findThrowable(throwable, IOException.class).isPresent();
--- End diff --
Is an `IOException` really indicating a timeout issue?
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r163160703
--- Diff: flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClientConfiguration.java ---
@@ -30,49 +28,20 @@
*/
public final class RestClusterClientConfiguration {
--- End diff --
Now it's needed again.
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r162696516
--- Diff: flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java ---
@@ -61,46 +69,77 @@
import org.apache.flink.runtime.rest.messages.job.savepoints.SavepointTriggerResponseBody;
import org.apache.flink.runtime.rest.messages.queue.AsynchronouslyCreatedResource;
import org.apache.flink.runtime.rest.messages.queue.QueueStatus;
+import org.apache.flink.runtime.rest.util.RestClientException;
import org.apache.flink.runtime.util.ExecutorThreadFactory;
+import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.ExecutorUtils;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.SerializedThrowable;
-import org.apache.flink.util.function.SupplierWithException;
+import org.apache.flink.util.function.CheckedSupplier;
+
+import org.apache.flink.shaded.netty4.io.netty.channel.ConnectTimeoutException;
+
+import akka.actor.AddressFromURIString;
import javax.annotation.Nullable;
import java.io.IOException;
import java.net.InetSocketAddress;
+import java.net.MalformedURLException;
import java.net.URL;
-import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
-import static java.util.Objects.requireNonNull;
+import scala.Option;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
/**
* A {@link ClusterClient} implementation that communicates via HTTP REST requests.
*/
public class RestClusterClient<T> extends ClusterClient<T> {
+ private static final long AWAIT_LEADER_TIMEOUT = 10_000;
+
+ private static final int MAX_RETRIES = 20;
+
+ private static final Time RETRY_DELAY = Time.seconds(3);
--- End diff --
These values should be made configurable.
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r162692567
--- Diff: flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java ---
@@ -61,46 +69,77 @@
import org.apache.flink.runtime.rest.messages.job.savepoints.SavepointTriggerResponseBody;
import org.apache.flink.runtime.rest.messages.queue.AsynchronouslyCreatedResource;
import org.apache.flink.runtime.rest.messages.queue.QueueStatus;
+import org.apache.flink.runtime.rest.util.RestClientException;
import org.apache.flink.runtime.util.ExecutorThreadFactory;
+import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.ExecutorUtils;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.SerializedThrowable;
-import org.apache.flink.util.function.SupplierWithException;
+import org.apache.flink.util.function.CheckedSupplier;
+
+import org.apache.flink.shaded.netty4.io.netty.channel.ConnectTimeoutException;
+
+import akka.actor.AddressFromURIString;
import javax.annotation.Nullable;
import java.io.IOException;
import java.net.InetSocketAddress;
+import java.net.MalformedURLException;
import java.net.URL;
-import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
-import static java.util.Objects.requireNonNull;
+import scala.Option;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
/**
* A {@link ClusterClient} implementation that communicates via HTTP REST requests.
*/
public class RestClusterClient<T> extends ClusterClient<T> {
+ private static final long AWAIT_LEADER_TIMEOUT = 10_000;
+
+ private static final int MAX_RETRIES = 20;
+
+ private static final Time RETRY_DELAY = Time.seconds(3);
+
private final RestClusterClientConfiguration restClusterClientConfiguration;
private final RestClient restClient;
private final ExecutorService executorService = Executors.newFixedThreadPool(4, new ExecutorThreadFactory("Flink-RestClusterClient-IO"));
+
private final WaitStrategy waitStrategy;
private final T clusterId;
+ private LeaderRetrievalService restServerRetrievalService;
+
+ private LeaderRetrievalService dispatcherLeaderRetriever;
+
+ private final LeaderHolder<URL> restServerLeaderHolder = new LeaderHolder<>("RestServer", AWAIT_LEADER_TIMEOUT);
+
+ private final LeaderHolder<String> dispatcherLeaderHolder = new LeaderHolder<>("Dispatcher", AWAIT_LEADER_TIMEOUT);
--- End diff --
I think we could use a `LeaderRetriever` implementation instead of the `LeaderHolder` here.
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r163160791
--- 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();
+ return restClient.sendRequest(restServerBaseUrl.getHost(), restServerBaseUrl.getPort(), messageHeaders, messageParameters, request);
+ }, retryPredicate);
+ }
+
+ private <C> CompletableFuture<C> retry(
+ CheckedSupplier<CompletableFuture<C>> operation,
+ Predicate<Throwable> retryPredicate) {
+ return FutureUtils.retryWithDelay(
+ CheckedSupplier.unchecked(operation),
+ MAX_RETRIES,
+ RETRY_DELAY,
+ retryPredicate,
+ new ScheduledExecutorServiceAdapter(retryExecutorService));
+ }
+
+ private static Predicate<Throwable> isTimeoutException() {
+ return (throwable) ->
+ ExceptionUtils.findThrowable(throwable, java.net.ConnectException.class).isPresent() ||
+ ExceptionUtils.findThrowable(throwable, java.net.SocketTimeoutException.class).isPresent() ||
+ ExceptionUtils.findThrowable(throwable, ConnectTimeoutException.class).isPresent() ||
+ ExceptionUtils.findThrowable(throwable, IOException.class).isPresent();
--- End diff --
renamed to `isConnectionProblemException`
---
[GitHub] flink pull request #5312: [FLINK-8344][flip6] Add support for HA to RestClus...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r163778106
--- Diff: flink-core/src/main/java/org/apache/flink/configuration/RestOptions.java ---
@@ -41,4 +41,34 @@
public static final ConfigOption<Integer> REST_PORT =
key("rest.port")
.defaultValue(9067);
+
+ /**
+ * The time in ms to wait for the leader address, e.g., Dispatcher or WebMonitorEndpoint.
+ */
+ public static final ConfigOption<Integer> AWAIT_LEADER_TIMEOUT =
+ key("rest.await-leader-timeout")
+ .defaultValue(30_000);
+
+ /**
+ * The number of attempts to retry a retryable operation.
+ * @see #RETRY_DELAY
+ */
+ public static final ConfigOption<Integer> RETRY_MAX_ATTEMPTS =
+ key("rest.retry.max-attempts")
+ .defaultValue(20);
+
+ /**
+ * The time in ms to wait between retries.
+ * @see #RETRY_MAX_ATTEMPTS
+ */
+ public static final ConfigOption<Integer> RETRY_DELAY =
+ key("rest.retry.delay")
+ .defaultValue(3_000);
+
+ /**
+ * The maximum time in ms to establish a TCP connection.
+ */
+ public static final ConfigOption<Integer> CONNECTION_TIMEOUT =
+ key("rest.connection-timeout")
+ .defaultValue(15_000);
--- End diff --
please add a description via `withDescription` to these options.
---
[GitHub] flink pull request #5312: [FLINK-8344][flip6] Add support for HA to RestClus...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r163776913
--- Diff: flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClientConfiguration.java ---
@@ -19,60 +19,73 @@
package org.apache.flink.client.program.rest;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.runtime.rest.RestClientConfiguration;
import org.apache.flink.util.ConfigurationException;
import org.apache.flink.util.Preconditions;
+import static org.apache.flink.util.Preconditions.checkArgument;
+
/**
* A configuration object for {@link RestClusterClient}s.
*/
public final class RestClusterClientConfiguration {
- private final String blobServerAddress;
-
private final RestClientConfiguration restClientConfiguration;
- private final String restServerAddress;
+ private final int awaitLeaderTimeout;
+
+ private final int retryMaxAttempts;
- private final int restServerPort;
+ private final int retryDelay;
private RestClusterClientConfiguration(
- String blobServerAddress,
- RestClientConfiguration endpointConfiguration,
- String restServerAddress,
- int restServerPort) {
- this.blobServerAddress = Preconditions.checkNotNull(blobServerAddress);
+ final RestClientConfiguration endpointConfiguration,
+ final int awaitLeaderTimeout,
+ final int retryMaxAttempts,
+ final int retryDelay) {
--- End diff --
Same here.
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r162412277
--- Diff: flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java ---
@@ -596,11 +596,11 @@ public int run(String[] args) throws CliArgsException, FlinkException {
//------------------ ClusterClient deployed, handle connection details
yarnApplicationId = clusterClient.getClusterId();
- String jobManagerAddress =
- clusterClient.getJobManagerAddress().getAddress().getHostName() +
- ':' + clusterClient.getJobManagerAddress().getPort();
+// String jobManagerAddress =
+// clusterClient.getJobManagerAddress().getAddress().getHostName() +
--- End diff --
commented temporarily because it does not work in flip6
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r162714628
--- 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();
+ return restClient.sendRequest(restServerBaseUrl.getHost(), restServerBaseUrl.getPort(), messageHeaders, messageParameters, request);
+ }, retryPredicate);
+ }
+
+ private <C> CompletableFuture<C> retry(
+ CheckedSupplier<CompletableFuture<C>> operation,
+ Predicate<Throwable> retryPredicate) {
+ return FutureUtils.retryWithDelay(
+ CheckedSupplier.unchecked(operation),
+ MAX_RETRIES,
+ RETRY_DELAY,
+ retryPredicate,
+ new ScheduledExecutorServiceAdapter(retryExecutorService));
+ }
+
+ private static Predicate<Throwable> isTimeoutException() {
+ return (throwable) ->
+ ExceptionUtils.findThrowable(throwable, java.net.ConnectException.class).isPresent() ||
+ ExceptionUtils.findThrowable(throwable, java.net.SocketTimeoutException.class).isPresent() ||
+ ExceptionUtils.findThrowable(throwable, ConnectTimeoutException.class).isPresent() ||
+ ExceptionUtils.findThrowable(throwable, IOException.class).isPresent();
--- End diff --
The method needs to be renamed but `IOException` can be thrown if the connection is reset. I have seen it happen while experimenting.
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r163161003
--- Diff: flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java ---
@@ -327,13 +376,14 @@ public T getClusterId() {
}
private <R, A extends AsynchronouslyCreatedResource<R>> R waitForResource(
- final SupplierWithException<CompletableFuture<A>, IOException> resourceFutureSupplier)
+ final Supplier<CompletableFuture<A>> resourceFutureSupplier)
throws IOException, InterruptedException, ExecutionException, TimeoutException {
A asynchronouslyCreatedResource;
long attempt = 0;
while (true) {
final CompletableFuture<A> responseFuture = resourceFutureSupplier.get();
--- End diff --
Made the method `waitForResource ` non-blocking.
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by GJL <gi...@git.apache.org>.
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r162714906
--- Diff: flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java ---
@@ -61,46 +69,77 @@
import org.apache.flink.runtime.rest.messages.job.savepoints.SavepointTriggerResponseBody;
import org.apache.flink.runtime.rest.messages.queue.AsynchronouslyCreatedResource;
import org.apache.flink.runtime.rest.messages.queue.QueueStatus;
+import org.apache.flink.runtime.rest.util.RestClientException;
import org.apache.flink.runtime.util.ExecutorThreadFactory;
+import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.ExecutorUtils;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.SerializedThrowable;
-import org.apache.flink.util.function.SupplierWithException;
+import org.apache.flink.util.function.CheckedSupplier;
+
+import org.apache.flink.shaded.netty4.io.netty.channel.ConnectTimeoutException;
+
+import akka.actor.AddressFromURIString;
import javax.annotation.Nullable;
import java.io.IOException;
import java.net.InetSocketAddress;
+import java.net.MalformedURLException;
import java.net.URL;
-import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
-import static java.util.Objects.requireNonNull;
+import scala.Option;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
/**
* A {@link ClusterClient} implementation that communicates via HTTP REST requests.
*/
public class RestClusterClient<T> extends ClusterClient<T> {
+ private static final long AWAIT_LEADER_TIMEOUT = 10_000;
+
+ private static final int MAX_RETRIES = 20;
+
+ private static final Time RETRY_DELAY = Time.seconds(3);
+
private final RestClusterClientConfiguration restClusterClientConfiguration;
private final RestClient restClient;
private final ExecutorService executorService = Executors.newFixedThreadPool(4, new ExecutorThreadFactory("Flink-RestClusterClient-IO"));
+
private final WaitStrategy waitStrategy;
private final T clusterId;
+ private LeaderRetrievalService restServerRetrievalService;
+
+ private LeaderRetrievalService dispatcherLeaderRetriever;
+
+ private final LeaderHolder<URL> restServerLeaderHolder = new LeaderHolder<>("RestServer", AWAIT_LEADER_TIMEOUT);
+
+ private final LeaderHolder<String> dispatcherLeaderHolder = new LeaderHolder<>("Dispatcher", AWAIT_LEADER_TIMEOUT);
--- End diff --
Ok, I will look into it. Somehow I missed this class.
---
[GitHub] flink pull request #5312: [FLINK-8344][WIP] Add support for HA to RestCluste...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/5312#discussion_r162690747
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java ---
@@ -144,6 +154,15 @@ public WebMonitorEndpoint(
metricQueryServiceRetriever,
executor,
restConfiguration.getTimeout());
+
+ this.leaderElectionService = Preconditions.checkNotNull(leaderElectionService);
+ this.fatalErrorHandler = Preconditions.checkNotNull(fatalErrorHandler);
+ }
+
+ @Override
+ public void start() throws Exception {
+ super.start();
+ leaderElectionService.start(this);
--- End diff --
We should also stop the `leaderElectionService`.
---