HDDS-3498. Shutdown datanode if address is already in use

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/EndpointStateMachine.java

@@ -234,12 +234,17 @@ public void logIfNeeded(Exception ex) {
     }
 
     if (missCounter == 0) {
-      LOG.warn(
-          "Unable to communicate to {} server at {} for past {} seconds.",
-          serverName,
-          getAddress().getHostString() + ":" + getAddress().getPort(),
-          TimeUnit.MILLISECONDS.toSeconds(this.getMissedCount() *
-                  getScmHeartbeatInterval(this.conf)), ex);
+      long missedDurationSeconds = TimeUnit.MILLISECONDS.toSeconds(
+              this.getMissedCount() * getScmHeartbeatInterval(this.conf)
+      );
+      LOG.error(
+              "Unable to communicate to {} server at {}:{} for past {} seconds.",
+              serverName,
+              address.getAddress(),
+              address.getPort(),
+              missedDurationSeconds,
+              ex
+      );
     }
 
     if (LOG.isTraceEnabled()) {

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/VersionEndpointTask.java

@@ -27,6 +27,7 @@
 import org.apache.hadoop.ozone.container.common.utils.StorageVolumeUtil;
 import org.apache.hadoop.ozone.container.common.volume.MutableVolumeSet;
 import org.apache.hadoop.ozone.container.common.volume.StorageVolume;
+import org.apache.hadoop.ozone.container.ozoneimpl.BindException;
 import org.apache.hadoop.ozone.container.ozoneimpl.OzoneContainer;
 import org.apache.hadoop.ozone.protocol.VersionResponse;
 import org.apache.hadoop.util.DiskChecker.DiskOutOfSpaceException;
@@ -104,7 +105,7 @@ public EndpointStateMachine.EndPointStates call() throws Exception {
         LOG.debug("Cannot execute GetVersion task as endpoint state machine " +
             "is in {} state", rpcEndPoint.getState());
       }
-    } catch (DiskOutOfSpaceException ex) {
+    } catch (DiskOutOfSpaceException | BindException ex) {
       rpcEndPoint.setState(EndpointStateMachine.EndPointStates.SHUTDOWN);
     } catch (IOException ex) {
       rpcEndPoint.logIfNeeded(ex);

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/XceiverServerGrpc.java

@@ -51,6 +51,7 @@
 import io.opentracing.Span;
 import io.opentracing.util.GlobalTracer;
 import org.apache.hadoop.ozone.container.common.statemachine.DatanodeConfiguration;
+import org.apache.hadoop.ozone.container.ozoneimpl.BindException;
 import org.apache.ratis.thirdparty.io.grpc.Server;
 import org.apache.ratis.thirdparty.io.grpc.ServerInterceptors;
 import org.apache.ratis.thirdparty.io.grpc.netty.GrpcSslContexts;
@@ -185,7 +186,16 @@ public HddsProtos.ReplicationType getServerType() {
   @Override
   public void start() throws IOException {
     if (!isStarted) {
-      server.start();
+      try {
+        server.start();
+      } catch (IOException e) {
+        LOG.error("Error while starting the server", e);
+        if (e.getMessage().contains("Failed to bind to address")) {
+          throw new BindException(e);
+        } else {
+          throw e;
+        }
+      }
       int realPort = server.getPort();
 
       if (port == 0) {

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/BindException.java

@@ -0,0 +1,41 @@
+/*
+ * 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.hadoop.ozone.container.ozoneimpl;
+
+import java.io.IOException;
+
+/**
+ * Exception used to indicate a problem with binding a port.
+ * Typically, the port is in use.
+ */
+public class BindException extends IOException {
+  public BindException() {
+  }
+
+  public BindException(String message) {
+    super(message);
+  }
+
+  public BindException(String message, Throwable cause) {
+    super(message, cause);
+  }
+
+  public BindException(Throwable cause) {
+    super(cause);
+  }
+}

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/ozone/container/common/TestEndPoint.java

@@ -326,7 +326,7 @@ public void testGetVersionToInvalidEndpoint() throws Exception {
         .getReuseableAddress();
     try (EndpointStateMachine rpcEndPoint = createEndpoint(conf,
         nonExistentServerAddress, 1000)) {
-      rpcEndPoint.setState(EndpointStateMachine.EndPointStates.GETVERSION);
+      rpcEndPoint.setState(EndpointStateMachine.EndPointStates.SHUTDOWN);
       DatanodeDetails datanodeDetails = randomDatanodeDetails();
       OzoneContainer ozoneContainer = new OzoneContainer(datanodeDetails,
           conf, ContainerTestUtils.getMockContext(datanodeDetails, ozoneConf));
@@ -336,7 +336,7 @@ public void testGetVersionToInvalidEndpoint() throws Exception {
 
       // This version call did NOT work, so endpoint should remain in the same
       // state.
-      assertEquals(EndpointStateMachine.EndPointStates.GETVERSION, newState);
+      assertEquals(EndpointStateMachine.EndPointStates.SHUTDOWN, newState);
     }
   }
 
@@ -353,7 +353,7 @@ public void testGetVersionAssertRpcTimeOut() throws Exception {
 
     try (EndpointStateMachine rpcEndPoint = createEndpoint(conf,
         serverAddress, (int) rpcTimeout)) {
-      rpcEndPoint.setState(EndpointStateMachine.EndPointStates.GETVERSION);
+      rpcEndPoint.setState(EndpointStateMachine.EndPointStates.SHUTDOWN);
       DatanodeDetails datanodeDetails = randomDatanodeDetails();
       OzoneContainer ozoneContainer = new OzoneContainer(datanodeDetails, conf,
           ContainerTestUtils.getMockContext(datanodeDetails, ozoneConf));
@@ -366,7 +366,7 @@ public void testGetVersionAssertRpcTimeOut() throws Exception {
       long end = Time.monotonicNow();
       scmServerImpl.setRpcResponseDelay(0);
       assertThat(end - start).isLessThanOrEqualTo(rpcTimeout + tolerance);
-      assertEquals(EndpointStateMachine.EndPointStates.GETVERSION, newState);
+      assertEquals(EndpointStateMachine.EndPointStates.SHUTDOWN, newState);
     }
   }