HADOOP-18024. SocketChannel is not closed when IOException happens in Server$Listener.doAccept

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java

@@ -1325,7 +1325,7 @@ public String toString() {
   }
 
   /** Listens on the socket. Creates jobs for the handler threads*/
-  private class Listener extends Thread {
+  protected class Listener extends Thread {
 
     private ServerSocketChannel acceptChannel = null; //the accept channel
     private Selector selector = null; //the selector that we use for the server
@@ -1530,15 +1530,30 @@ private void closeCurrentConnection(SelectionKey key, Throwable e) {
     InetSocketAddress getAddress() {
       return (InetSocketAddress)acceptChannel.socket().getLocalSocketAddress();
     }
-
+
+    protected void configureSocketChannel(SocketChannel channel) throws IOException {
+      channel.configureBlocking(false);
+      channel.socket().setTcpNoDelay(tcpNoDelay);
+      channel.socket().setKeepAlive(true);
+    }
+
     void doAccept(SelectionKey key) throws InterruptedException, IOException,  OutOfMemoryError {
       ServerSocketChannel server = (ServerSocketChannel) key.channel();
       SocketChannel channel;
       while ((channel = server.accept()) != null) {
 
-        channel.configureBlocking(false);
-        channel.socket().setTcpNoDelay(tcpNoDelay);
-        channel.socket().setKeepAlive(true);
+        try {
+          configureSocketChannel(channel);
+        } catch (IOException e) {
+          LOG.warn("Error in an accepted SocketChannel", e);
+          try {
+            channel.socket().close();
+            channel.close();
+          } catch (IOException ex) {
+            LOG.warn("Error in closing SocketChannel", ex);
+          }
+          continue;
+        }
 
         Reader reader = getReader();
         Connection c = connectionManager.register(channel,
@@ -3249,7 +3264,7 @@ protected Server(String bindAddress, int port,
     this.negotiateResponse = buildNegotiateResponse(enabledAuthMethods);
 
     // Start the listener here and let it bind to the port
-    listener = new Listener(port);
+    listener = createListener(port);
     // set the server port to the default listener port.
     this.port = listener.getAddress().getPort();
     connectionManager = new ConnectionManager();
@@ -3280,6 +3295,10 @@ protected Server(String bindAddress, int port,
         HealthCheckFailedException.class);
   }
 
+  protected Listener createListener(int port) throws IOException {
+    return new Listener(port);
+  }
+
   public synchronized void addAuxiliaryListener(int auxiliaryPort)
       throws IOException {
     if (auxiliaryListenerMap == null) {
@@ -3289,7 +3308,7 @@ public synchronized void addAuxiliaryListener(int auxiliaryPort)
       throw new IOException(
           "There is already a listener binding to: " + auxiliaryPort);
     }
-    Listener newListener = new Listener(auxiliaryPort);
+    Listener newListener = createListener(auxiliaryPort);
     newListener.setIsAuxiliary();
 
     // in the case of port = 0, the listener would be on a != 0 port.

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPC.java

@@ -34,6 +34,7 @@
 import java.io.ByteArrayOutputStream;
 import java.io.DataInput;
 import java.io.DataOutput;
+import java.io.EOFException;
 import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;
@@ -48,6 +49,7 @@
 import java.net.SocketException;
 import java.net.SocketTimeoutException;
 import java.net.UnknownHostException;
+import java.nio.channels.SocketChannel;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashSet;
@@ -616,6 +618,55 @@ public void testIOEOnWriteAfterPingClient() throws Exception {
       WRITABLE_FAULTS_SLEEP = 0;
     }
   }
+
+  /**
+   * Test for HADOOP-18024
+   */
+  @Test(timeout=60000)
+  public void testIOEOnListenerAccept() throws Exception {
+    // start server
+    Server server = new TestServer(1, false,
+            LongWritable.class, LongWritable.class) {
+      @Override
+      protected Listener createListener(int port) throws IOException {
+        return new Listener(port) {
+          @Override
+          protected void configureSocketChannel(SocketChannel channel) throws IOException {
+            maybeThrowIOE();
+            super.configureSocketChannel(channel);
+          }
+        };
+      }
+    };
+    InetSocketAddress addr = NetUtils.getConnectAddress(server);
+    server.start();
+
+    // start client
+    WRITABLE_FAULTS_ENABLED = true;
+    Client client = new Client(LongWritable.class, conf);
+    try {
+      LongWritable param = LongWritable.class.newInstance();
+
+      try {
+        call(client, param, addr, 0, conf);
+        fail("Expected an exception to have been thrown");
+      } catch (EOFException e) {
+        LOG.info("Got expected exception", e);
+      } catch (Throwable t) {
+        fail("Expected an EOFException to have been thrown");
+      }
+
+      // Doing a second call with faults disabled should return fine --
+      // ie the internal state of the client or server should not be broken
+      // by the failed call
+      WRITABLE_FAULTS_ENABLED = false;
+      call(client, param, addr, 0, conf);
+
+    } finally {
+      client.stop();
+      server.stop();
+    }
+  }
 
   private static void assertExceptionContains(
       Throwable t, String substring) {