tillrohrmann commented on a change in pull request #7082: [FLINK-10419] Using
DeclineCheckpoint message class when invoking RPC declineCheckpoint
URL: https://github.com/apache/flink/pull/7082#discussion_r232988684
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
##########
@@ -248,6 +257,88 @@ public static void teardownClass() {
}
}
+ @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 -> {
+ try {
+ gateway.declineCheckpoint(new
DeclineCheckpoint(
Review comment:
We could create the `DeclineCheckpoint` instance outside of the future
callback. Then we would not need to handle the exception.
----------------------------------------------------------------
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:
[email protected]
With regards,
Apache Git Services
