Unify http channels and exception handling

libs/core/src/main/java/org/elasticsearch/core/internal/io/IOUtils.java

@@ -112,6 +112,41 @@ public static void close(final Exception ex, final Iterable<? extends Closeable>
  }
  }
 
+ /**
+ * Closes all given {@link Closeable}s. Some of the {@linkplain Closeable}s may be null; they are
+ * ignored. After everything is closed, the method either throws the first {@link RuntimeException} it
+ * hit while closing with other exceptions added as suppressed, or completes normally if there were
+ * no exceptions. If the first exception is not a {@linkplain RuntimeException}, it is wrapped in a
+ * {@linkplain RuntimeException}.
+ *
+ * @param objects objects to close
+ */
+ public static void closeAndConvertToRuntimeExceptions(final Iterable<? extends Closeable> objects) {
+ RuntimeException closingException = null;
+ for (final Closeable object : objects) {
+ try {
+ if (object != null) {
+ object.close();
+ }
+ } catch (RuntimeException e) {
+ if (closingException == null) {
+ closingException = e;
+ } else {
+ closingException.addSuppressed(e);
+ }
+ } catch (IOException e) {
+ if (closingException == null) {
+ closingException = new RuntimeException(e);
+ } {
+ closingException.addSuppressed(e);
+ }
+ }
+ }
+ if (closingException != null) {
+ throw closingException;
+ }
+ }
+
  /**
  * Closes all given {@link Closeable}s, suppressing all thrown exceptions. Some of the {@link Closeable}s may be null, they are ignored.
  *

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpServerTransport.java

@@ -318,8 +318,9 @@ protected void doStop() {
  Netty4Utils.closeChannels(serverChannels);
  } catch (IOException e) {
  logger.trace("exception while closing channels", e);
+ } finally {
+ serverChannels.clear();
  }
- serverChannels.clear();
  }
  }
 
@@ -345,8 +346,7 @@ protected void doClose() {
 
  @Override
  public HttpStats stats() {
- int serverChannelCount = serverChannels.size();
- return new HttpStats(serverChannelCount, httpChannels.size() + serverChannelCount);
+ return new HttpStats(httpChannels.size(), totalChannelsAccepted.get());
  }
 
  @Override

plugins/transport-nio/src/main/java/org/elasticsearch/http/nio/NioHttpServerTransport.java

@@ -234,8 +234,7 @@ protected TransportAddress bindAddress(InetAddress hostAddress) {
 
  @Override
  public HttpStats stats() {
- int serverChannelCount = serverChannels.size();
- return new HttpStats(serverChannels.size(), httpChannels.size() + serverChannelCount);
+ return new HttpStats(serverChannels.size(), totalChannelsAccepted.get());
  }
 
  static NioCorsConfig buildCorsConfig(Settings settings) {

server/src/main/java/org/elasticsearch/common/network/CloseableChannel.java

@@ -69,7 +69,7 @@ static <C extends CloseableChannel> void closeChannel(C channel) {
  /**
  * Closes the channel.
  *
- * @param channel to close
+ * @param channel  to close
  * @param blocking indicates if we should block on channel close
  */
  static <C extends CloseableChannel> void closeChannel(C channel, boolean blocking) {
@@ -83,19 +83,15 @@ static <C extends CloseableChannel> void closeChannel(C channel, boolean blockin
  * @param blocking indicates if we should block on channel close
  */
  static <C extends CloseableChannel> void closeChannels(List<C> channels, boolean blocking) {
+ IOUtils.closeAndConvertToRuntimeExceptions(channels);
  if (blocking) {
  ArrayList<ActionFuture<Void>> futures = new ArrayList<>(channels.size());
  for (final C channel : channels) {
- if (channel.isOpen()) {
- PlainActionFuture<Void> closeFuture = PlainActionFuture.newFuture();
- channel.addCloseListener(closeFuture);
- channel.close();
- futures.add(closeFuture);
- }
+ PlainActionFuture<Void> closeFuture = PlainActionFuture.newFuture();
+ channel.addCloseListener(closeFuture);
+ futures.add(closeFuture);
  }
  blockOnFutures(futures);
- } else {
- IOUtils.closeWhileHandlingException(channels);
  }
  }
 
@@ -108,7 +104,6 @@ static void blockOnFutures(List<ActionFuture<Void>> futures) {
  // close exceptions happens elsewhere.
  } catch (InterruptedException e) {
  Thread.currentThread().interrupt();
- throw new IllegalStateException("Future got interrupted", e);
  }
  }
  }

