You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2018/01/11 23:19:57 UTC
[8/8] flink git commit: [FLINK-8299][flip6] Retrieve
JobExecutionResult after job submission
[FLINK-8299][flip6] Retrieve JobExecutionResult after job submission
[FLINK-8299][flip6] Improve ExponentialWaitStrategy
Add additional argument validation. Add more unit tests.
This closes #5207.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/06922753
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/06922753
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/06922753
Branch: refs/heads/master
Commit: 06922753a55dc322b96919ebb407d531e2b79d3e
Parents: e2f1ba9
Author: gyao <ga...@data-artisans.com>
Authored: Sat Dec 23 00:02:10 2017 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Thu Jan 11 20:55:17 2018 +0100
----------------------------------------------------------------------
.../flink/client/program/ClusterClient.java | 2 +-
.../client/program/rest/RestClusterClient.java | 86 +++++-
.../rest/retry/ExponentialWaitStrategy.java | 46 ++++
.../client/program/rest/retry/WaitStrategy.java | 35 +++
.../program/rest/RestClusterClientTest.java | 273 ++++++++++++-------
.../rest/retry/ExponentialWaitStrategyTest.java | 89 ++++++
.../rest/messages/JobMessageParameters.java | 2 +-
7 files changed, 424 insertions(+), 109 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/06922753/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java b/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java
index 5f0c25f..7c05e3e 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java
@@ -124,7 +124,7 @@ public abstract class ClusterClient {
* been run inside the user JAR. We pass the Client to every instance of the ContextEnvironment
* which lets us access the execution result here.
*/
- private JobExecutionResult lastJobExecutionResult;
+ protected JobExecutionResult lastJobExecutionResult;
/** Switch for blocking/detached job submission of the client. */
private boolean detachedJobSubmission = false;
http://git-wip-us.apache.org/repos/asf/flink/blob/06922753/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java b/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
index a7d9d02..6b21cff 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
@@ -18,12 +18,16 @@
package org.apache.flink.client.program.rest;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobSubmissionResult;
+import org.apache.flink.api.common.accumulators.AccumulatorHelper;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.client.program.ProgramInvocationException;
+import org.apache.flink.client.program.rest.retry.ExponentialWaitStrategy;
+import org.apache.flink.client.program.rest.retry.WaitStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.blob.BlobClient;
import org.apache.flink.runtime.blob.PermanentBlobKey;
@@ -31,25 +35,31 @@ import org.apache.flink.runtime.client.JobStatusMessage;
import org.apache.flink.runtime.client.JobSubmissionException;
import org.apache.flink.runtime.clusterframework.messages.GetClusterStatusResponse;
import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobmaster.JobResult;
import org.apache.flink.runtime.messages.webmonitor.JobDetails;
import org.apache.flink.runtime.messages.webmonitor.MultipleJobsDetails;
import org.apache.flink.runtime.rest.RestClient;
import org.apache.flink.runtime.rest.messages.BlobServerPortHeaders;
import org.apache.flink.runtime.rest.messages.BlobServerPortResponseBody;
import org.apache.flink.runtime.rest.messages.EmptyResponseBody;
+import org.apache.flink.runtime.rest.messages.JobMessageParameters;
import org.apache.flink.runtime.rest.messages.JobTerminationHeaders;
import org.apache.flink.runtime.rest.messages.JobTerminationMessageParameters;
import org.apache.flink.runtime.rest.messages.JobsOverviewHeaders;
import org.apache.flink.runtime.rest.messages.TerminationModeQueryParameter;
+import org.apache.flink.runtime.rest.messages.job.JobExecutionResultHeaders;
+import org.apache.flink.runtime.rest.messages.job.JobExecutionResultResponseBody;
import org.apache.flink.runtime.rest.messages.job.JobSubmitHeaders;
import org.apache.flink.runtime.rest.messages.job.JobSubmitRequestBody;
import org.apache.flink.runtime.rest.messages.job.JobSubmitResponseBody;
import org.apache.flink.runtime.rest.messages.job.savepoints.SavepointMessageParameters;
import org.apache.flink.runtime.rest.messages.job.savepoints.SavepointTriggerHeaders;
import org.apache.flink.runtime.rest.messages.job.savepoints.SavepointTriggerResponseBody;
+import org.apache.flink.runtime.rest.messages.queue.QueueStatus;
import org.apache.flink.runtime.util.ExecutorThreadFactory;
import org.apache.flink.util.ExecutorUtils;
import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.SerializedThrowable;
import javax.annotation.Nullable;
@@ -61,9 +71,14 @@ import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import static java.util.Objects.requireNonNull;
+import static org.apache.flink.util.Preconditions.checkState;
/**
* A {@link ClusterClient} implementation that communicates via HTTP REST requests.
@@ -73,15 +88,18 @@ public class RestClusterClient extends ClusterClient {
private final RestClusterClientConfiguration restClusterClientConfiguration;
private final RestClient restClient;
private final ExecutorService executorService = Executors.newFixedThreadPool(4, new ExecutorThreadFactory("Flink-RestClusterClient-IO"));
+ private final WaitStrategy waitStrategy;
public RestClusterClient(Configuration config) throws Exception {
- this(config, RestClusterClientConfiguration.fromConfiguration(config));
+ this(config, new ExponentialWaitStrategy(10, 2000));
}
- public RestClusterClient(Configuration config, RestClusterClientConfiguration configuration) throws Exception {
- super(config);
- this.restClusterClientConfiguration = configuration;
- this.restClient = new RestClient(configuration.getRestClientConfiguration(), executorService);
+ @VisibleForTesting
+ RestClusterClient(Configuration configuration, WaitStrategy waitStrategy) throws Exception {
+ super(configuration);
+ this.restClusterClientConfiguration = RestClusterClientConfiguration.fromConfiguration(configuration);
+ this.restClient = new RestClient(restClusterClientConfiguration.getRestClientConfiguration(), executorService);
+ this.waitStrategy = requireNonNull(waitStrategy);
}
@Override
@@ -106,11 +124,28 @@ public class RestClusterClient extends ClusterClient {
} catch (JobSubmissionException e) {
throw new ProgramInvocationException(e);
}
- // don't return just a JobSubmissionResult here, the signature is lying
- // The CliFrontend expects this to be a JobExecutionResult
- // TOOD: do not exit this method until job is finished
- return new JobExecutionResult(jobGraph.getJobID(), 1, Collections.emptyMap());
+ final JobResult jobExecutionResult = waitForJobExecutionResult(jobGraph.getJobID());
+
+ if (jobExecutionResult.getSerializedThrowable().isPresent()) {
+ final SerializedThrowable serializedThrowable = jobExecutionResult.getSerializedThrowable().get();
+ final Throwable throwable = serializedThrowable.deserializeError(classLoader);
+ throw new ProgramInvocationException(throwable);
+ }
+
+ try {
+ // don't return just a JobSubmissionResult here, the signature is lying
+ // The CliFrontend expects this to be a JobExecutionResult
+ this.lastJobExecutionResult = new JobExecutionResult(
+ jobExecutionResult.getJobId(),
+ jobExecutionResult.getNetRuntime(),
+ AccumulatorHelper.deserializeAccumulators(
+ jobExecutionResult.getAccumulatorResults(),
+ classLoader));
+ return lastJobExecutionResult;
+ } catch (IOException | ClassNotFoundException e) {
+ throw new ProgramInvocationException(e);
+ }
}
private void submitJob(JobGraph jobGraph) throws JobSubmissionException {
@@ -150,6 +185,39 @@ public class RestClusterClient extends ClusterClient {
}
}
+ private JobResult waitForJobExecutionResult(
+ final JobID jobId) throws ProgramInvocationException {
+
+ final JobMessageParameters messageParameters = new JobMessageParameters();
+ messageParameters.jobPathParameter.resolve(jobId);
+ JobExecutionResultResponseBody jobExecutionResultResponseBody;
+ try {
+ long attempt = 0;
+ do {
+ final CompletableFuture<JobExecutionResultResponseBody> responseFuture =
+ restClient.sendRequest(
+ restClusterClientConfiguration.getRestServerAddress(),
+ restClusterClientConfiguration.getRestServerPort(),
+ JobExecutionResultHeaders.getInstance(),
+ messageParameters);
+ jobExecutionResultResponseBody = responseFuture.get(timeout.toMillis(), TimeUnit.MILLISECONDS);
+ Thread.sleep(waitStrategy.sleepTime(attempt));
+ attempt++;
+ }
+ while (jobExecutionResultResponseBody.getStatus().getStatusId() != QueueStatus.StatusId.COMPLETED);
+ } catch (IOException | TimeoutException | ExecutionException e) {
+ throw new ProgramInvocationException(e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new ProgramInvocationException(e);
+ }
+
+ final JobResult jobExecutionResult = jobExecutionResultResponseBody.getJobExecutionResult();
+ checkState(jobExecutionResult != null, "jobExecutionResult must not be null");
+
+ return jobExecutionResult;
+ }
+
@Override
public void stop(JobID jobID) throws Exception {
JobTerminationMessageParameters params = new JobTerminationMessageParameters();
http://git-wip-us.apache.org/repos/asf/flink/blob/06922753/flink-clients/src/main/java/org/apache/flink/client/program/rest/retry/ExponentialWaitStrategy.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/rest/retry/ExponentialWaitStrategy.java b/flink-clients/src/main/java/org/apache/flink/client/program/rest/retry/ExponentialWaitStrategy.java
new file mode 100644
index 0000000..2bb5051
--- /dev/null
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/rest/retry/ExponentialWaitStrategy.java
@@ -0,0 +1,46 @@
+/*
+ * 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.retry;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/**
+ * {@link WaitStrategy} with exponentially increasing sleep time.
+ */
+public class ExponentialWaitStrategy implements WaitStrategy {
+
+ private final long initialWait;
+
+ private final long maxWait;
+
+ public ExponentialWaitStrategy(final long initialWait, final long maxWait) {
+ checkArgument(initialWait > 0, "initialWait must be positive, was %s", initialWait);
+ checkArgument(maxWait > 0, "maxWait must be positive, was %s", maxWait);
+ checkArgument(initialWait <= maxWait, "initialWait must be lower than or equal to maxWait", maxWait);
+ this.initialWait = initialWait;
+ this.maxWait = maxWait;
+ }
+
+ @Override
+ public long sleepTime(final long attempt) {
+ checkArgument(attempt >= 0, "attempt must not be negative (%d)", attempt);
+ final long exponentialSleepTime = initialWait * Math.round(Math.pow(2, attempt));
+ return exponentialSleepTime >= 0 && exponentialSleepTime < maxWait ? exponentialSleepTime : maxWait;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/06922753/flink-clients/src/main/java/org/apache/flink/client/program/rest/retry/WaitStrategy.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/rest/retry/WaitStrategy.java b/flink-clients/src/main/java/org/apache/flink/client/program/rest/retry/WaitStrategy.java
new file mode 100644
index 0000000..c4dccce
--- /dev/null
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/rest/retry/WaitStrategy.java
@@ -0,0 +1,35 @@
+/*
+ * 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.retry;
+
+/**
+ * Operations that are polling for a result to arrive require a waiting time between consecutive
+ * polls. A {@code WaitStrategy} determines this waiting time.
+ */
+@FunctionalInterface
+public interface WaitStrategy {
+
+ /**
+ * Returns the time to wait until the next attempt. Attempts start at {@code 0}.
+ * @param attempt The number of the last attempt.
+ * @return Waiting time in ms.
+ */
+ long sleepTime(long attempt);
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/06922753/flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientTest.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientTest.java b/flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientTest.java
index b36b135..b17b030 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientTest.java
@@ -21,6 +21,7 @@ package org.apache.flink.client.program.rest;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.client.program.ProgramInvocationException;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.runtime.client.JobStatusMessage;
@@ -28,6 +29,7 @@ import org.apache.flink.runtime.dispatcher.Dispatcher;
import org.apache.flink.runtime.dispatcher.DispatcherGateway;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobStatus;
+import org.apache.flink.runtime.jobmaster.JobResult;
import org.apache.flink.runtime.messages.webmonitor.JobDetails;
import org.apache.flink.runtime.messages.webmonitor.MultipleJobsDetails;
import org.apache.flink.runtime.rest.RestServerEndpoint;
@@ -41,6 +43,7 @@ import org.apache.flink.runtime.rest.messages.BlobServerPortResponseBody;
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.JobMessageParameters;
import org.apache.flink.runtime.rest.messages.JobTerminationHeaders;
import org.apache.flink.runtime.rest.messages.JobTerminationMessageParameters;
import org.apache.flink.runtime.rest.messages.JobsOverviewHeaders;
@@ -49,6 +52,8 @@ import org.apache.flink.runtime.rest.messages.MessageParameters;
import org.apache.flink.runtime.rest.messages.RequestBody;
import org.apache.flink.runtime.rest.messages.ResponseBody;
import org.apache.flink.runtime.rest.messages.TerminationModeQueryParameter;
+import org.apache.flink.runtime.rest.messages.job.JobExecutionResultHeaders;
+import org.apache.flink.runtime.rest.messages.job.JobExecutionResultResponseBody;
import org.apache.flink.runtime.rest.messages.job.JobSubmitHeaders;
import org.apache.flink.runtime.rest.messages.job.JobSubmitRequestBody;
import org.apache.flink.runtime.rest.messages.job.JobSubmitResponseBody;
@@ -59,13 +64,19 @@ import org.apache.flink.runtime.rest.messages.job.savepoints.SavepointTriggerRes
import org.apache.flink.runtime.rpc.RpcUtils;
import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
import org.apache.flink.testutils.category.Flip6;
+import org.apache.flink.util.SerializedThrowable;
+import org.apache.flink.util.SerializedValue;
import org.apache.flink.util.TestLogger;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelInboundHandler;
+import org.junit.After;
import org.junit.Assert;
+import org.junit.Before;
import org.junit.Test;
import org.junit.experimental.categories.Category;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
import javax.annotation.Nonnull;
@@ -75,11 +86,14 @@ import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
-import java.util.Optional;
import java.util.concurrent.CompletableFuture;
+import static org.apache.flink.util.Preconditions.checkState;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
import static org.mockito.Matchers.any;
-import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
/**
@@ -91,67 +105,75 @@ import static org.mockito.Mockito.when;
@Category(Flip6.class)
public class RestClusterClientTest extends TestLogger {
- private static final String restAddress = "http://localhost:1234";
- private static final Dispatcher mockRestfulGateway = mock(Dispatcher.class);
- private static final GatewayRetriever<DispatcherGateway> mockGatewayRetriever = mock(GatewayRetriever.class);
+ private static final String REST_ADDRESS = "http://localhost:1234";
- static {
- when(mockRestfulGateway.requestRestAddress(any(Time.class))).thenReturn(CompletableFuture.completedFuture(restAddress));
- when(mockGatewayRetriever.getNow()).thenReturn(Optional.of(mockRestfulGateway));
- }
+ @Mock
+ private Dispatcher mockRestfulGateway;
- @Test
- public void testJobSubmitCancelStop() throws Exception {
+ @Mock
+ private GatewayRetriever<DispatcherGateway> mockGatewayRetriever;
- Configuration config = new Configuration();
- config.setString(JobManagerOptions.ADDRESS, "localhost");
+ private RestServerEndpointConfiguration restServerEndpointConfiguration;
- RestServerEndpointConfiguration rsec = RestServerEndpointConfiguration.fromConfiguration(config);
+ private RestClusterClient restClusterClient;
- TestBlobServerPortHandler portHandler = new TestBlobServerPortHandler();
- TestJobSubmitHandler submitHandler = new TestJobSubmitHandler();
- TestJobTerminationHandler terminationHandler = new TestJobTerminationHandler();
+ @Before
+ public void setUp() throws Exception {
+ MockitoAnnotations.initMocks(this);
+ when(mockRestfulGateway.requestRestAddress(any(Time.class))).thenReturn(CompletableFuture.completedFuture(REST_ADDRESS));
- RestServerEndpoint rse = new RestServerEndpoint(rsec) {
- @Override
- protected List<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> initializeHandlers(CompletableFuture<String> restAddressFuture) {
+ final Configuration config = new Configuration();
+ config.setString(JobManagerOptions.ADDRESS, "localhost");
+ restServerEndpointConfiguration = RestServerEndpointConfiguration.fromConfiguration(config);
+ mockGatewayRetriever = () -> CompletableFuture.completedFuture(mockRestfulGateway);
+ restClusterClient = new RestClusterClient(config, (attempt) -> 0);
+ }
- List<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> handlers = new ArrayList<>();
- handlers.add(Tuple2.of(portHandler.getMessageHeaders(), portHandler));
- handlers.add(Tuple2.of(submitHandler.getMessageHeaders(), submitHandler));
- handlers.add(Tuple2.of(terminationHandler.getMessageHeaders(), terminationHandler));
- return handlers;
- }
- };
+ @After
+ public void tearDown() throws Exception {
+ if (restClusterClient != null) {
+ restClusterClient.shutdown();
+ }
+ }
- RestClusterClient rcc = new RestClusterClient(config);
- try {
- rse.start();
+ @Test
+ public void testJobSubmitCancelStop() throws Exception {
+ final JobGraph job = new JobGraph("testjob");
+ final JobID id = job.getJobID();
- JobGraph job = new JobGraph("testjob");
- JobID id = job.getJobID();
+ TestBlobServerPortHandler portHandler = new TestBlobServerPortHandler();
+ TestJobSubmitHandler submitHandler = new TestJobSubmitHandler();
+ TestJobTerminationHandler terminationHandler = new TestJobTerminationHandler();
+ TestJobExecutionResultHandler testJobExecutionResultHandler =
+ new TestJobExecutionResultHandler(Collections.singletonList(
+ JobExecutionResultResponseBody.created(new JobResult.Builder()
+ .jobId(id)
+ .netRuntime(Long.MAX_VALUE)
+ .build())).iterator());
+
+ try (TestRestServerEndpoint ignored = createRestServerEndpoint(
+ portHandler,
+ submitHandler,
+ terminationHandler,
+ testJobExecutionResultHandler)) {
Assert.assertFalse(portHandler.portRetrieved);
Assert.assertFalse(submitHandler.jobSubmitted);
- rcc.submitJob(job, ClassLoader.getSystemClassLoader());
+ restClusterClient.submitJob(job, ClassLoader.getSystemClassLoader());
Assert.assertTrue(portHandler.portRetrieved);
Assert.assertTrue(submitHandler.jobSubmitted);
Assert.assertFalse(terminationHandler.jobCanceled);
- rcc.cancel(id);
+ restClusterClient.cancel(id);
Assert.assertTrue(terminationHandler.jobCanceled);
Assert.assertFalse(terminationHandler.jobStopped);
- rcc.stop(id);
+ restClusterClient.stop(id);
Assert.assertTrue(terminationHandler.jobStopped);
-
- } finally {
- rcc.shutdown();
- rse.shutdown(Time.seconds(5));
}
}
- private static class TestBlobServerPortHandler extends TestHandler<EmptyRequestBody, BlobServerPortResponseBody, EmptyMessageParameters> {
+ private class TestBlobServerPortHandler extends TestHandler<EmptyRequestBody, BlobServerPortResponseBody, EmptyMessageParameters> {
private volatile boolean portRetrieved = false;
private TestBlobServerPortHandler() {
@@ -165,7 +187,7 @@ public class RestClusterClientTest extends TestLogger {
}
}
- private static class TestJobSubmitHandler extends TestHandler<JobSubmitRequestBody, JobSubmitResponseBody, EmptyMessageParameters> {
+ private class TestJobSubmitHandler extends TestHandler<JobSubmitRequestBody, JobSubmitResponseBody, EmptyMessageParameters> {
private volatile boolean jobSubmitted = false;
private TestJobSubmitHandler() {
@@ -179,7 +201,7 @@ public class RestClusterClientTest extends TestLogger {
}
}
- private static class TestJobTerminationHandler extends TestHandler<EmptyRequestBody, EmptyResponseBody, JobTerminationMessageParameters> {
+ private class TestJobTerminationHandler extends TestHandler<EmptyRequestBody, EmptyResponseBody, JobTerminationMessageParameters> {
private volatile boolean jobCanceled = false;
private volatile boolean jobStopped = false;
@@ -201,52 +223,96 @@ public class RestClusterClientTest extends TestLogger {
}
}
- @Test
- public void testTriggerSavepoint() throws Exception {
-
- Configuration config = new Configuration();
- config.setString(JobManagerOptions.ADDRESS, "localhost");
-
- RestServerEndpointConfiguration rsec = RestServerEndpointConfiguration.fromConfiguration(config);
+ private class TestJobExecutionResultHandler
+ extends TestHandler<EmptyRequestBody, JobExecutionResultResponseBody, JobMessageParameters> {
- String targetSavepointDirectory = "/alternate";
+ private final Iterator<JobExecutionResultResponseBody> jobExecutionResults;
- TestSavepointTriggerHandler triggerHandler = new TestSavepointTriggerHandler(targetSavepointDirectory);
+ private JobExecutionResultResponseBody lastJobExecutionResult;
- RestServerEndpoint rse = new RestServerEndpoint(rsec) {
- @Override
- protected List<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> initializeHandlers(CompletableFuture<String> restAddressFuture) {
+ private TestJobExecutionResultHandler(
+ final Iterator<JobExecutionResultResponseBody> jobExecutionResults) {
+ super(JobExecutionResultHeaders.getInstance());
+ this.jobExecutionResults = jobExecutionResults;
+ }
- List<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> handlers = new ArrayList<>();
- handlers.add(Tuple2.of(triggerHandler.getMessageHeaders(), triggerHandler));
- return handlers;
+ @Override
+ protected CompletableFuture<JobExecutionResultResponseBody> handleRequest(
+ @Nonnull HandlerRequest<EmptyRequestBody, JobMessageParameters> request,
+ @Nonnull DispatcherGateway gateway) throws RestHandlerException {
+ if (jobExecutionResults.hasNext()) {
+ lastJobExecutionResult = jobExecutionResults.next();
}
- };
+ checkState(lastJobExecutionResult != null);
+ return CompletableFuture.completedFuture(lastJobExecutionResult);
+ }
+ }
- RestClusterClient rcc = new RestClusterClient(config);
- try {
- rse.start();
+ @Test
+ public void testSubmitJobAndWaitForExecutionResult() throws Exception {
+ final JobGraph jobGraph = new JobGraph("testjob");
+ final JobID jobId = jobGraph.getJobID();
+
+ final TestJobExecutionResultHandler testJobExecutionResultHandler =
+ new TestJobExecutionResultHandler(Arrays.asList(
+ JobExecutionResultResponseBody.inProgress(),
+ JobExecutionResultResponseBody.created(new JobResult.Builder()
+ .jobId(jobId)
+ .netRuntime(Long.MAX_VALUE)
+ .accumulatorResults(Collections.singletonMap("testName", new SerializedValue<>(1.0)))
+ .build()),
+ JobExecutionResultResponseBody.created(new JobResult.Builder()
+ .jobId(jobId)
+ .netRuntime(Long.MAX_VALUE)
+ .serializedThrowable(new SerializedThrowable(new RuntimeException("expected")))
+ .build())).iterator());
+
+ try (TestRestServerEndpoint ignored = createRestServerEndpoint(
+ testJobExecutionResultHandler,
+ new TestBlobServerPortHandler(),
+ new TestJobSubmitHandler())) {
+
+ final org.apache.flink.api.common.JobExecutionResult jobExecutionResult =
+ (org.apache.flink.api.common.JobExecutionResult) restClusterClient.submitJob(
+ jobGraph,
+ ClassLoader.getSystemClassLoader());
+ assertThat(jobExecutionResult.getJobID(), equalTo(jobId));
+ assertThat(jobExecutionResult.getNetRuntime(), equalTo(Long.MAX_VALUE));
+ assertThat(
+ jobExecutionResult.getAllAccumulatorResults(),
+ equalTo(Collections.singletonMap("testName", 1.0)));
+
+ try {
+ restClusterClient.submitJob(jobGraph, ClassLoader.getSystemClassLoader());
+ fail("Expected exception not thrown.");
+ } catch (final ProgramInvocationException e) {
+ assertThat(e.getCause(), instanceOf(RuntimeException.class));
+ assertThat(e.getCause().getMessage(), equalTo("expected"));
+ }
+ }
+ }
+ @Test
+ public void testTriggerSavepoint() throws Exception {
+ String targetSavepointDirectory = "/alternate";
+ TestSavepointTriggerHandler triggerHandler = new TestSavepointTriggerHandler(targetSavepointDirectory);
+ try (TestRestServerEndpoint ignored = createRestServerEndpoint(triggerHandler)) {
JobID id = new JobID();
-
{
- CompletableFuture<String> savepointPathFuture = rcc.triggerSavepoint(id, null);
+ CompletableFuture<String> savepointPathFuture = restClusterClient.triggerSavepoint(id, null);
String savepointPath = savepointPathFuture.get();
Assert.assertEquals("/universe", savepointPath);
}
{
- CompletableFuture<String> savepointPathFuture = rcc.triggerSavepoint(id, targetSavepointDirectory);
+ CompletableFuture<String> savepointPathFuture = restClusterClient.triggerSavepoint(id, targetSavepointDirectory);
String savepointPath = savepointPathFuture.get();
Assert.assertEquals(targetSavepointDirectory + "/universe", savepointPath);
}
- } finally {
- rcc.shutdown();
- rse.shutdown(Time.seconds(5));
}
}
- private static class TestSavepointTriggerHandler extends TestHandler<EmptyRequestBody, SavepointTriggerResponseBody, SavepointMessageParameters> {
+ private class TestSavepointTriggerHandler extends TestHandler<EmptyRequestBody, SavepointTriggerResponseBody, SavepointMessageParameters> {
private final String expectedSavepointDirectory;
@@ -273,42 +339,19 @@ public class RestClusterClientTest extends TestLogger {
@Test
public void testListJobs() throws Exception {
-
- Configuration config = new Configuration();
- config.setString(JobManagerOptions.ADDRESS, "localhost");
-
- RestServerEndpointConfiguration rsec = RestServerEndpointConfiguration.fromConfiguration(config);
-
- TestListJobsHandler listJobsHandler = new TestListJobsHandler();
-
- RestServerEndpoint rse = new RestServerEndpoint(rsec) {
- @Override
- protected List<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> initializeHandlers(CompletableFuture<String> restAddressFuture) {
-
- List<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> handlers = new ArrayList<>();
- handlers.add(Tuple2.of(listJobsHandler.getMessageHeaders(), listJobsHandler));
- return handlers;
- }
- };
-
- RestClusterClient rcc = new RestClusterClient(config);
- try {
- rse.start();
-
+ try (TestRestServerEndpoint ignored = createRestServerEndpoint(new TestListJobsHandler())) {
{
- CompletableFuture<Collection<JobStatusMessage>> jobDetailsFuture = rcc.listJobs();
+ CompletableFuture<Collection<JobStatusMessage>> jobDetailsFuture = restClusterClient.listJobs();
Collection<JobStatusMessage> jobDetails = jobDetailsFuture.get();
Iterator<JobStatusMessage> jobDetailsIterator = jobDetails.iterator();
JobStatusMessage job1 = jobDetailsIterator.next();
JobStatusMessage job2 = jobDetailsIterator.next();
Assert.assertNotEquals("The job statues should not be equal.", job1.getJobState(), job2.getJobState());
}
- } finally {
- rcc.shutdown();
- rse.shutdown(Time.seconds(5));
- }}
+ }
+ }
- private static class TestListJobsHandler extends TestHandler<EmptyRequestBody, MultipleJobsDetails, EmptyMessageParameters> {
+ private class TestListJobsHandler extends TestHandler<EmptyRequestBody, MultipleJobsDetails, EmptyMessageParameters> {
private TestListJobsHandler() {
super(JobsOverviewHeaders.getInstance());
@@ -322,15 +365,49 @@ public class RestClusterClientTest extends TestLogger {
}
}
- private abstract static class TestHandler<R extends RequestBody, P extends ResponseBody, M extends MessageParameters> extends AbstractRestHandler<DispatcherGateway, R, P, M> {
+ private abstract class TestHandler<R extends RequestBody, P extends ResponseBody, M extends MessageParameters> extends AbstractRestHandler<DispatcherGateway, R, P, M> {
private TestHandler(MessageHeaders<R, P, M> headers) {
super(
- CompletableFuture.completedFuture(restAddress),
+ CompletableFuture.completedFuture(REST_ADDRESS),
mockGatewayRetriever,
RpcUtils.INF_TIMEOUT,
Collections.emptyMap(),
headers);
}
}
+
+ private TestRestServerEndpoint createRestServerEndpoint(
+ final AbstractRestHandler<?, ?, ?, ?>... abstractRestHandlers) throws Exception {
+ final TestRestServerEndpoint testRestServerEndpoint = new TestRestServerEndpoint(abstractRestHandlers);
+ testRestServerEndpoint.start();
+ return testRestServerEndpoint;
+ }
+
+ private class TestRestServerEndpoint extends RestServerEndpoint implements AutoCloseable {
+
+ private final AbstractRestHandler<?, ?, ?, ?>[] abstractRestHandlers;
+
+ TestRestServerEndpoint(final AbstractRestHandler<?, ?, ?, ?>... abstractRestHandlers) {
+ super(restServerEndpointConfiguration);
+ this.abstractRestHandlers = abstractRestHandlers;
+ }
+
+ @Override
+ protected List<Tuple2<RestHandlerSpecification, ChannelInboundHandler>>
+ initializeHandlers(CompletableFuture<String> restAddressFuture) {
+ final List<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> handlers = new ArrayList<>();
+ for (final AbstractRestHandler abstractRestHandler : abstractRestHandlers) {
+ handlers.add(Tuple2.of(
+ abstractRestHandler.getMessageHeaders(),
+ abstractRestHandler));
+ }
+ return handlers;
+ }
+
+ @Override
+ public void close() throws Exception {
+ shutdown(Time.seconds(5));
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/06922753/flink-clients/src/test/java/org/apache/flink/client/program/rest/retry/ExponentialWaitStrategyTest.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/test/java/org/apache/flink/client/program/rest/retry/ExponentialWaitStrategyTest.java b/flink-clients/src/test/java/org/apache/flink/client/program/rest/retry/ExponentialWaitStrategyTest.java
new file mode 100644
index 0000000..747ab6c
--- /dev/null
+++ b/flink-clients/src/test/java/org/apache/flink/client/program/rest/retry/ExponentialWaitStrategyTest.java
@@ -0,0 +1,89 @@
+/*
+ * 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.retry;
+
+import org.junit.Test;
+
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
+
+/**
+ * Tests for {@link ExponentialWaitStrategy}.
+ */
+public class ExponentialWaitStrategyTest {
+
+ @Test
+ public void testNegativeInitialWait() {
+ try {
+ new ExponentialWaitStrategy(0, 1);
+ fail("Expected exception not thrown.");
+ } catch (final IllegalArgumentException e) {
+ assertThat(e.getMessage(), containsString("initialWait must be positive"));
+ }
+ }
+
+ @Test
+ public void testNegativeMaxWait() {
+ try {
+ new ExponentialWaitStrategy(1, -1);
+ fail("Expected exception not thrown.");
+ } catch (final IllegalArgumentException e) {
+ assertThat(e.getMessage(), containsString("maxWait must be positive"));
+ }
+ }
+
+ @Test
+ public void testInitialWaitGreaterThanMaxWait() {
+ try {
+ new ExponentialWaitStrategy(2, 1);
+ fail("Expected exception not thrown.");
+ } catch (final IllegalArgumentException e) {
+ assertThat(e.getMessage(), containsString("initialWait must be lower than or equal to maxWait"));
+ }
+ }
+
+ @Test
+ public void testMaxSleepTime() {
+ final long sleepTime = new ExponentialWaitStrategy(1, 1).sleepTime(100);
+ assertThat(sleepTime, equalTo(1L));
+ }
+
+ @Test
+ public void testExponentialGrowth() {
+ final ExponentialWaitStrategy exponentialWaitStrategy = new ExponentialWaitStrategy(1, 1000);
+ assertThat(exponentialWaitStrategy.sleepTime(3) / exponentialWaitStrategy.sleepTime(2), equalTo(2L));
+ }
+
+ @Test
+ public void testMaxAttempts() {
+ final long maxWait = 1000;
+ final ExponentialWaitStrategy exponentialWaitStrategy = new ExponentialWaitStrategy(1, maxWait);
+ assertThat(exponentialWaitStrategy.sleepTime(Long.MAX_VALUE), equalTo(maxWait));
+ }
+
+ @Test
+ public void test64Attempts() {
+ final long maxWait = 1000;
+ final ExponentialWaitStrategy exponentialWaitStrategy = new ExponentialWaitStrategy(1, maxWait);
+ assertThat(exponentialWaitStrategy.sleepTime(64), equalTo(maxWait));
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/06922753/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobMessageParameters.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobMessageParameters.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobMessageParameters.java
index 1155892..be3bf98 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobMessageParameters.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobMessageParameters.java
@@ -26,7 +26,7 @@ import java.util.Collections;
*/
public class JobMessageParameters extends MessageParameters {
- protected final JobIDPathParameter jobPathParameter = new JobIDPathParameter();
+ public final JobIDPathParameter jobPathParameter = new JobIDPathParameter();
@Override
public Collection<MessagePathParameter<?>> getPathParameters() {