[Fixed] close connection when flush fails during drain.

Fixeds #324

Signed-off-by: Matthias Hanel <mh@synadia.com>

src/examples/java/io/nats/examples/NatsSub.java

@@ -15,6 +15,7 @@
 
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
 
 import io.nats.client.Connection;
 import io.nats.client.Message;

src/main/java/io/nats/client/impl/NatsConnection.java

@@ -1791,8 +1791,13 @@ public CompletableFuture<Boolean> drain(Duration timeout) throws TimeoutExceptio
             cons.sendUnsubForDrain();
         });
 
-        this.flush(timeout); // Flush and wait up to the timeout, if this fails, let the caller know
-
+        try {
+            this.flush(timeout); // Flush and wait up to the timeout, if this fails, let the caller know
+        } catch (Exception e) {
+            this.close(false);
+            throw e;
+        }
+
         consumers.forEach((cons) -> {
             cons.markUnsubedForDrain();
         });
@@ -1843,7 +1848,7 @@ public CompletableFuture<Boolean> drain(Duration timeout) throws TimeoutExceptio
                 this.processException(e);
             } finally {
                 try {
-                    this.close();// close the connection after the last flush
+                    this.close(false);// close the connection after the last flush
                 } catch (InterruptedException e) {
                     this.processException(e);
                 }

src/test/java/io/nats/client/impl/DrainTests.java

@@ -42,6 +42,39 @@
 
 public class DrainTests {
 
+    @Test
+    public void testLeftOverTimer() throws Exception {
+        try (NatsTestServer ts = new NatsTestServer(false);
+            final Connection subCon = Nats.connect(new Options.Builder().server(ts.getURI()).maxReconnects(0).build())){
+            assertTrue("Connected Status", Connection.Status.CONNECTED == subCon.getStatus());
+
+            Subscription sub = subCon.subscribe("draintest");
+            subCon.flush(Duration.ofSeconds(1)); // Get the sub to the server, so drain has things to do
+
+            assertTrue("Timer exists", Thread.getAllStackTraces().keySet().stream().anyMatch(thread -> {
+                return thread.getName().equals("Nats Connection Timer");
+            }));
+
+            ts.shutdown(); // shut down the server to fail drain and subsequent close
+            boolean timedOut = false;
+            try {
+                subCon.drain(Duration.ofSeconds(1));
+            } catch (java.util.concurrent.TimeoutException e) {
+                timedOut = true;
+            } finally {
+                subCon.close();
+                assertFalse("Timer gone", Thread.getAllStackTraces().keySet().stream().anyMatch(thread -> {
+                    return thread.getName().equals("Nats Connection Timer");
+                }));
+            }
+            assertTrue("Expect timeout exception", timedOut);
+            assertTrue("Expect closed connection", Connection.Status.CLOSED == subCon.getStatus());
+            assertFalse("Timer gone", Thread.getAllStackTraces().keySet().stream().anyMatch(thread -> {
+                return thread.getName().equals("Nats Connection Timer");
+            }));
+        }
+    }
+
     @Test
     public void testSimpleSubDrain() throws Exception {
         try (NatsTestServer ts = new NatsTestServer(false);