ZOOKEEPER-1441: JAVA 11 - Some test cases are failing because Port bind issue.

Fixes the Java 11 build issue.

**Issue**

`NIOServerCnxnFactory` doesn't properly close the socket when the shutdown() is called before the factory has even started. This is mostly the case in tests which use `QuorumUtil` that creates multiple QuorumPeers when instantiated without starting them and when `startAll()` gets called it shuts down the previous ones.

**Reason**

`Selectors` which have been registered with the socket must be closed in order to properly release the socket. `NIOServerCnxnFactory` registers selectors on instantiation, but only releases them in the thread run() method. So, if factory doesn't get started, it won't release the registered selectors.

This wasn't the issue before Java 11 and probably caused by:
https://www.oracle.com/technetwork/java/javase/11-relnote-issues-5012449.html#JDK-8198562

Also this is not an issue when ZooKeeper is used as a separate process (not embedded), because on shutdown the entire JVM stops anyway.

**Resolution**

I decided to try fixing the issue in the connection factory instead of fixing the tests only, because originally it's a bug in the way factory works. Resolution is to open selectors in lazy way: only when accept and selector thread starts, so they don't need to be released if the thread was not even started.

Author: Andor Molnar <andor@apache.org>

Reviewers: hanm@apache.org

Closes #700 from anmolnar/ZOOKEEPER-1441

zookeeper-server/src/main/java/org/apache/zookeeper/server/NIOServerCnxnFactory.java

@@ -698,11 +698,6 @@ public class NIOServerCnxnFactory extends ServerCnxnFactory {
     public void reconfigure(InetSocketAddress addr) {
         ServerSocketChannel oldSS = ss;        
         try {
-            this.ss = ServerSocketChannel.open();
-            ss.socket().setReuseAddress(true);
-            LOG.info("binding to port " + addr);
-            ss.socket().bind(addr);
-            ss.configureBlocking(false);
             acceptThread.setReconfiguring();
             tryClose(oldSS);
             acceptThread.wakeupSelector();
@@ -713,6 +708,11 @@ public class NIOServerCnxnFactory extends ServerCnxnFactory {
                             e.getMessage());
                 Thread.currentThread().interrupt();
             }
+            this.ss = ServerSocketChannel.open();
+            ss.socket().setReuseAddress(true);
+            LOG.info("binding to port " + addr);
+            ss.socket().bind(addr);
+            ss.configureBlocking(false);
             acceptThread = new AcceptThread(ss, addr, selectorThreads);
             acceptThread.start();
         } catch(IOException e) {
@@ -878,13 +878,21 @@ public class NIOServerCnxnFactory extends ServerCnxnFactory {
         }
 
         if (acceptThread != null) {
-            acceptThread.wakeupSelector();
+            if (acceptThread.isAlive()) {
+                acceptThread.wakeupSelector();
+            } else {
+                acceptThread.closeSelector();
+            }
         }
         if (expirerThread != null) {
             expirerThread.interrupt();
         }
         for (SelectorThread thread : selectorThreads) {
-            thread.wakeupSelector();
+            if (thread.isAlive()) {
+                thread.wakeupSelector();
+            } else {
+                thread.closeSelector();
+            }
         }
         if (workerPool != null) {
             workerPool.stop();

zookeeper-server/src/test/java/org/apache/zookeeper/server/NIOServerCnxnFactoryTest.java

@@ -0,0 +1,68 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF 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.apache.zookeeper.server;
+
+import org.apache.zookeeper.PortAssignment;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.net.SocketException;
+
+public class NIOServerCnxnFactoryTest {
+    private InetSocketAddress listenAddress;
+    private NIOServerCnxnFactory factory;
+
+    @Before
+    public void setUp() throws IOException {
+        listenAddress = new InetSocketAddress(PortAssignment.unique());
+        factory = new NIOServerCnxnFactory();
+        factory.configure(listenAddress, 100);
+    }
+
+    @After
+    public void tearDown() {
+        if (factory != null) {
+            factory.shutdown();
+        }
+    }
+
+    @Test(expected = SocketException.class)
+    public void testStartupWithoutStart_SocketAlreadyBound() throws IOException {
+        ServerSocket ss = new ServerSocket(listenAddress.getPort());
+    }
+
+    @Test(expected = SocketException.class)
+    public void testStartupWithStart_SocketAlreadyBound() throws IOException {
+        factory.start();
+        ServerSocket ss = new ServerSocket(listenAddress.getPort());
+    }
+
+    @Test
+    public void testShutdownWithoutStart_SocketReleased() throws IOException {
+        factory.shutdown();
+        factory = null;
+
+        ServerSocket ss = new ServerSocket(listenAddress.getPort());
+        ss.close();
+    }
+}