netty: prevent interruption during bind from leaking channel

Fixes #6850

core/src/test/java/io/grpc/internal/AbstractTransportTest.java

@@ -400,6 +400,36 @@ public void serverAlreadyListening() throws Exception {
     server2.start(new MockServerListener());
   }
 
+  @Test
+  public void serverStartInterrupted() throws Exception {
+    client = null;
+
+    // Just get free port
+    server.start(serverListener);
+    int port = -1;
+    SocketAddress addr = server.getListenSocketAddress();
+    if (addr instanceof InetSocketAddress) {
+      port = ((InetSocketAddress) addr).getPort();
+    }
+    assumeTrue("transport is not using InetSocketAddress", port != -1);
+    server.shutdown();
+
+    server = Iterables.getOnlyElement(newServer(port, Arrays.asList(serverStreamTracerFactory)));
+    boolean success;
+    Thread.currentThread().interrupt();
+    try {
+      server.start(serverListener);
+      success = true;
+    } catch (Exception ex) {
+      success = false;
+    } finally {
+      Thread.interrupted(); // clear interruption
+    }
+    assumeTrue("apparently start is not impacted by interruption, so nothing to test", !success);
+    // second time should not throw, as the first time should not have bound to the port
+    server.start(serverListener);
+  }
+
   @Test
   public void openStreamPreventsTermination() throws Exception {
     server.start(serverListener);

netty/src/main/java/io/grpc/netty/NettyServer.java

@@ -242,12 +242,10 @@ public void operationComplete(ChannelFuture future) throws Exception {
     });
     // Bind and start to accept incoming connections.
     ChannelFuture future = b.bind(address);
-    try {
-      future.await();
-    } catch (InterruptedException ex) {
-      Thread.currentThread().interrupt();
-      throw new RuntimeException("Interrupted waiting for bind");
-    }
+    // We'd love to observe interruption, but if interrupted we will need to close the channel,
+    // which itself would need an await() to guarantee the port is not used when the method returns.
+    // See #6850
+    future.awaitUninterruptibly();
     if (!future.isSuccess()) {
       throw new IOException("Failed to bind", future.cause());
     }