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

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

@@ -39,6 +39,13 @@ public final class HddsConfigKeys {
       "hdds.heartbeat.recon.initial-interval";
   public static final String HDDS_RECON_INITIAL_HEARTBEAT_INTERVAL_DEFAULT =
       "2s";
+  /**
+   * Number of consecutive heartbeat failures the DN tolerates against the same SCM endpoint before
+   * re-resolving its hostname. Only consulted when ozone.client.failover.resolve-needed is true.
+   */
+  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/ha/SCMNodeInfo.java

@@ -229,22 +229,22 @@ public String getNodeId() {
   }
 
   public String getBlockClientAddress() {
-    return blockClientAddress.getHostAndPortString();
+    return blockClientAddress == null ? null : blockClientAddress.getHostAndPortString();
   }
 
   public String getScmClientAddress() {
-    return scmClientAddress.getHostAndPortString();
+    return scmClientAddress == null ? null : scmClientAddress.getHostAndPortString();
   }
 
   public String getScmSecurityAddress() {
-    return scmSecurityAddress.getHostAndPortString();
+    return scmSecurityAddress == null ? null : scmSecurityAddress.getHostAndPortString();
   }
 
   public HostAndPort getScmDatanodeHostPortAddress() {
     return scmDatanodeAddress;
   }
 
   public String getScmDatanodeAddress() {
-    return scmDatanodeAddress.getHostAndPortString();
+    return scmDatanodeAddress == null ? null : scmDatanodeAddress.getHostAndPortString();
   }
 }

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

@@ -17,6 +17,7 @@
 
 package org.apache.hadoop.hdds.scm.net;
 
+import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import org.apache.hadoop.net.NetUtils;
 
@@ -44,6 +45,12 @@ public HostAndPort(String host, int port) {
     this(host, port, null);
   }
 
+  /**
+   * Derives the host from {@code address.getHostName()}, which can differ from the configured
+   * spelling used by {@link #HostAndPort(String, int)} (IP literal vs hostname). Since equals/hashCode
+   * key on the host string, callers must use one consistent spelling per node or instances will not
+   * match in the endpoint maps.
+   */
   public HostAndPort(InetSocketAddress address) {
     this(address.getHostName(), address.getPort(), address);
   }
@@ -64,6 +71,26 @@ public synchronized InetSocketAddress getAddress() {
     return address;
   }
 
+  /**
+   * Re-resolves the host:port and swaps the cached address if the IP changed. The DNS lookup runs
+   * outside the monitor; only the swap is synchronized, mirroring {@link #getAddress()}.
+   * @return true if the cached address changed
+   */
+  public boolean refresh() {
+    final InetSocketAddress latest = NetUtils.createSocketAddr(hostAndPortString);
+    final InetAddress latestIp = latest.getAddress();
+    if (latestIp == null) {
+      return false;
+    }
+    synchronized (this) {
+      if (latestIp.equals(address.getAddress())) {
+        return false;
+      }
+      address = latest;
+      return true;
+    }
+  }
+
   @Override
   public int hashCode() {
     return hash;

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

@@ -3964,6 +3964,19 @@
     </description>
   </property>
 
+  <property>
+    <name>hdds.heartbeat.address.refresh.threshold</name>
+    <value>3</value>
+    <tag>OZONE, DATANODE, HA</tag>
+    <description>Number of consecutive heartbeat failures the DataNode
+      tolerates against the same SCM endpoint before re-resolving its
+      hostname. Only consulted when ozone.client.failover.resolve-needed
+      is true. Conservative default avoids re-resolution on transient
+      blips while still recovering from a peer pod IP change within
+      seconds.
+    </description>
+  </property>
+
   <property>
     <name>ozone.directory.deleting.service.interval</name>
     <value>1m</value>

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

@@ -139,40 +139,81 @@ 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, threadNamePrefix);
       endPoint.setPassive(false);
       scmMachines.put(address, endPoint);
     } finally {
       writeUnlock();
     }
   }
 
+  /**
+   * Builds (but does not register) an active-SCM endpoint dialing {@code address.getAddress()}.
+   * Factored out so {@link #refreshSCMServer} can rebuild an endpoint on a freshly-resolved IP.
+   */
+  private EndpointStateMachine buildScmEndpoint(HostAndPort address, 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,
+        address.getAddress(), 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 hostname of the active SCM endpoint at {@code address} and, if the IP changed,
+   * rebuilds the endpoint in place under the same key. Because the key is the stable host:port, no
+   * re-keying or StateContext queue migration is needed -- only the {@link EndpointStateMachine}
+   * value (which holds the frozen RPC proxy) is replaced.
+   *
+   * @return true if the endpoint was rebuilt on a new address
+   */
+  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();
+    }
+    // DNS lookup runs outside the map lock; HostAndPort swaps its cached address only on a change.
+    if (!address.refresh()) {
+      return false;
+    }
+    final EndpointStateMachine stale;
+    writeLock();
+    try {
+      if (scmMachines.get(address) != current) {
+        // Lost the race: another refresh or removeSCMServer replaced this entry. Abandon.
+        return false;
+      }
+      EndpointStateMachine rebuilt = buildScmEndpoint(address, threadNamePrefix);
+      rebuilt.setPassive(false);
+      scmMachines.put(address, rebuilt);
+      stale = current;
+    } finally {
+      writeUnlock();
+    }
+    // Close the stale proxy outside the lock; teardown can block on socket close.
+    stale.close();
+    LOG.info("Re-resolved SCM endpoint {} after connection failure.", address);
+    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,34 @@ 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, once the missed-count threshold is reached and DNS
+   * refresh is enabled, ask the connection manager to re-resolve this SCM peer and rebuild the
+   * endpoint if its IP changed. Off by default (ozone.client.failover.resolve-needed).
+   */
+  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) {