RoundRobinLoadBalancer should close connections gracefully (#1028)

__Motivation__

When a host used by `RoundRobinLoadBalancer` is unregistered from service discovery, we force close the connection. This means any outstanding requests on the connection will get interrupted. Since service discovery status is a hint we should err on the side of caution and let the outstanding requests complete gracefully.
A fallout of this decision is that a connection may actually be dead and an absence of writes may mean that the we will never detect closure hence graceful closure will never complete.
This is a trade-off between interrupting requests which may otherwise be successfully completed vs not force closing requests that may never complete. As requests could be infinitely latent in general, we assume that users have configured timeouts on the requests and hence all in-flight requests will eventually complete.

__Modification__

Use `closeAsyncGracefully()` instead of `closeAsync()` when a host turns inactive.

__Result__

Host inactivation does not interrupt in-flight requests.

servicetalk-loadbalancer/src/main/java/io/servicetalk/loadbalancer/RoundRobinLoadBalancer.java

@@ -378,7 +378,7 @@ void markInactive() {
             @SuppressWarnings("unchecked")
             List<C> toRemove = connectionsUpdater.getAndSet(this, INACTIVE);
             for (C conn : toRemove) {
-                conn.closeAsync().subscribe();
+                conn.closeAsyncGracefully().subscribe();
             }
         }
 

servicetalk-loadbalancer/src/test/java/io/servicetalk/loadbalancer/RoundRobinLoadBalancerTest.java

@@ -22,7 +22,6 @@
 import io.servicetalk.client.api.LoadBalancerReadyEvent;
 import io.servicetalk.client.api.NoAvailableHostException;
 import io.servicetalk.client.api.ServiceDiscovererEvent;
-import io.servicetalk.concurrent.CompletableSource.Processor;
 import io.servicetalk.concurrent.PublisherSource.Subscriber;
 import io.servicetalk.concurrent.PublisherSource.Subscription;
 import io.servicetalk.concurrent.api.Completable;
@@ -62,11 +61,10 @@
 import java.util.function.Function;
 import java.util.function.Predicate;
 
+import static io.servicetalk.concurrent.api.AsyncCloseables.emptyAsyncCloseable;
 import static io.servicetalk.concurrent.api.BlockingTestUtils.awaitIndefinitely;
-import static io.servicetalk.concurrent.api.Processors.newCompletableProcessor;
 import static io.servicetalk.concurrent.api.Single.failed;
 import static io.servicetalk.concurrent.api.Single.succeeded;
-import static io.servicetalk.concurrent.api.SourceAdapters.fromSource;
 import static io.servicetalk.concurrent.api.SourceAdapters.toSource;
 import static io.servicetalk.concurrent.internal.DeliberateException.DELIBERATE_EXCEPTION;
 import static io.servicetalk.concurrent.internal.ServiceTalkTestTimeout.DEFAULT_TIMEOUT_SECONDS;
@@ -89,6 +87,8 @@
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 public class RoundRobinLoadBalancerTest {
@@ -105,7 +105,7 @@ public class RoundRobinLoadBalancerTest {
     private final List<TestLoadBalancedConnection> connectionsCreated = new CopyOnWriteArrayList<>();
     private final Queue<Runnable> connectionRealizers = new ConcurrentLinkedQueue<>();
 
-    private TestPublisher<ServiceDiscovererEvent<String>> serviceDiscoveryPublisher = new TestPublisher<>();
+    private final TestPublisher<ServiceDiscovererEvent<String>> serviceDiscoveryPublisher = new TestPublisher<>();
     private RoundRobinLoadBalancer<String, TestLoadBalancedConnection> lb;
     private DelegatingConnectionFactory connectionFactory;
 
@@ -406,6 +406,16 @@ public void newConnectionIsClosedWhenSelectorRejects() throws Exception {
         awaitIndefinitely(connection.onClose());
     }
 
+    @Test
+    public void hostDownGracefulCloseConnection() throws Exception {
+        sendServiceDiscoveryEvents(upEvent("address-1"));
+        TestLoadBalancedConnection conn = lb.selectConnection(any()).toFuture().get();
+        sendServiceDiscoveryEvents(downEvent("address-1"));
+        conn.onClose().toFuture().get();
+        verify(conn).closeAsyncGracefully();
+        verify(conn, times(0)).closeAsync();
+    }
+
     @SuppressWarnings("unchecked")
     private void sendServiceDiscoveryEvents(final ServiceDiscovererEvent... events) {
         serviceDiscoveryPublisher.onNext((ServiceDiscovererEvent<String>[]) events);
@@ -437,12 +447,10 @@ private Single<TestLoadBalancedConnection> newRealizedConnectionSingle(final Str
     @SuppressWarnings("unchecked")
     private TestLoadBalancedConnection newConnection(final String address) {
         final TestLoadBalancedConnection cnx = mock(TestLoadBalancedConnection.class);
-        final Processor closeCompletable = newCompletableProcessor();
-        when(cnx.closeAsync()).thenAnswer(__ -> {
-            closeCompletable.onComplete();
-            return closeCompletable;
-        });
-        when(cnx.onClose()).thenReturn(fromSource(closeCompletable));
+        final ListenableAsyncCloseable closeable = emptyAsyncCloseable();
+        when(cnx.closeAsync()).thenReturn(closeable.closeAsync());
+        when(cnx.closeAsyncGracefully()).thenReturn(closeable.closeAsyncGracefully());
+        when(cnx.onClose()).thenReturn(closeable.onClose());
         when(cnx.address()).thenReturn(address);
         when(cnx.toString()).thenReturn(address + '@' + cnx.hashCode());