asfgit closed pull request #6601: [FLINK-10193] Add @RpcTimeout to JobMasterGateway.triggerSavepoint
URL: https://github.com/apache/flink/pull/6601
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
index 01cb2b6b099..4a66d32a2ac 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
@@ -95,7 +95,6 @@
import org.apache.flink.runtime.rpc.FatalErrorHandler;
import org.apache.flink.runtime.rpc.FencedRpcEndpoint;
import org.apache.flink.runtime.rpc.RpcService;
-import org.apache.flink.runtime.rpc.RpcTimeout;
import org.apache.flink.runtime.rpc.akka.AkkaRpcServiceUtils;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.taskexecutor.AccumulatorReport;
@@ -909,7 +908,7 @@ public void heartbeatFromResourceManager(final ResourceID resourceID) {
}
@Override
- public CompletableFuture<JobDetails> requestJobDetails(@RpcTimeout Time timeout) {
+ public CompletableFuture<JobDetails> requestJobDetails(Time timeout) {
final ExecutionGraph currentExecutionGraph = executionGraph;
return CompletableFuture.supplyAsync(() -> WebMonitorUtils.createDetailsForJob(currentExecutionGraph), scheduledExecutorService);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java
index 981222d17a6..bc073c192bd 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java
@@ -268,7 +268,7 @@ void heartbeatFromTaskManager(
CompletableFuture<String> triggerSavepoint(
@Nullable final String targetDirectory,
final boolean cancelJob,
- final Time timeout);
+ @RpcTimeout final Time timeout);
/**
* Requests the statistics on operator back pressure.
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
index 66ca769165a..9a2bc97b62b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
@@ -92,6 +92,7 @@
import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
import org.apache.flink.runtime.testtasks.NoOpInvokable;
import org.apache.flink.runtime.util.TestingFatalErrorHandler;
+import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.InstantiationUtil;
import org.apache.flink.util.TestLogger;
@@ -108,6 +109,7 @@
import org.junit.rules.TemporaryFolder;
import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
import java.io.File;
import java.io.FileOutputStream;
@@ -122,24 +124,28 @@
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.contains;
import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.nullValue;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
/**
* Tests for {@link JobMaster}.
*/
public class JobMasterTest extends TestLogger {
- static final TestingInputSplit[] EMPTY_TESTING_INPUT_SPLITS = new TestingInputSplit[0];
+ private static final TestingInputSplit[] EMPTY_TESTING_INPUT_SPLITS = new TestingInputSplit[0];
@ClassRule
public static TemporaryFolder temporaryFolder = new TemporaryFolder();
@@ -418,7 +424,7 @@ public void testAutomaticRestartingWhenCheckpointing() throws Exception {
}
/**
- * Tests that an existing checkpoint will have precedence over an savepoint
+ * Tests that an existing checkpoint will have precedence over an savepoint.
*/
@Test
public void testCheckpointPrecedesSavepointRecovery() throws Exception {
@@ -470,7 +476,7 @@ public void testCheckpointPrecedesSavepointRecovery() throws Exception {
/**
* Tests that the JobMaster retries the scheduling of a job
- * in case of a missing slot offering from a registered TaskExecutor
+ * in case of a missing slot offering from a registered TaskExecutor.
*/
@Test
public void testSlotRequestTimeoutWhenNoSlotOffering() throws Exception {
@@ -874,9 +880,9 @@ public void testRequestPartitionState() throws Exception {
final CompletableFuture<TaskDeploymentDescriptor> tddFuture = new CompletableFuture<>();
final TestingTaskExecutorGateway testingTaskExecutorGateway = new TestingTaskExecutorGatewayBuilder()
.setSubmitTaskConsumer((taskDeploymentDescriptor, jobMasterId) -> {
- tddFuture.complete(taskDeploymentDescriptor);
- return CompletableFuture.completedFuture(Acknowledge.get());
- })
+ tddFuture.complete(taskDeploymentDescriptor);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ })
.createTestingTaskExecutorGateway();
rpcService.registerGateway(testingTaskExecutorGateway.getAddress(), testingTaskExecutorGateway);
@@ -917,6 +923,58 @@ public void testRequestPartitionState() throws Exception {
}
}
+ /**
+ * Tests that the timeout in {@link JobMasterGateway#triggerSavepoint(String, boolean, Time)}
+ * is respected.
+ */
+ @Test
+ public void testTriggerSavepointTimeout() throws Exception {
+ final JobMaster jobMaster = new JobMaster(
+ rpcService,
+ JobMasterConfiguration.fromConfiguration(configuration),
+ jmResourceId,
+ jobGraph,
+ haServices,
+ DefaultSlotPoolFactory.fromConfiguration(configuration, rpcService),
+ new TestingJobManagerSharedServicesBuilder().build(),
+ heartbeatServices,
+ blobServer,
+ UnregisteredJobManagerJobMetricGroupFactory.INSTANCE,
+ new NoOpOnCompletionActions(),
+ testingFatalErrorHandler,
+ JobMasterTest.class.getClassLoader()) {
+
+ @Override
+ public CompletableFuture<String> triggerSavepoint(
+ @Nullable final String targetDirectory,
+ final boolean cancelJob,
+ final Time timeout) {
+ return new CompletableFuture<>();
+ }
+ };
+
+ try {
+ final CompletableFuture<Acknowledge> startFuture = jobMaster.start(jobMasterId, testingTimeout);
+ startFuture.get(testingTimeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+
+ final JobMasterGateway jobMasterGateway = jobMaster.getSelfGateway(JobMasterGateway.class);
+ final CompletableFuture<String> savepointFutureLowTimeout = jobMasterGateway.triggerSavepoint("/tmp", false, Time.milliseconds(1));
+ final CompletableFuture<String> savepointFutureHighTimeout = jobMasterGateway.triggerSavepoint("/tmp", false, RpcUtils.INF_TIMEOUT);
+
+ try {
+ savepointFutureLowTimeout.get(testingTimeout.getSize(), testingTimeout.getUnit());
+ fail();
+ } catch (final ExecutionException e) {
+ final Throwable cause = ExceptionUtils.stripExecutionException(e);
+ assertThat(cause, instanceOf(TimeoutException.class));
+ }
+
+ assertThat(savepointFutureHighTimeout.isDone(), is(equalTo(false)));
+ } finally {
+ RpcUtils.terminateRpcEndpoint(jobMaster, testingTimeout);
+ }
+ }
+
private JobGraph producerConsumerJobGraph() {
final JobVertex producer = new JobVertex("Producer");
producer.setInvokableClass(NoOpInvokable.class);
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services