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_r232989225
##########
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(
+ jobGraph.getJobID(),
+ new
ExecutionAttemptID(1, 1),
+ 1,
+ userException
+ )
+ );
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+
+ Throwable throwable =
declineCheckpointMessageFuture.get(testingTimeout.toMilliseconds(),
+ TimeUnit.MILLISECONDS);
+ assertThat(throwable,
instanceOf(SerializedThrowable.class));
+ assertThat(throwable.getMessage(),
equalTo(userException.getMessage()));
+ } finally {
+ final Collection<CompletableFuture<?>>
terminationFutures = new ArrayList<>(2);
+
+ if (rpcService1 != null) {
+
terminationFutures.add(rpcService1.stopService());
+ }
+ if (rpcService2 != null) {
+
terminationFutures.add(rpcService2.stopService());
+ }
+
+
FutureUtils.waitForAll(terminationFutures).get(testingTimeout.toMilliseconds(),
TimeUnit.MILLISECONDS);
Review comment:
We could also use `RpcUtils#terminateRpcService` or extend to take multiple
`RpcServices` for termination.
----------------------------------------------------------------
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
