### Question:
NettyMessagingService implements ManagedMessagingService { @Override public CompletableFuture<Void> sendAsync(Address address, String type, byte[] payload, boolean keepAlive) { long messageId = messageIdGenerator.incrementAndGet(); ProtocolRequest message = new ProtocolRequest( messageId, returnAddress, type, payload); return executeOnPooledConnection(address, type, c -> c.sendAsync(message), MoreExecutors.directExecutor()); } NettyMessagingService(String cluster, Address address, MessagingConfig config);   NettyMessagingService(String cluster, Address address, MessagingConfig config, ProtocolVersion protocolVersion); @Override Address address(); @Override CompletableFuture<MessagingService> start(); @Override boolean isRunning(); @Override CompletableFuture<Void> sendAsync(Address address, String type, byte[] payload, boolean keepAlive); @Override CompletableFuture<byte[]> sendAndReceive(Address address, String type, byte[] payload, boolean keepAlive); @Override CompletableFuture<byte[]> sendAndReceive(Address address, String type, byte[] payload, boolean keepAlive, Executor executor); @Override CompletableFuture<byte[]> sendAndReceive(Address address, String type, byte[] payload, boolean keepAlive, Duration timeout); @Override CompletableFuture<byte[]> sendAndReceive(Address address, String type, byte[] payload, boolean keepAlive, Duration timeout, Executor executor); @Override void registerHandler(String type, BiConsumer<Address, byte[]> handler, Executor executor); @Override void registerHandler(String type, BiFunction<Address, byte[], byte[]> handler, Executor executor); @Override void registerHandler(String type, BiFunction<Address, byte[], CompletableFuture<byte[]>> handler); @Override void unregisterHandler(String type); @Override CompletableFuture<Void> stop();  }### Answer:
@Test public void testSendAsyncToUnresolvable() { final Address unresolvable = Address.from("unknown.local", address1.port()); final String subject = nextSubject(); final CompletableFuture<Void> response = netty1.sendAsync(unresolvable, subject, "hello world".getBytes()); assertTrue(response.isCompletedExceptionally()); }
@Test public void testSendAsync() { String subject = nextSubject(); CountDownLatch latch1 = new CountDownLatch(1); CompletableFuture<Void> response = netty1.sendAsync(address2, subject, "hello world".getBytes()); response.whenComplete((r, e) -> { assertNull(e); latch1.countDown(); }); Uninterruptibles.awaitUninterruptibly(latch1); CountDownLatch latch2 = new CountDownLatch(1); response = netty1.sendAsync(invalidAddress, subject, "hello world".getBytes()); response.whenComplete((r, e) -> { assertNotNull(e); assertTrue(e instanceof ConnectException); latch2.countDown(); }); Uninterruptibles.awaitUninterruptibly(latch2); }