### Question:
CompletionCallbackActor extends AbstractLoggingActor { static Props props(CompletableFuture<SagaResponse> future) { return Props.create(CompletionCallbackActor.class, () -> new CompletionCallbackActor(future)); } private  CompletionCallbackActor(CompletableFuture<SagaResponse> future); @Override Receive createReceive();  }### Answer:
@Test public void killAllOnTransactionComplete() throws Exception { new TestKit(actorSystem) {{ CompletableFuture<SagaResponse> future = new CompletableFuture<>(); ActorRef actor = actorSystem.actorOf(CompletionCallbackActor.props(future)); actor.tell(context, noSender()); actor.tell(new TransactMessage(null, response), noSender()); await().atMost(2, TimeUnit.SECONDS) .until(() -> actor1.isTerminated() && actor2.isTerminated() && actor.isTerminated()); assertThat(future.get(), is(response)); }}; }
@Test public void killAllOnCompensationComplete() throws Exception { new TestKit(actorSystem) {{ CompletableFuture<SagaResponse> future = new CompletableFuture<>(); ActorRef actor = actorSystem.actorOf(CompletionCallbackActor.props(future)); actor.tell(context, noSender()); actor.tell(new CompensateMessage(response), noSender()); await().atMost(2, TimeUnit.SECONDS) .until(() -> actor1.isTerminated() && actor2.isTerminated() && actor.isTerminated()); assertThat(future.get(), is(response)); }}; }
@Test public void killAllOnFailure() throws Exception { new TestKit(actorSystem) {{ CompletableFuture<SagaResponse> future = new CompletableFuture<>(); ActorRef actor = actorSystem.actorOf(CompletionCallbackActor.props(future)); actor.tell(context, noSender()); actor.tell(new FailMessage(new RuntimeException("oops")), noSender()); await().atMost(2, TimeUnit.SECONDS) .until(() -> actor1.isTerminated() && actor2.isTerminated() && actor.isTerminated()); assertThat(future.get(), is(instanceOf(FailedSagaResponse.class))); }}; }
@Test public void tellLeafToCompensateOnAbort() throws Exception { new TestKit(actorSystem) {{ context.addActor(NoOpSagaRequest.SAGA_END_REQUEST.id(), getRef()); CompletableFuture<SagaResponse> future = new CompletableFuture<>(); ActorRef actor = actorSystem.actorOf(CompletionCallbackActor.props(future)); actor.tell(context, noSender()); AbortMessage message = new AbortMessage(new RuntimeException("oops")); actor.tell(message, noSender()); CompensateMessage compensateMessage = (CompensateMessage) receiveOne(Duration.of(2, SECONDS)); assertThat(compensateMessage.response(), is(message.response())); }}; }