You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by dw...@apache.org on 2018/11/14 08:07:09 UTC
[flink] branch release-1.6 updated (76d7fb1 -> 12a35d1)
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a change to branch release-1.6
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 76d7fb1 [FLINK-10863][tests] Assign UIDs to all operators in DataStreamAllroundTestProgram
new f4dd641 [FLINK-10419] Using DeclineCheckpoint message class when invoking RPC declineCheckpoint
new 12a35d1 [FLINK-10419] Added IT test to check declineCheckpoint invocation
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../checkpoint/CheckpointCoordinatorGateway.java | 7 +-
.../apache/flink/runtime/jobmaster/JobMaster.java | 8 +--
.../org/apache/flink/runtime/rpc/RpcUtils.java | 27 ++++++++
.../taskexecutor/rpc/RpcCheckpointResponder.java | 6 +-
.../flink/runtime/jobmaster/JobMasterTest.java | 77 ++++++++++++++++++++++
.../jobmaster/utils/TestingJobMasterGateway.java | 9 +--
.../utils/TestingJobMasterGatewayBuilder.java | 5 +-
7 files changed, 120 insertions(+), 19 deletions(-)
[flink] 02/02: [FLINK-10419] Added IT test to check
declineCheckpoint invocation
Posted by dw...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch release-1.6
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 12a35d1c96a07448fbf0dbe7d3025a65a92b04e5
Author: Dawid Wysakowicz <dw...@apache.org>
AuthorDate: Tue Nov 13 11:42:08 2018 +0100
[FLINK-10419] Added IT test to check declineCheckpoint invocation
---
.../org/apache/flink/runtime/rpc/RpcUtils.java | 27 ++++++++
.../flink/runtime/jobmaster/JobMasterTest.java | 77 ++++++++++++++++++++++
2 files changed, 104 insertions(+)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcUtils.java
index c90a8b5..2f656d0 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcUtils.java
@@ -19,9 +19,13 @@
package org.apache.flink.runtime.rpc;
import org.apache.flink.api.common.time.Time;
+import org.apache.flink.runtime.concurrent.FutureUtils;
+import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashSet;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -87,6 +91,29 @@ public class RpcUtils {
rpcService.stopService().get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
}
+ /**
+ * Shuts the given rpc services down and waits for their termination.
+ *
+ * @param rpcServices to shut down
+ * @param timeout for this operation
+ * @throws InterruptedException if the operation has been interrupted
+ * @throws ExecutionException if a problem occurred
+ * @throws TimeoutException if a timeout occurred
+ */
+ public static void terminateRpcServices(
+ Time timeout,
+ RpcService... rpcServices) throws InterruptedException, ExecutionException, TimeoutException {
+ final Collection<CompletableFuture<?>> terminationFutures = new ArrayList<>(rpcServices.length);
+
+ for (RpcService service : rpcServices) {
+ if (service != null) {
+ terminationFutures.add(service.stopService());
+ }
+ }
+
+ FutureUtils.waitForAll(terminationFutures).get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+ }
+
// We don't want this class to be instantiable
private RpcUtils() {}
}
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 86afa90..184f5a6 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
@@ -27,6 +27,7 @@ import org.apache.flink.configuration.BlobServerOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.core.testutils.OneShotLatch;
+import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.blob.BlobServer;
import org.apache.flink.runtime.blob.VoidBlobStore;
import org.apache.flink.runtime.checkpoint.CheckpointProperties;
@@ -67,12 +68,15 @@ import org.apache.flink.runtime.jobmaster.factories.UnregisteredJobManagerJobMet
import org.apache.flink.runtime.jobmaster.slotpool.DefaultSlotPoolFactory;
import org.apache.flink.runtime.leaderretrieval.SettableLeaderRetrievalService;
import org.apache.flink.runtime.messages.Acknowledge;
+import org.apache.flink.runtime.messages.checkpoint.DeclineCheckpoint;
import org.apache.flink.runtime.registration.RegistrationResponse;
import org.apache.flink.runtime.resourcemanager.ResourceManagerId;
import org.apache.flink.runtime.resourcemanager.SlotRequest;
import org.apache.flink.runtime.resourcemanager.utils.TestingResourceManagerGateway;
+import org.apache.flink.runtime.rpc.RpcService;
import org.apache.flink.runtime.rpc.RpcUtils;
import org.apache.flink.runtime.rpc.TestingRpcService;
+import org.apache.flink.runtime.rpc.akka.AkkaRpcService;
import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation;
import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.runtime.taskexecutor.TestingTaskExecutorGateway;
@@ -83,10 +87,13 @@ import org.apache.flink.runtime.taskmanager.TaskExecutionState;
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.testutils.ClassLoaderUtils;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.SerializedThrowable;
import org.apache.flink.util.TestLogger;
+import akka.actor.ActorSystem;
import org.hamcrest.Matchers;
import org.junit.After;
import org.junit.AfterClass;
@@ -102,6 +109,7 @@ import javax.annotation.Nullable;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
+import java.net.URLClassLoader;
import java.util.Collection;
import java.util.Collections;
import java.util.concurrent.ArrayBlockingQueue;
@@ -212,6 +220,75 @@ public class JobMasterTest extends TestLogger {
}
@Test
+ public void testDeclineCheckpointInvocationWithUserException() throws Exception {
+ RpcService rpcService1 = null;
+ RpcService rpcService2 = null;
+ try {
+ final ActorSystem actorSystem1 = AkkaUtils.createDefaultActorSystem();
+ final ActorSystem actorSystem2 = AkkaUtils.createDefaultActorSystem();
+
+ rpcService1 = new AkkaRpcService(actorSystem1, testingTimeout);
+ rpcService2 = new AkkaRpcService(actorSystem2, testingTimeout);
+
+ final CompletableFuture<Throwable> declineCheckpointMessageFuture = new CompletableFuture<>();
+
+ final JobManagerSharedServices jobManagerSharedServices = new TestingJobManagerSharedServicesBuilder().build();
+ final JobMasterConfiguration jobMasterConfiguration = JobMasterConfiguration.fromConfiguration(configuration);
+ final JobMaster jobMaster = new JobMaster(
+ rpcService1,
+ jobMasterConfiguration,
+ jmResourceId,
+ jobGraph,
+ haServices,
+ DefaultSlotPoolFactory.fromConfiguration(configuration, rpcService1),
+ jobManagerSharedServices,
+ heartbeatServices,
+ blobServer,
+ UnregisteredJobManagerJobMetricGroupFactory.INSTANCE,
+ new NoOpOnCompletionActions(),
+ testingFatalErrorHandler,
+ JobMasterTest.class.getClassLoader()) {
+ @Override
+ public void declineCheckpoint(DeclineCheckpoint declineCheckpoint) {
+ declineCheckpointMessageFuture.complete(declineCheckpoint.getReason());
+ }
+ };
+
+ jobMaster.start(jobMasterId, testingTimeout).get();
+
+ final String className = "UserException";
+ final URLClassLoader userClassLoader = ClassLoaderUtils.compileAndLoadJava(
+ temporaryFolder.newFolder(),
+ className + ".java",
+ String.format("public class %s extends RuntimeException { public %s() {super(\"UserMessage\");} }",
+ className,
+ className));
+
+ Throwable userException = (Throwable) Class.forName(className, false, userClassLoader).newInstance();
+
+ CompletableFuture<JobMasterGateway> jobMasterGateway =
+ rpcService2.connect(jobMaster.getAddress(), jobMaster.getFencingToken(), JobMasterGateway.class);
+
+ jobMasterGateway.thenAccept(gateway -> {
+ gateway.declineCheckpoint(new DeclineCheckpoint(
+ jobGraph.getJobID(),
+ new ExecutionAttemptID(1, 1),
+ 1,
+ userException
+ )
+ );
+ });
+
+ Throwable throwable = declineCheckpointMessageFuture.get(testingTimeout.toMilliseconds(),
+ TimeUnit.MILLISECONDS);
+ assertThat(throwable, instanceOf(SerializedThrowable.class));
+ assertThat(throwable.getMessage(), equalTo(userException.getMessage()));
+ } finally {
+ RpcUtils.terminateRpcServices(testingTimeout, rpcService1, rpcService2);
+ }
+ }
+
+ @Test
public void testHeartbeatTimeoutWithTaskManager() throws Exception {
final CompletableFuture<ResourceID> heartbeatResourceIdFuture = new CompletableFuture<>();
final CompletableFuture<JobID> disconnectedJobManagerFuture = new CompletableFuture<>();
[flink] 01/02: [FLINK-10419] Using DeclineCheckpoint message class
when invoking RPC declineCheckpoint
Posted by dw...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch release-1.6
in repository https://gitbox.apache.org/repos/asf/flink.git
commit f4dd6410c3383d3de20fb2e9991609b22d02acc6
Author: Dawid Wysakowicz <dw...@apache.org>
AuthorDate: Mon Nov 12 18:21:40 2018 +0100
[FLINK-10419] Using DeclineCheckpoint message class when invoking RPC declineCheckpoint
---
.../flink/runtime/checkpoint/CheckpointCoordinatorGateway.java | 7 ++-----
.../main/java/org/apache/flink/runtime/jobmaster/JobMaster.java | 8 +-------
.../flink/runtime/taskexecutor/rpc/RpcCheckpointResponder.java | 6 +++++-
.../flink/runtime/jobmaster/utils/TestingJobMasterGateway.java | 9 +++++----
.../runtime/jobmaster/utils/TestingJobMasterGatewayBuilder.java | 5 +++--
5 files changed, 16 insertions(+), 19 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorGateway.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorGateway.java
index 22244f6..b8dc554 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorGateway.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorGateway.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.checkpoint;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+import org.apache.flink.runtime.messages.checkpoint.DeclineCheckpoint;
import org.apache.flink.runtime.rpc.RpcGateway;
public interface CheckpointCoordinatorGateway extends RpcGateway {
@@ -31,9 +32,5 @@ public interface CheckpointCoordinatorGateway extends RpcGateway {
final CheckpointMetrics checkpointMetrics,
final TaskStateSnapshot subtaskState);
- void declineCheckpoint(
- JobID jobID,
- ExecutionAttemptID executionAttemptID,
- long checkpointId,
- Throwable cause);
+ void declineCheckpoint(DeclineCheckpoint declineCheckpoint);
}
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 c19346f..c8a8e88 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
@@ -688,13 +688,7 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
// TODO: This method needs a leader session ID
@Override
- public void declineCheckpoint(
- final JobID jobID,
- final ExecutionAttemptID executionAttemptID,
- final long checkpointID,
- final Throwable reason) {
- final DeclineCheckpoint decline = new DeclineCheckpoint(
- jobID, executionAttemptID, checkpointID, reason);
+ public void declineCheckpoint(DeclineCheckpoint decline) {
final CheckpointCoordinator checkpointCoordinator = executionGraph.getCheckpointCoordinator();
if (checkpointCoordinator != null) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/rpc/RpcCheckpointResponder.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/rpc/RpcCheckpointResponder.java
index c8f7357..10c9e2f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/rpc/RpcCheckpointResponder.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/rpc/RpcCheckpointResponder.java
@@ -23,6 +23,7 @@ import org.apache.flink.runtime.checkpoint.CheckpointCoordinatorGateway;
import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+import org.apache.flink.runtime.messages.checkpoint.DeclineCheckpoint;
import org.apache.flink.runtime.taskmanager.CheckpointResponder;
import org.apache.flink.util.Preconditions;
@@ -57,6 +58,9 @@ public class RpcCheckpointResponder implements CheckpointResponder {
long checkpointId,
Throwable cause) {
- checkpointCoordinatorGateway.declineCheckpoint(jobID, executionAttemptID, checkpointId, cause);
+ checkpointCoordinatorGateway.declineCheckpoint(new DeclineCheckpoint(jobID,
+ executionAttemptID,
+ checkpointId,
+ cause));
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGateway.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGateway.java
index c9c55a1..f5f7f8e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGateway.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGateway.java
@@ -41,6 +41,7 @@ import org.apache.flink.runtime.jobmaster.RescalingBehaviour;
import org.apache.flink.runtime.jobmaster.SerializedInputSplit;
import org.apache.flink.runtime.jobmaster.message.ClassloadingProps;
import org.apache.flink.runtime.messages.Acknowledge;
+import org.apache.flink.runtime.messages.checkpoint.DeclineCheckpoint;
import org.apache.flink.runtime.messages.webmonitor.JobDetails;
import org.apache.flink.runtime.query.KvStateLocation;
import org.apache.flink.runtime.registration.RegistrationResponse;
@@ -144,7 +145,7 @@ public class TestingJobMasterGateway implements JobMasterGateway {
private final Consumer<Tuple5<JobID, ExecutionAttemptID, Long, CheckpointMetrics, TaskStateSnapshot>> acknowledgeCheckpointConsumer;
@Nonnull
- private final Consumer<Tuple4<JobID, ExecutionAttemptID, Long, Throwable>> declineCheckpointConsumer;
+ private final Consumer<DeclineCheckpoint> declineCheckpointConsumer;
@Nonnull
private final Supplier<JobMasterId> fencingTokenSupplier;
@@ -183,7 +184,7 @@ public class TestingJobMasterGateway implements JobMasterGateway {
@Nonnull Function<JobVertexID, CompletableFuture<OperatorBackPressureStatsResponse>> requestOperatorBackPressureStatsFunction,
@Nonnull BiConsumer<AllocationID, Throwable> notifyAllocationFailureConsumer,
@Nonnull Consumer<Tuple5<JobID, ExecutionAttemptID, Long, CheckpointMetrics, TaskStateSnapshot>> acknowledgeCheckpointConsumer,
- @Nonnull Consumer<Tuple4<JobID, ExecutionAttemptID, Long, Throwable>> declineCheckpointConsumer,
+ @Nonnull Consumer<DeclineCheckpoint> declineCheckpointConsumer,
@Nonnull Supplier<JobMasterId> fencingTokenSupplier,
@Nonnull BiFunction<JobID, String, CompletableFuture<KvStateLocation>> requestKvStateLocationFunction,
@Nonnull Function<Tuple6<JobID, JobVertexID, KeyGroupRange, String, KvStateID, InetSocketAddress>, CompletableFuture<Acknowledge>> notifyKvStateRegisteredFunction,
@@ -335,8 +336,8 @@ public class TestingJobMasterGateway implements JobMasterGateway {
}
@Override
- public void declineCheckpoint(JobID jobID, ExecutionAttemptID executionAttemptID, long checkpointId, Throwable cause) {
- declineCheckpointConsumer.accept(Tuple4.of(jobID, executionAttemptID, checkpointId, cause));
+ public void declineCheckpoint(DeclineCheckpoint declineCheckpoint) {
+ declineCheckpointConsumer.accept(declineCheckpoint);
}
@Override
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGatewayBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGatewayBuilder.java
index e40b752..b52df9e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGatewayBuilder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGatewayBuilder.java
@@ -40,6 +40,7 @@ import org.apache.flink.runtime.jobmaster.RescalingBehaviour;
import org.apache.flink.runtime.jobmaster.SerializedInputSplit;
import org.apache.flink.runtime.jobmaster.message.ClassloadingProps;
import org.apache.flink.runtime.messages.Acknowledge;
+import org.apache.flink.runtime.messages.checkpoint.DeclineCheckpoint;
import org.apache.flink.runtime.messages.webmonitor.JobDetails;
import org.apache.flink.runtime.query.KvStateLocation;
import org.apache.flink.runtime.query.UnknownKvStateLocation;
@@ -96,7 +97,7 @@ public class TestingJobMasterGatewayBuilder {
private Function<JobVertexID, CompletableFuture<OperatorBackPressureStatsResponse>> requestOperatorBackPressureStatsFunction = ignored -> CompletableFuture.completedFuture(OperatorBackPressureStatsResponse.of(null));
private BiConsumer<AllocationID, Throwable> notifyAllocationFailureConsumer = (ignoredA, ignoredB) -> {};
private Consumer<Tuple5<JobID, ExecutionAttemptID, Long, CheckpointMetrics, TaskStateSnapshot>> acknowledgeCheckpointConsumer = ignored -> {};
- private Consumer<Tuple4<JobID, ExecutionAttemptID, Long, Throwable>> declineCheckpointConsumer = ignored -> {};
+ private Consumer<DeclineCheckpoint> declineCheckpointConsumer = ignored -> {};
private Supplier<JobMasterId> fencingTokenSupplier = () -> JOB_MASTER_ID;
private BiFunction<JobID, String, CompletableFuture<KvStateLocation>> requestKvStateLocationFunction = (ignoredA, registrationName) -> FutureUtils.completedExceptionally(new UnknownKvStateLocation(registrationName));
private Function<Tuple6<JobID, JobVertexID, KeyGroupRange, String, KvStateID, InetSocketAddress>, CompletableFuture<Acknowledge>> notifyKvStateRegisteredFunction = ignored -> CompletableFuture.completedFuture(Acknowledge.get());
@@ -222,7 +223,7 @@ public class TestingJobMasterGatewayBuilder {
return this;
}
- public TestingJobMasterGatewayBuilder setDeclineCheckpointConsumer(Consumer<Tuple4<JobID, ExecutionAttemptID, Long, Throwable>> declineCheckpointConsumer) {
+ public TestingJobMasterGatewayBuilder setDeclineCheckpointConsumer(Consumer<DeclineCheckpoint> declineCheckpointConsumer) {
this.declineCheckpointConsumer = declineCheckpointConsumer;
return this;
}