HDDS-15533. DNS refresh on heartbeat failure for DN to SCM

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java

@@ -39,6 +39,10 @@ public final class HddsConfigKeys {
       "hdds.heartbeat.recon.initial-interval";
   public static final String HDDS_RECON_INITIAL_HEARTBEAT_INTERVAL_DEFAULT =
       "2s";
+  /** Missed heartbeats against one SCM before the DN re-resolves its hostname (HDDS-15533). */
+  public static final String HDDS_HEARTBEAT_ADDRESS_REFRESH_THRESHOLD =
+      "hdds.heartbeat.address.refresh.threshold";
+  public static final int HDDS_HEARTBEAT_ADDRESS_REFRESH_THRESHOLD_DEFAULT = 3;
   public static final String HDDS_NODE_REPORT_INTERVAL =
       "hdds.node.report.interval";
   public static final String HDDS_NODE_REPORT_INTERVAL_DEFAULT =

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/net/HostAndPort.java

@@ -17,7 +17,9 @@
 
 package org.apache.hadoop.hdds.scm.net;
 
+import java.net.InetAddress;
 import java.net.InetSocketAddress;
+import java.util.Objects;
 import org.apache.hadoop.net.NetUtils;
 
 /**
@@ -30,14 +32,13 @@ public class HostAndPort {
   private final String hostAndPortString;
   private final int hash;
   /** The address can be updated from time to time. */
-  private final InetSocketAddress address;
+  private volatile InetSocketAddress address;
 
   public HostAndPort(String host, int port) {
     this.host = host;
     this.port = port;
     this.hostAndPortString = host + ":" + port;
     this.hash = host.hashCode() ^ Integer.hashCode(port);
-    // TODO: HDDS-15533 change the address resolution logic and make this.address threadsafe.
     this.address = NetUtils.createSocketAddr(hostAndPortString);
   }
 
@@ -57,6 +58,21 @@ public InetSocketAddress getAddress() {
     return address;
   }
 
+  /** Re-resolves host:port and returns the new address if its IP changed, else null. No mutation. */
+  public InetSocketAddress resolveLatest() {
+    final InetSocketAddress latest = NetUtils.createSocketAddr(hostAndPortString);
+    final InetAddress latestIp = latest.getAddress();
+    if (latestIp == null || latestIp.equals(address.getAddress())) {
+      return null;
+    }
+    return latest;
+  }
+
+  /** Commits an address re-resolved via {@link #resolveLatest()}. */
+  public void setAddress(InetSocketAddress newAddress) {
+    this.address = Objects.requireNonNull(newAddress, "newAddress == null");
+  }
+
   @Override
   public int hashCode() {
     return hash;

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -4006,6 +4006,16 @@
     </description>
   </property>
 
+  <property>
+    <name>hdds.heartbeat.address.refresh.threshold</name>
+    <value>3</value>
+    <tag>OZONE, DATANODE, HA</tag>
+    <description>Consecutive heartbeat failures the DataNode tolerates
+      against one SCM endpoint before re-resolving its hostname. Only
+      consulted when ozone.client.failover.resolve-needed is true.
+    </description>
+  </property>
+
   <property>
     <name>ozone.directory.deleting.service.interval</name>
     <value>1m</value>

hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/scm/net/TestHostAndPort.java

@@ -0,0 +1,56 @@
+/*
+ * 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.hdds.scm.net;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for {@link HostAndPort} address refresh (HDDS-15533).
+ */
+public class TestHostAndPort {
+
+  @Test
+  public void resolveLatestReturnsNullWhenIpUnchanged() {
+    HostAndPort address = new HostAndPort("127.0.0.1", 9861);
+    assertNull(address.resolveLatest());
+  }
+
+  @Test
+  public void setAddressRejectsNull() {
+    HostAndPort address = new HostAndPort("127.0.0.1", 9861);
+    assertThrows(NullPointerException.class, () -> address.setAddress(null));
+  }
+
+  @Test
+  public void setAddressDoesNotChangeIdentity() throws Exception {
+    HostAndPort address = new HostAndPort("127.0.0.1", 9861);
+    InetSocketAddress refreshed =
+        new InetSocketAddress(InetAddress.getByAddress(new byte[]{10, 0, 0, 7}), 9861);
+    address.setAddress(refreshed);
+    assertEquals(refreshed, address.getAddress());
+    // equals/hashCode stay keyed on host:port so the instance remains a stable map key.
+    assertEquals(new HostAndPort("127.0.0.1", 9861), address);
+    assertEquals(new HostAndPort("127.0.0.1", 9861).hashCode(), address.hashCode());
+  }
+}

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

@@ -152,10 +152,14 @@ public ExecutorService getExecutorService() {
    */
   @Override
   public void close() {
-    if (endPoint != null) {
-      endPoint.close();
+    try {
+      if (endPoint != null) {
+        endPoint.close();
+      }
+    } finally {
+      // Always release the executor thread, even if closing the RPC proxy throws.
+      executorService.shutdown();
     }
-    executorService.shutdown();
   }
 
   /**

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

@@ -22,8 +22,10 @@
 import static org.apache.hadoop.hdds.utils.HddsServerUtil.getScmRpcRetryInterval;
 import static org.apache.hadoop.hdds.utils.HddsServerUtil.getScmRpcTimeOutInMilliseconds;
 
+import com.google.common.annotations.VisibleForTesting;
 import java.io.Closeable;
 import java.io.IOException;
+import java.net.InetSocketAddress;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
@@ -139,40 +141,82 @@ public void addSCMServer(HostAndPort address,
             "Ignoring the request.");
         return;
       }
-
-      Configuration hadoopConfig =
-          LegacyHadoopConfigurationSource.asHadoopConfiguration(this.conf);
-      RPC.setProtocolEngine(
-          hadoopConfig,
-          StorageContainerDatanodeProtocolPB.class,
-          ProtobufRpcEngine.class);
-      long version =
-          RPC.getProtocolVersion(StorageContainerDatanodeProtocolPB.class);
-
-      RetryPolicy retryPolicy =
-          RetryPolicies.retryUpToMaximumCountWithFixedSleep(
-              getScmRpcRetryCount(conf), getScmRpcRetryInterval(conf),
-              TimeUnit.MILLISECONDS);
-
-      StorageContainerDatanodeProtocolPB rpcProxy = RPC.getProtocolProxy(
-          StorageContainerDatanodeProtocolPB.class, version,
-          address.getAddress(), UserGroupInformation.getCurrentUser(), hadoopConfig,
-          NetUtils.getDefaultSocketFactory(hadoopConfig), getRpcTimeout(),
-          retryPolicy).getProxy();
-
-      StorageContainerDatanodeProtocolClientSideTranslatorPB rpcClient =
-          new StorageContainerDatanodeProtocolClientSideTranslatorPB(
-              rpcProxy);
-
-      EndpointStateMachine endPoint = new EndpointStateMachine(address,
-          rpcClient, this.conf, threadNamePrefix);
+      EndpointStateMachine endPoint = buildScmEndpoint(address, address.getAddress(), threadNamePrefix);
       endPoint.setPassive(false);
       scmMachines.put(address, endPoint);
     } finally {
       writeUnlock();
     }
   }
 
+  @VisibleForTesting
+  EndpointStateMachine buildScmEndpoint(HostAndPort address, InetSocketAddress dialAddress,
+      String threadNamePrefix) throws IOException {
+    Configuration hadoopConfig =
+        LegacyHadoopConfigurationSource.asHadoopConfiguration(this.conf);
+    RPC.setProtocolEngine(hadoopConfig, StorageContainerDatanodeProtocolPB.class,
+        ProtobufRpcEngine.class);
+    long version = RPC.getProtocolVersion(StorageContainerDatanodeProtocolPB.class);
+    RetryPolicy retryPolicy = RetryPolicies.retryUpToMaximumCountWithFixedSleep(
+        getScmRpcRetryCount(conf), getScmRpcRetryInterval(conf), TimeUnit.MILLISECONDS);
+    StorageContainerDatanodeProtocolPB rpcProxy = RPC.getProtocolProxy(
+        StorageContainerDatanodeProtocolPB.class, version,
+        dialAddress, UserGroupInformation.getCurrentUser(), hadoopConfig,
+        NetUtils.getDefaultSocketFactory(hadoopConfig), getRpcTimeout(),
+        retryPolicy).getProxy();
+    StorageContainerDatanodeProtocolClientSideTranslatorPB rpcClient =
+        new StorageContainerDatanodeProtocolClientSideTranslatorPB(rpcProxy);
+    return new EndpointStateMachine(address, rpcClient, this.conf, threadNamePrefix);
+  }
+
+  /**
+   * Re-resolves the active SCM endpoint at {@code address}; on an IP change, rebuilds it under the
+   * same key and closes the stale proxy. Returns true if rebuilt.
+   */
+  public boolean refreshSCMServer(HostAndPort address, String threadNamePrefix)
+      throws IOException {
+    final EndpointStateMachine current;
+    readLock();
+    try {
+      current = scmMachines.get(address);
+      if (current == null || current.isPassive()) {
+        return false;
+      }
+    } finally {
+      readUnlock();
+    }
+    // Resolve outside the lock, but commit the new address only after the replacement is built,
+    // so a build failure or a lost race never leaves the cached address ahead of the live proxy.
+    final InetSocketAddress latest = address.resolveLatest();
+    if (latest == null) {
+      return false;
+    }
+    final EndpointStateMachine stale;
+    final InetSocketAddress previous;
+    writeLock();
+    try {
+      if (scmMachines.get(address) != current) {
+        return false;
+      }
+      EndpointStateMachine rebuilt = buildScmEndpoint(address, latest, threadNamePrefix);
+      rebuilt.setPassive(false);
+      previous = address.getAddress();
+      address.setAddress(latest);
+      scmMachines.put(address, rebuilt);
+      stale = current;
+    } finally {
+      writeUnlock();
+    }
+    // The swap is committed; failing to close the stale proxy is cleanup-only, not a refresh failure.
+    try {
+      stale.close();
+    } catch (RuntimeException e) {
+      LOG.warn("Failed to close stale endpoint for {}", address, e);
+    }
+    LOG.info("SCM endpoint {} re-resolved: {} -> {}", address.getHostAndPortString(), previous, latest);
+    return true;
+  }
+
   /**
    * Adds a new Recon server to the set of endpoints.
    *

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

@@ -19,8 +19,12 @@
 
 import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_CONTAINER_ACTION_MAX_LIMIT;
 import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_CONTAINER_ACTION_MAX_LIMIT_DEFAULT;
+import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_ADDRESS_REFRESH_THRESHOLD;
+import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_ADDRESS_REFRESH_THRESHOLD_DEFAULT;
 import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_PIPELINE_ACTION_MAX_LIMIT;
 import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_PIPELINE_ACTION_MAX_LIMIT_DEFAULT;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_DEFAULT;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY;
 import static org.apache.hadoop.ozone.container.upgrade.UpgradeUtils.toLayoutVersionProto;
 
 import com.google.common.base.Preconditions;
@@ -44,6 +48,7 @@
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMHeartbeatRequestProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMHeartbeatResponseProto;
 import org.apache.hadoop.hdds.upgrade.HDDSLayoutVersionManager;
+import org.apache.hadoop.hdds.utils.ConnectionFailureUtils;
 import org.apache.hadoop.hdfs.util.EnumCounters;
 import org.apache.hadoop.ozone.container.common.helpers.DeletedContainerBlocksSummary;
 import org.apache.hadoop.ozone.container.common.statemachine.EndpointStateMachine;
@@ -77,6 +82,8 @@ public class HeartbeatEndpointTask
   private int maxContainerActionsPerHB;
   private int maxPipelineActionsPerHB;
   private HDDSLayoutVersionManager layoutVersionManager;
+  private final boolean resolveOnFailureEnabled;
+  private final int refreshThreshold;
 
   /**
    * Constructs a SCM heart beat.
@@ -100,6 +107,10 @@ public HeartbeatEndpointTask(EndpointStateMachine rpcEndpoint,
     } else {
       this.layoutVersionManager = context.getParent().getLayoutVersionManager();
     }
+    this.resolveOnFailureEnabled = conf.getBoolean(OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY,
+        OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_DEFAULT);
+    this.refreshThreshold = Math.max(1, conf.getInt(HDDS_HEARTBEAT_ADDRESS_REFRESH_THRESHOLD,
+        HDDS_HEARTBEAT_ADDRESS_REFRESH_THRESHOLD_DEFAULT));
   }
 
   /**
@@ -157,12 +168,33 @@ public EndpointStateMachine.EndPointStates call() throws Exception {
       // put back the reports which failed to be sent
       putBackIncrementalReports(requestBuilder);
       rpcEndpoint.logIfNeeded(ex);
+      maybeRefreshScmAddress(ex);
     } finally {
       rpcEndpoint.unlock();
     }
     return rpcEndpoint.getState();
   }
 
+  /**
+   * On a connection-class heartbeat failure past the threshold (and when resolve-needed is on),
+   * asks the connection manager to re-resolve this SCM peer and rebuild the endpoint.
+   */
+  private void maybeRefreshScmAddress(IOException heartbeatFailure) {
+    if (!resolveOnFailureEnabled
+        || rpcEndpoint.isPassive()
+        || rpcEndpoint.getMissedCount() < refreshThreshold
+        || !ConnectionFailureUtils.isConnectionFailure(heartbeatFailure)) {
+      return;
+    }
+    try {
+      context.getParent().getConnectionManager()
+          .refreshSCMServer(rpcEndpoint.getAddress(), context.getThreadNamePrefix());
+    } catch (IOException ex) {
+      LOG.warn("Failed to refresh SCM address {} after {} missed heartbeats",
+          rpcEndpoint.getAddress(), rpcEndpoint.getMissedCount(), ex);
+    }
+  }
+
   // TODO: Make it generic.
   private void putBackIncrementalReports(
       SCMHeartbeatRequestProto.Builder requestBuilder) {