JAVA-1547: Abort pending requests when connection dropped

olim7t · olim7t · commit 0cfc799aef36 · 2017-07-13T14:34:12.000-07:00
diff --git a/changelog/README.md b/changelog/README.md
@@ -4,6 +4,7 @@
 
 ### 4.0.0-alpha1 (in progress)
 
+- [bug] JAVA-1547: Abort pending requests when connection dropped
 - [new feature] JAVA-1497: Port timestamp generators from 3.x
 - [improvement] JAVA-1539: Configure for deployment to Maven central
 - [new feature] JAVA-1519: Close channel if number of orphan stream ids exceeds a configurable 
diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/channel/InFlightHandler.java b/core/src/main/java/com/datastax/oss/driver/internal/core/channel/InFlightHandler.java
@@ -281,6 +281,15 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object event) throws E
     }
   }
 
+  @Override
+  public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+    // If the channel was closed normally (normal or forced shutdown), inFlight is already empty by
+    // the time we get here. So if it's not, it means the channel closed unexpectedly (e.g. the
+    // connection was dropped).
+    abortAllInFlight(new ClosedConnectionException("Lost connection to remote peer"));
+    super.channelInactive(ctx);
+  }
+
   private ResponseCallback release(int streamId, ChannelHandlerContext ctx) {
     LOG.debug("[{}] Releasing stream id {}", logPrefix, streamId);
     ResponseCallback responseCallback = inFlight.remove(streamId);
diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/channel/InFlightHandlerTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/channel/InFlightHandlerTest.java
@@ -370,6 +370,33 @@ public void should_fail_all_pending_and_close_on_unexpected_inbound_exception()
     }
   }
 
+  @Test
+  public void should_fail_all_pending_if_connection_lost() {
+    // Given
+    addToPipeline();
+    Mockito.when(streamIds.acquire()).thenReturn(42, 43);
+    MockResponseCallback responseCallback1 = new MockResponseCallback();
+    MockResponseCallback responseCallback2 = new MockResponseCallback();
+    channel
+        .writeAndFlush(
+            new DriverChannel.RequestMessage(QUERY, false, Frame.NO_PAYLOAD, responseCallback1))
+        .awaitUninterruptibly();
+    channel
+        .writeAndFlush(
+            new DriverChannel.RequestMessage(QUERY, false, Frame.NO_PAYLOAD, responseCallback2))
+        .awaitUninterruptibly();
+
+    // When
+    channel.pipeline().fireChannelInactive();
+
+    // Then
+    for (MockResponseCallback callback : ImmutableList.of(responseCallback1, responseCallback2)) {
+      assertThat(callback.getFailure())
+          .isInstanceOf(ClosedConnectionException.class)
+          .hasMessageContaining("Lost connection to remote peer");
+    }
+  }
+
   @Test
   public void should_hold_stream_id_if_required() {
     // Given
