apache · kerneltime · Jul 23, 2026
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java b/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 =

diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/net/HostAndPort.java b/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;
 
@@ -30,14 +31,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 +57,17 @@ public InetSocketAddress getAddress() {
     return address;
   }
 
+  /** Re-resolves host:port; on an IP change swaps the cached address and returns true. */
+  public boolean refresh() {
+    final InetSocketAddress latest = NetUtils.createSocketAddr(hostAndPortString);
+    final InetAddress latestIp = latest.getAddress();
+    if (latestIp == null || latestIp.equals(address.getAddress())) {
+      return false;
+    }
+    address = latest;
+    return true;
+  }
+
   @Override
   public int hashCode() {
     return hash;

diff --git a/hadoop-hdds/common/src/main/resources/ozone-default.xml b/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>

diff --git a/...main/java/org/apache/hadoop/ozone/container/common/statemachine/SCMConnectionManager.java b/...main/java/org/apache/hadoop/ozone/container/common/statemachine/SCMConnectionManager.java
@@ -139,40 +139,70 @@ 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();
     }
   }
 
+  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 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; refresh() swaps the cached address only on a real IP change.
+    if (!address.refresh()) {
+      return false;
+    }
+    final EndpointStateMachine stale;
+    writeLock();
+    try {
+      if (scmMachines.get(address) != current) {
+        return false;
+      }
+      EndpointStateMachine rebuilt = buildScmEndpoint(address, threadNamePrefix);
+      rebuilt.setPassive(false);
+      scmMachines.put(address, rebuilt);
+      stale = current;
+    } finally {
+      writeUnlock();
+    }
+    stale.close();
+    return true;
+  }
+
   /**
    * Adds a new Recon server to the set of endpoints.
    *

diff --git a/.../java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java b/.../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) {

diff --git a/...he/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTaskDnsRefresh.java b/...he/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTaskDnsRefresh.java
@@ -0,0 +1,133 @@
+/*
+ * 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.common.states.endpoint;
+
+import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_ADDRESS_REFRESH_THRESHOLD;
+import static org.apache.hadoop.hdds.upgrade.HDDSLayoutVersionManager.maxLayoutVersion;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.net.ConnectException;
+import java.util.UUID;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto;
+import org.apache.hadoop.hdds.scm.net.HostAndPort;
+import org.apache.hadoop.hdds.upgrade.HDDSLayoutVersionManager;
+import org.apache.hadoop.hdfs.util.EnumCounters;
+import org.apache.hadoop.ozone.container.common.statemachine.DatanodeStateMachine;
+import org.apache.hadoop.ozone.container.common.statemachine.DatanodeStateMachine.DatanodeStates;
+import org.apache.hadoop.ozone.container.common.statemachine.EndpointStateMachine;
+import org.apache.hadoop.ozone.container.common.statemachine.SCMConnectionManager;
+import org.apache.hadoop.ozone.container.common.statemachine.StateContext;
+import org.apache.hadoop.ozone.protocolPB.StorageContainerDatanodeProtocolClientSideTranslatorPB;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Verifies a connection-class heartbeat failure past the threshold triggers a DNS re-resolution of
+ * the SCM peer (HDDS-15533); flag-off, application errors, and below-threshold do not.
+ */
+public class TestHeartbeatEndpointTaskDnsRefresh {
+
+  private static final HostAndPort SCM = new HostAndPort("test-scm-1", 9861);
+
+  @Test
+  public void connectionFailureAtThresholdTriggersRefresh() throws Exception {
+    OzoneConfiguration conf = new OzoneConfiguration();
+    conf.setBoolean(OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY, true);
+    conf.setInt(HDDS_HEARTBEAT_ADDRESS_REFRESH_THRESHOLD, 2);
+    SCMConnectionManager cm = runHeartbeat(conf, 3, new ConnectException("refused"));
+    verify(cm, times(1)).refreshSCMServer(eq(SCM), any());
+  }
+
+  @Test
+  public void flagOffSuppressesRefresh() throws Exception {
+    OzoneConfiguration conf = new OzoneConfiguration();
+    conf.setBoolean(OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY, false);
+    SCMConnectionManager cm = runHeartbeat(conf, 5, new ConnectException("refused"));
+    verify(cm, never()).refreshSCMServer(any(), any());
+  }
+
+  @Test
+  public void applicationErrorDoesNotTriggerRefresh() throws Exception {
+    OzoneConfiguration conf = new OzoneConfiguration();
+    conf.setBoolean(OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY, true);
+    SCMConnectionManager cm = runHeartbeat(conf, 5, new IOException("application-level"));
+    verify(cm, never()).refreshSCMServer(any(), any());
+  }
+
+  @Test
+  public void belowThresholdDoesNotTriggerRefresh() throws Exception {
+    OzoneConfiguration conf = new OzoneConfiguration();
+    conf.setBoolean(OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY, true);
+    conf.setInt(HDDS_HEARTBEAT_ADDRESS_REFRESH_THRESHOLD, 5);
+    SCMConnectionManager cm = runHeartbeat(conf, 1, new ConnectException("refused"));
+    verify(cm, never()).refreshSCMServer(any(), any());
+  }
+
+  /**
+   * Drives one heartbeat that fails with {@code failure}, with the endpoint reporting
+   * {@code missedCount} missed heartbeats, and returns the mocked connection manager to verify.
+   */
+  private SCMConnectionManager runHeartbeat(OzoneConfiguration conf, long missedCount,
+      IOException failure) throws Exception {
+    StorageContainerDatanodeProtocolClientSideTranslatorPB proxy =
+        mock(StorageContainerDatanodeProtocolClientSideTranslatorPB.class);
+    when(proxy.sendHeartbeat(any())).thenThrow(failure);
+
+    EndpointStateMachine endpoint = mock(EndpointStateMachine.class);
+    when(endpoint.getEndPoint()).thenReturn(proxy);
+    when(endpoint.getAddress()).thenReturn(SCM);
+    when(endpoint.getMissedCount()).thenReturn(missedCount);
+    when(endpoint.isPassive()).thenReturn(false);
+
+    SCMConnectionManager connectionManager = mock(SCMConnectionManager.class);
+    DatanodeStateMachine dsm = mock(DatanodeStateMachine.class);
+    when(dsm.getConnectionManager()).thenReturn(connectionManager);
+    when(dsm.getQueuedCommandCount())
+        .thenReturn(new EnumCounters<>(SCMCommandProto.Type.class));
+    StateContext context = new StateContext(conf, DatanodeStates.RUNNING, dsm, "");
+
+    HDDSLayoutVersionManager lvm = mock(HDDSLayoutVersionManager.class);
+    when(lvm.getSoftwareLayoutVersion()).thenReturn(maxLayoutVersion());
+    when(lvm.getMetadataLayoutVersion()).thenReturn(maxLayoutVersion());
+
+    DatanodeDetails dn = DatanodeDetails.newBuilder()
+        .setUuid(UUID.randomUUID())
+        .setHostName("localhost")
+        .setIpAddress("127.0.0.1")
+        .build();
+
+    HeartbeatEndpointTask.newBuilder()
+        .setConfig(conf)
+        .setDatanodeDetails(dn)
+        .setContext(context)
+        .setLayoutVersionManager(lvm)
+        .setEndpointStateMachine(endpoint)
+        .build()
+        .call();
+    return connectionManager;
+  }
+}