Remove Throwable usage from transport modules (#30845)

Currently nio and netty modules use the CompletableFuture class for
managing listeners. This is unfortunate as that class accepts
Throwable. This commit adds a class CompletableContext that wraps
the CompletableFuture but does not accept Throwable. This allows the
modification of netty and nio logic to no longer handle Throwable.

Author: Tim-Brooks

libs/elasticsearch-core/src/main/java/org/elasticsearch/common/concurrent/CompletableContext.java

@@ -0,0 +1,63 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.common.concurrent;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.function.BiConsumer;
+
+/**
+ * A thread-safe completable context that allows listeners to be attached. This class relies on the
+ * {@link CompletableFuture} for the concurrency logic. However, it does not accept {@link Throwable} as
+ * an exceptional result. This allows attaching listeners that only handle {@link Exception}.
+ *
+ * @param <T> the result type
+ */
+public class CompletableContext<T> {
+
+    private final CompletableFuture<T> completableFuture = new CompletableFuture<>();
+
+    public void addListener(BiConsumer<T, ? super Exception> listener) {
+        BiConsumer<T, Throwable> castThrowable = (v, t) -> {
+            if (t == null) {
+                listener.accept(v, null);
+            } else {
+                assert !(t instanceof Error) : "Cannot be error";
+                listener.accept(v, (Exception) t);
+            }
+        };
+        completableFuture.whenComplete(castThrowable);
+    }
+
+    public boolean isDone() {
+        return completableFuture.isDone();
+    }
+
+    public boolean isCompletedExceptionally() {
+        return completableFuture.isCompletedExceptionally();
+    }
+
+    public boolean completeExceptionally(Exception ex) {
+        return completableFuture.completeExceptionally(ex);
+    }
+
+    public boolean complete(T value) {
+        return completableFuture.complete(value);
+    }
+}

libs/elasticsearch-nio/build.gradle

@@ -33,6 +33,8 @@ publishing {
 }
 
 dependencies {
+  compile "org.elasticsearch:elasticsearch-core:${version}"
+
   testCompile "com.carrotsearch.randomizedtesting:randomizedtesting-runner:${versions.randomizedrunner}"
   testCompile "junit:junit:${versions.junit}"
   testCompile "org.hamcrest:hamcrest-all:${versions.hamcrest}"

libs/elasticsearch-nio/src/main/java/org/elasticsearch/nio/BytesWriteHandler.java

@@ -28,7 +28,7 @@ public abstract class BytesWriteHandler implements ReadWriteHandler {
 
     private static final List<FlushOperation> EMPTY_LIST = Collections.emptyList();
 
-    public WriteOperation createWriteOperation(SocketChannelContext context, Object message, BiConsumer<Void, Throwable> listener) {
+    public WriteOperation createWriteOperation(SocketChannelContext context, Object message, BiConsumer<Void, Exception> listener) {
         assert message instanceof ByteBuffer[] : "This channel only supports messages that are of type: " + ByteBuffer[].class
             + ". Found type: " + message.getClass() + ".";
         return new FlushReadyWrite(context, (ByteBuffer[]) message, listener);

libs/elasticsearch-nio/src/main/java/org/elasticsearch/nio/ChannelContext.java

@@ -19,11 +19,12 @@
 
 package org.elasticsearch.nio;
 
+import org.elasticsearch.common.concurrent.CompletableContext;
+
 import java.io.IOException;
 import java.nio.channels.NetworkChannel;
 import java.nio.channels.SelectableChannel;
 import java.nio.channels.SelectionKey;
-import java.util.concurrent.CompletableFuture;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 
@@ -37,7 +38,7 @@ public abstract class ChannelContext<S extends SelectableChannel & NetworkChanne
 
     protected final S rawChannel;
     private final Consumer<Exception> exceptionHandler;
-    private final CompletableFuture<Void> closeContext = new CompletableFuture<>();
+    private final CompletableContext<Void> closeContext = new CompletableContext<>();
     private volatile SelectionKey selectionKey;
 
     ChannelContext(S rawChannel, Consumer<Exception> exceptionHandler) {
@@ -81,8 +82,8 @@ public void closeFromSelector() throws IOException {
      *
      * @param listener to be called
      */
-    public void addCloseListener(BiConsumer<Void, Throwable> listener) {
-        closeContext.whenComplete(listener);
+    public void addCloseListener(BiConsumer<Void, Exception> listener) {
+        closeContext.addListener(listener);
     }
 
     public boolean isOpen() {

libs/elasticsearch-nio/src/main/java/org/elasticsearch/nio/FlushOperation.java

@@ -25,13 +25,13 @@
 
 public class FlushOperation {
 
-    private final BiConsumer<Void, Throwable> listener;
+    private final BiConsumer<Void, Exception> listener;
     private final ByteBuffer[] buffers;
     private final int[] offsets;
     private final int length;
     private int internalIndex;
 
-    public FlushOperation(ByteBuffer[] buffers, BiConsumer<Void, Throwable> listener) {
+    public FlushOperation(ByteBuffer[] buffers, BiConsumer<Void, Exception> listener) {
         this.listener = listener;
         this.buffers = buffers;
         this.offsets = new int[buffers.length];
@@ -44,7 +44,7 @@ public FlushOperation(ByteBuffer[] buffers, BiConsumer<Void, Throwable> listener
         length = offset;
     }
 
-    public BiConsumer<Void, Throwable> getListener() {
+    public BiConsumer<Void, Exception> getListener() {
         return listener;
     }
 

libs/elasticsearch-nio/src/main/java/org/elasticsearch/nio/FlushReadyWrite.java

@@ -27,7 +27,7 @@ public class FlushReadyWrite extends FlushOperation implements WriteOperation {
     private final SocketChannelContext channelContext;
     private final ByteBuffer[] buffers;
 
-    FlushReadyWrite(SocketChannelContext channelContext, ByteBuffer[] buffers, BiConsumer<Void, Throwable> listener) {
+    FlushReadyWrite(SocketChannelContext channelContext, ByteBuffer[] buffers, BiConsumer<Void, Exception> listener) {
         super(buffers, listener);
         this.channelContext = channelContext;
         this.buffers = buffers;

libs/elasticsearch-nio/src/main/java/org/elasticsearch/nio/NioChannel.java

@@ -53,7 +53,7 @@ public InetSocketAddress getLocalAddress() {
      *
      * @param listener to be called at close
      */
-    public void addCloseListener(BiConsumer<Void, Throwable> listener) {
+    public void addCloseListener(BiConsumer<Void, Exception> listener) {
         getContext().addCloseListener(listener);
     }
 

libs/elasticsearch-nio/src/main/java/org/elasticsearch/nio/NioSocketChannel.java

@@ -60,7 +60,7 @@ public InetSocketAddress getRemoteAddress() {
         return remoteAddress;
     }
 
-    public void addConnectListener(BiConsumer<Void, Throwable> listener) {
+    public void addConnectListener(BiConsumer<Void, Exception> listener) {
         context.addConnectListener(listener);
     }
 

libs/elasticsearch-nio/src/main/java/org/elasticsearch/nio/ReadWriteHandler.java

@@ -38,7 +38,7 @@ public interface ReadWriteHandler {
      * @param listener the listener to be called when the message is sent
      * @return the write operation to be queued
      */
-    WriteOperation createWriteOperation(SocketChannelContext context, Object message, BiConsumer<Void, Throwable> listener);
+    WriteOperation createWriteOperation(SocketChannelContext context, Object message, BiConsumer<Void, Exception> listener);
 
     /**
      * This method is called on the event loop thread. It should serialize a write operation object to bytes

libs/elasticsearch-nio/src/main/java/org/elasticsearch/nio/SocketChannelContext.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.nio;
 
+import org.elasticsearch.common.concurrent.CompletableContext;
 import org.elasticsearch.nio.utils.ExceptionsHelper;
 
 import java.io.IOException;
@@ -27,7 +28,6 @@
 import java.nio.channels.SocketChannel;
 import java.util.ArrayList;
 import java.util.LinkedList;
-import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
@@ -48,7 +48,7 @@ public abstract class SocketChannelContext extends ChannelContext<SocketChannel>
     protected final AtomicBoolean isClosing = new AtomicBoolean(false);
     private final ReadWriteHandler readWriteHandler;
     private final SocketSelector selector;
-    private final CompletableFuture<Void> connectContext = new CompletableFuture<>();
+    private final CompletableContext<Void> connectContext = new CompletableContext<>();
     private final LinkedList<FlushOperation> pendingFlushes = new LinkedList<>();
     private boolean ioException;
     private boolean peerClosed;
@@ -73,8 +73,8 @@ public NioSocketChannel getChannel() {
         return channel;
     }
 
-    public void addConnectListener(BiConsumer<Void, Throwable> listener) {
-        connectContext.whenComplete(listener);
+    public void addConnectListener(BiConsumer<Void, Exception> listener) {
+        connectContext.addListener(listener);
     }
 
     public boolean isConnectComplete() {
@@ -121,7 +121,7 @@ public boolean connect() throws IOException {
         return isConnected;
     }
 
-    public void sendMessage(Object message, BiConsumer<Void, Throwable> listener) {
+    public void sendMessage(Object message, BiConsumer<Void, Exception> listener) {
         if (isClosing.get()) {
             listener.accept(null, new ClosedChannelException());
             return;