Ensure send listener is called on IOException

Currently there is an issue where the send listener is not called in the
nio transport when an exception is throw during channel flush. This
leads to memory leaks. This commit ensures that the listener is called

test/framework/src/main/java/org/elasticsearch/transport/nio/channel/TcpWriteContext.java

@@ -88,7 +88,12 @@ public class TcpWriteContext implements WriteContext {
     }
 
     private void singleFlush(WriteOperation headOp) throws IOException {
-        headOp.flush();
+        try {
+            headOp.flush();
+        } catch (IOException e) {
+            headOp.getListener().onFailure(e);
+            throw e;
+        }
 
         if (headOp.isFullyFlushed()) {
             headOp.getListener().onResponse(channel);

test/framework/src/test/java/org/elasticsearch/transport/nio/channel/TcpWriteContextTests.java

@@ -30,7 +30,6 @@ import org.mockito.ArgumentCaptor;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.ClosedChannelException;
-import java.nio.channels.SocketChannel;
 
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.mock;
@@ -186,6 +185,23 @@ public class TcpWriteContextTests extends ESTestCase {
         assertFalse(writeContext.hasQueuedWriteOps());
     }
 
+    public void testWhenIOExceptionThrownListenerIsCalled() throws IOException {
+        assertFalse(writeContext.hasQueuedWriteOps());
+
+        WriteOperation writeOperation = mock(WriteOperation.class);
+        writeContext.queueWriteOperations(writeOperation);
+
+        assertTrue(writeContext.hasQueuedWriteOps());
+
+        IOException exception = new IOException();
+        when(writeOperation.flush()).thenThrow(exception);
+        when(writeOperation.getListener()).thenReturn(listener);
+        expectThrows(IOException.class, () -> writeContext.flushChannel());
+
+        verify(listener).onFailure(exception);
+        assertFalse(writeContext.hasQueuedWriteOps());
+    }
+
     private byte[] generateBytes(int n) {
         n += 10;
         byte[] bytes = new byte[n];