server/src/main/java/org/elasticsearch/http/AbstractHttpServerTransport.java

@@ -23,6 +23,7 @@
 import com.carrotsearch.hppc.IntSet;
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.elasticsearch.ExceptionsHelper;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.component.AbstractLifecycleComponent;
 import org.elasticsearch.common.network.CloseableChannel;
@@ -51,6 +52,7 @@
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
 
 import static org.elasticsearch.http.HttpTransportSettings.SETTING_HTTP_BIND_HOST;
 import static org.elasticsearch.http.HttpTransportSettings.SETTING_HTTP_MAX_CONTENT_LENGTH;
@@ -72,6 +74,7 @@ public abstract class AbstractHttpServerTransport extends AbstractLifecycleCompo
  private final String[] bindHosts;
  private final String[] publishHosts;
 
+ protected final AtomicLong totalChannelsAccepted = new AtomicLong();
  protected final Set<HttpChannel> httpChannels = Collections.newSetFromMap(new ConcurrentHashMap<>());
  protected volatile BoundTransportAddress boundAddress;
 
@@ -181,7 +184,7 @@ protected void onException(HttpChannel channel, Exception e) {
  return;
  }
  if (NetworkExceptionHelper.isCloseConnectionException(e)) {
- logger.warn(() -> new ParameterizedMessage(
+ logger.trace(() -> new ParameterizedMessage(
  "close connection exception caught while handling client http traffic, closing connection {}", channel), e);
  CloseableChannel.closeChannel(channel);
  } else if (NetworkExceptionHelper.isConnectException(e)) {
@@ -193,7 +196,7 @@ protected void onException(HttpChannel channel, Exception e) {
  "cancelled key exception caught while handling client http traffic, closing connection {}", channel), e);
  CloseableChannel.closeChannel(channel);
  } else {
- logger.debug(() -> new ParameterizedMessage(
+ logger.warn(() -> new ParameterizedMessage(
  "caught exception while handling client http traffic, closing connection {}", channel), e);
  CloseableChannel.closeChannel(channel);
  }
@@ -210,7 +213,11 @@ protected void onNonChannelException(Exception exception) {
  }
 
  protected void serverAcceptedChannel(HttpChannel httpChannel) {
- httpChannels.add(httpChannel);
+ boolean addedOnThisCall = httpChannels.add(httpChannel);
+ assert addedOnThisCall : "Channel should only be added to http channel set once";
+ totalChannelsAccepted.incrementAndGet();
+ httpChannel.addCloseListener(ActionListener.wrap(() -> httpChannels.remove(httpChannel)));
+ logger.trace(() -> new ParameterizedMessage("Http channel accepted: {}", httpChannel));
  }
 
  /**

server/src/main/java/org/elasticsearch/http/HttpStats.java

@@ -32,9 +32,9 @@ public class HttpStats implements Writeable, ToXContentFragment {
  private final long serverOpen;
  private final long totalOpen;
 
- public HttpStats(long serverOpen, long totalOpen) {
+ public HttpStats(long serverOpen, long totalOpened) {
  this.serverOpen = serverOpen;
- this.totalOpen = totalOpen;
+ this.totalOpen = totalOpened;
  }
 
  public HttpStats(StreamInput in) throws IOException {

server/src/main/java/org/elasticsearch/transport/TcpTransport.java

@@ -1061,7 +1061,7 @@ protected void onNonChannelException(Exception exception) {
 
  protected void serverAcceptedChannel(TcpChannel channel) {
  boolean addedOnThisCall = acceptedChannels.add(channel);
- assert addedOnThisCall : "Channel should only be added to accept channel set once";
+ assert addedOnThisCall : "Channel should only be added to accepted channel set once";
  channel.addCloseListener(ActionListener.wrap(() -> acceptedChannels.remove(channel)));
  logger.trace(() -> new ParameterizedMessage("Tcp transport channel accepted: {}", channel));
  }