ModuleShardBackendResolver extends AbstractShardBackendResolver { @Override public void close() { shardAvailabilityChangesRegFuture.onComplete(new OnComplete<Registration>() { @Override public void onComplete(Throwable failure, Registration reply) { reply.close(); } }, ExecutionContexts.global()); } ModuleShardBackendResolver(final ClientIdentifier clientId, final ActorUtils actorUtils); @Override CompletionStage<ShardBackendInfo> getBackendInfo(final Long cookie); @Override CompletionStage<ShardBackendInfo> refreshBackendInfo(final Long cookie,
            final ShardBackendInfo staleInfo); @Override void close(); @Override String resolveCookieName(Long cookie);  }
@SuppressWarnings("unchecked") @Test public void testNotifyWhenBackendInfoIsStale() { final RegisterForShardAvailabilityChanges regMessage = shardManagerProbe.expectMsgClass(RegisterForShardAvailabilityChanges.class); Registration mockReg = mock(Registration.class); shardManagerProbe.reply(new Status.Success(mockReg)); Consumer<Long> mockCallback = mock(Consumer.class); final Registration callbackReg = moduleShardBackendResolver.notifyWhenBackendInfoIsStale(mockCallback); regMessage.getCallback().accept(DefaultShardStrategy.DEFAULT_SHARD); verify(mockCallback, timeout(5000)).accept(Long.valueOf(0)); reset(mockCallback); callbackReg.close(); regMessage.getCallback().accept(DefaultShardStrategy.DEFAULT_SHARD); Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS); verifyNoMoreInteractions(mockCallback); }