CDSShardAccessImpl implements CDSShardAccess, LeaderLocationListener, AutoCloseable { @Override public CompletionStage<Void> makeLeaderLocal() { checkNotClosed(); final Future<ActorRef> localShardReply = actorUtils.findLocalShardAsync(ClusterUtils.getCleanShardName(prefix.getRootIdentifier())); final scala.concurrent.Promise<Object> makeLeaderLocalAsk = Futures.promise(); localShardReply.onComplete(new OnComplete<ActorRef>() { @Override public void onComplete(final Throwable failure, final ActorRef actorRef) { if (failure instanceof LocalShardNotFoundException) { LOG.debug("No local shard found for {} - Cannot request leadership transfer to local shard.", getShardIdentifier(), failure); makeLeaderLocalAsk.failure(failure); } else if (failure != null) { LOG.debug("Failed to find local shard for {} - Cannot request leadership transfer to local shard.", getShardIdentifier(), failure); makeLeaderLocalAsk.failure(failure); } else { makeLeaderLocalAsk .completeWith(actorUtils .executeOperationAsync(actorRef, MakeLeaderLocal.INSTANCE, makeLeaderLocalTimeout)); } } }, actorUtils.getClientDispatcher()); Future<Void> makeLeaderLocalFuture = makeLeaderLocalAsk.future() .transform(new Mapper<Object, Void>() { @Override public Void apply(final Object parameter) { return null; } }, new Mapper<Throwable, Throwable>() { @Override public Throwable apply(final Throwable parameter) { if (parameter instanceof LeadershipTransferFailedException) { return parameter; } return new LeadershipTransferFailedException("Leadership transfer failed", parameter); } }, actorUtils.getClientDispatcher()); return FutureConverters.toJava(makeLeaderLocalFuture); } CDSShardAccessImpl(final DOMDataTreeIdentifier prefix, final ActorUtils actorUtils); @Override DOMDataTreeIdentifier getShardIdentifier(); @Override LeaderLocation getLeaderLocation(); @Override CompletionStage<Void> makeLeaderLocal(); @Override LeaderLocationListenerRegistration<L> registerLeaderLocationListener(final L listener); @Override @SuppressWarnings("checkstyle:IllegalCatch") void onLeaderLocationChanged(final LeaderLocation location); @Override void close();  }
@Test @SuppressWarnings("checkstyle:IllegalCatch") public void testMakeLeaderLocal() throws Exception { final FiniteDuration timeout = new FiniteDuration(5, TimeUnit.SECONDS); final ActorRef localShardRef = mock(ActorRef.class); final Future<ActorRef> localShardRefFuture = Futures.successful(localShardRef); doReturn(localShardRefFuture).when(context).findLocalShardAsync(any()); doReturn(Futures.successful(null)).when(context).executeOperationAsync((ActorRef) any(), any(), any()); doReturn(getSystem().dispatcher()).when(context).getClientDispatcher(); assertEquals(waitOnAsyncTask(shardAccess.makeLeaderLocal(), timeout), null); doReturn(Futures.failed(new LeadershipTransferFailedException("Failure"))) .when(context).executeOperationAsync((ActorRef) any(), any(), any()); try { waitOnAsyncTask(shardAccess.makeLeaderLocal(), timeout); fail("makeLeaderLocal operation should not be successful"); } catch (final Exception e) { assertTrue(e instanceof LeadershipTransferFailedException); } doReturn(Futures.failed(new LocalShardNotFoundException("Local shard not found"))) .when(context).findLocalShardAsync(any()); try { waitOnAsyncTask(shardAccess.makeLeaderLocal(), timeout); fail("makeLeaderLocal operation should not be successful"); } catch (final Exception e) { assertTrue(e instanceof LeadershipTransferFailedException); assertTrue(e.getCause() instanceof LocalShardNotFoundException); } shardAccess.close(); try { shardAccess.makeLeaderLocal(); fail("Should have thrown IllegalStateEx. ShardAccess is closed"); } catch (final Exception e) { assertTrue(e instanceof IllegalStateException); } }