HDDS-15531. DNS refresh on connection failure for Client to OM

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/ConnectionFailureUtils.java

@@ -0,0 +1,101 @@
+/*
+ * 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.utils;
+
+import java.io.EOFException;
+import java.net.ConnectException;
+import java.net.NoRouteToHostException;
+import java.net.SocketException;
+import java.net.SocketTimeoutException;
+import java.net.UnknownHostException;
+
+/**
+ * Shared classifier for exceptions where the cached peer IP is no longer
+ * reachable and DNS re-resolution is the only plausible recovery path.
+ * <p>
+ * Used by both {@code SCMFailoverProxyProviderBase} and
+ * {@code OMFailoverProxyProviderBase} to gate the DNS-refresh-on-failure
+ * code path so that application-level errors (NotLeader, AccessControl,
+ * OMException, RetryAction) do not trigger spurious DNS lookups.
+ * <p>
+ * The classifier must match the failure shapes seen in production
+ * Kubernetes deployments where the peer pod has been rescheduled to a
+ * new IP under a stable hostname:
+ * <ul>
+ *   <li>{@link ConnectException} -- the TCP SYN was refused. Seen on
+ *       OpenStack / fast-RST environments. </li>
+ *   <li>{@link SocketTimeoutException} (and its IPC subclass
+ *       {@code ConnectTimeoutException}) -- the SYN was dropped silently.
+ *       This is the dominant failure shape on AWS EC2 / EKS where the
+ *       network silently drops packets to a defunct pod IP. The PR that
+ *       introduced this helper (HDDS-15514) is sold on this case; it
+ *       must be in the filter. </li>
+ *   <li>{@link NoRouteToHostException} -- routing table no longer
+ *       reaches the cached IP. </li>
+ *   <li>{@link UnknownHostException} -- the hostname itself failed to
+ *       resolve at the time the IPC layer reconstructed the address. </li>
+ *   <li>{@link EOFException} -- a load balancer or iptables RST closed
+ *       the half-open connection cleanly. Common in Kubernetes when an
+ *       IP is reassigned to an unrelated pod that rejects the RPC
+ *       handshake. </li>
+ *   <li>{@link SocketException} (e.g. "Connection reset") -- the peer
+ *       sent RST mid-stream. </li>
+ * </ul>
+ * The walk is bounded to {@value #MAX_CAUSE_DEPTH} levels to defend
+ * against cause chains that have been constructed (in violation of
+ * {@code Throwable.initCause}'s contract) into a cycle of length &gt; 1.
+ */
+public final class ConnectionFailureUtils {
+
+  /**
+   * Maximum depth of the {@code Throwable.getCause()} chain we walk
+   * before giving up. Matches Hadoop's own walkers in
+   * {@code RemoteException} handling.
+   */
+  static final int MAX_CAUSE_DEPTH = 16;
+
+  private ConnectionFailureUtils() {
+  }
+
+  /**
+   * Returns true when any link in {@code t}'s cause chain (up to
+   * {@link #MAX_CAUSE_DEPTH} levels) is one of the connection-class
+   * exceptions documented on this class.
+   *
+   * @param t the throwable to classify. {@code null} returns false.
+   */
+  public static boolean isConnectionFailure(Throwable t) {
+    Throwable cause = t;
+    for (int depth = 0; cause != null && depth < MAX_CAUSE_DEPTH; depth++) {
+      if (cause instanceof ConnectException
+          || cause instanceof SocketTimeoutException
+          || cause instanceof NoRouteToHostException
+          || cause instanceof UnknownHostException
+          || cause instanceof EOFException
+          || cause instanceof SocketException) {
+        return true;
+      }
+      Throwable next = cause.getCause();
+      if (next == cause) {
+        break;
+      }
+      cause = next;
+    }
+    return false;
+  }
+}

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java

@@ -604,6 +604,19 @@ public final class OzoneConfigKeys {
   public static final boolean OZONE_JVM_NETWORK_ADDRESS_CACHE_ENABLED_DEFAULT =
           true;
 
+  /**
+   * When true, RPC clients (DN heartbeat, OM client, SCM client) re-resolve
+   * cached hostnames on connection failure and rebuild the proxy if the
+   * resolved IP has changed. Set to true in environments where server pod
+   * IPs may change while DNS names remain stable, such as Kubernetes.
+   * Default false preserves pre-fix behavior. Mirrors the design intent of
+   * HADOOP-17068 / HDFS-14118.
+   */
+  public static final String OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY =
+          "ozone.client.failover.resolve-needed";
+  public static final boolean OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_DEFAULT =
+          false;
+
   public static final String OZONE_CLIENT_REQUIRED_OM_VERSION_MIN_KEY =
       "ozone.client.required.om.version.min";
 

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

@@ -3913,6 +3913,39 @@
     </description>
   </property>
 
+  <property>
+    <name>ozone.client.failover.resolve-needed</name>
+    <value>false</value>
+    <tag>OZONE, CLIENT, OM, SCM, HA</tag>
+    <description>When true, RPC clients (DN heartbeat, OM client, SCM
+      client) re-resolve cached hostnames on connection-class failures
+      (ConnectException, SocketTimeoutException, NoRouteToHostException,
+      UnknownHostException, EOFException, SocketException) and rebuild
+      the proxy if the resolved IP has changed. Set to true in
+      environments where server pod IPs may change while DNS names
+      remain stable, such as Kubernetes. Default false preserves
+      pre-fix behaviour. Mirrors the design intent of HADOOP-17068 /
+      HDFS-14118.
+
+      Required co-config for SECURE clusters: when this flag is true,
+      operators must ALSO set hadoop.security.token.service.use_ip=false
+      (in core-site.xml). Reason: the Hadoop delegation-token service
+      identifier defaults to an IP:port string. After a refresh, the
+      per-OM service identifier built from the new IP no longer matches
+      the IP-based service captured on long-lived tokens, and token
+      selection (OzoneDelegationTokenSelector) silently fails for the
+      refreshed peer. With use_ip=false the service identifier is the
+      stable hostname:port, which survives any IP change. Insecure
+      clusters do not need the co-config.
+
+      Note: ozone.network.jvm.address.cache.enabled controls a related
+      but distinct concern -- the JVM-level positive DNS cache TTL.
+      That setting only affects future name lookups; this setting
+      additionally rebuilds long-lived RPC proxies whose
+      InetSocketAddress was frozen at process start.
+    </description>
+  </property>
+
   <property>
     <name>ozone.directory.deleting.service.interval</name>
     <value>1m</value>

hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/utils/TestConnectionFailureUtils.java

@@ -0,0 +1,165 @@
+/*
+ * 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.utils;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.net.ConnectException;
+import java.net.NoRouteToHostException;
+import java.net.SocketException;
+import java.net.SocketTimeoutException;
+import java.net.UnknownHostException;
+import java.util.stream.Stream;
+import org.apache.hadoop.security.AccessControlException;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+/**
+ * Verifies the connection-class exception classifier used to gate the
+ * DNS-refresh-on-failure code path on both the OM and SCM failover
+ * proxy providers and the DataNode heartbeat catch block.
+ * <p>
+ * The classifier must:
+ * <ul>
+ *   <li>Match every exception type that signals "the cached IP is no
+ *       longer reachable" -- including the AWS EC2 / EKS silent-drop
+ *       case which surfaces as {@link SocketTimeoutException}, the
+ *       case the PR motivating this helper (HDDS-15514) is sold on. </li>
+ *   <li>Reject application-level errors (NotLeader, AccessControl,
+ *       protocol mismatch) so we don't add DNS load on logical
+ *       failures where the cached IP is fine. </li>
+ *   <li>Walk wrapped cause chains so a {@code RemoteException(...)} or
+ *       {@code IOException(...)} carrying a connection-class cause is
+ *       still classified correctly. </li>
+ *   <li>Defend against pathological cycles in the cause chain. </li>
+ * </ul>
+ */
+public class TestConnectionFailureUtils {
+
+  static Stream<Arguments> connectionClassExceptions() {
+    return Stream.of(
+        Arguments.of(new ConnectException("refused"),         "ConnectException"),
+        Arguments.of(new SocketTimeoutException("EC2 drop"),  "SocketTimeoutException (AWS silent drop)"),
+        Arguments.of(new NoRouteToHostException("gone"),      "NoRouteToHostException"),
+        Arguments.of(new UnknownHostException("dns failed"),  "UnknownHostException"),
+        Arguments.of(new EOFException("LB closed"),           "EOFException"),
+        Arguments.of(new SocketException("Connection reset"), "SocketException")
+    );
+  }
+
+  @ParameterizedTest(name = "isConnectionFailure detects {1}")
+  @MethodSource("connectionClassExceptions")
+  public void testDetectsBareConnectionClass(Throwable t, String label) {
+    assertTrue(ConnectionFailureUtils.isConnectionFailure(t),
+        label + " must be classified as a connection failure");
+  }
+
+  @ParameterizedTest(name = "isConnectionFailure walks IOException wrap of {1}")
+  @MethodSource("connectionClassExceptions")
+  public void testDetectsThroughIOExceptionWrap(Throwable t, String label) {
+    IOException wrapped = new IOException("rpc failed", t);
+    assertTrue(ConnectionFailureUtils.isConnectionFailure(wrapped),
+        "IOException wrapping " + label + " must still be classified");
+  }
+
+  @Test
+  public void testDeeplyNestedChainStillClassified() {
+    // ConnectException three levels deep, the way Hadoop RPC's RetriableException
+    // wraps ServiceException wraps IOException wraps the real cause.
+    Throwable deep = new RuntimeException("outer",
+        new IOException("middle",
+            new IOException("inner", new ConnectException("dead"))));
+    assertTrue(ConnectionFailureUtils.isConnectionFailure(deep));
+  }
+
+  static Stream<Arguments> applicationLevelExceptions() {
+    return Stream.of(
+        Arguments.of(new AccessControlException("denied"),
+            "AccessControlException"),
+        Arguments.of(new IllegalArgumentException("bad request"),
+            "IllegalArgumentException"),
+        Arguments.of(new IOException("application error: not leader"),
+            "plain IOException without connection-class cause"),
+        Arguments.of(new RuntimeException("retry, please"),
+            "plain RuntimeException")
+    );
+  }
+
+  @ParameterizedTest(name = "isConnectionFailure rejects {1}")
+  @MethodSource("applicationLevelExceptions")
+  public void testRejectsApplicationLevel(Throwable t, String label) {
+    assertFalse(ConnectionFailureUtils.isConnectionFailure(t),
+        label + " is an application error, refresh must NOT trigger");
+  }
+
+  @Test
+  public void testNullIsNotAConnectionFailure() {
+    assertFalse(ConnectionFailureUtils.isConnectionFailure(null));
+  }
+
+  /**
+   * {@code Throwable.initCause} contractually rejects setting cause to
+   * the throwable itself, but cycles of length 2+ have appeared in
+   * practice (proxy frameworks and faulty initCause callers can
+   * construct them). The walk must terminate within the configured
+   * depth bound rather than looping forever.
+   * <p>
+   * We build the length-2 cycle through {@link Throwable#initCause}
+   * (no reflection) -- the no-arg ctor leaves cause uninitialized
+   * (cause==this sentinel), so a single initCause call on each side
+   * is permitted and lets us close the cycle.
+   */
+  @Test
+  public void testCycleOfLengthTwoTerminates() {
+    Throwable a = new IOException();
+    Throwable b = new IOException();
+    a.initCause(b);
+    b.initCause(a);
+    // Neither a nor b is a connection-class type. The walk must return
+    // false (not loop forever and not throw).
+    assertFalse(ConnectionFailureUtils.isConnectionFailure(a),
+        "length-2 cycle must terminate cleanly");
+  }
+
+  /**
+   * Defense against an unbounded chain of non-connection-class
+   * exceptions: the depth bound must kick in.
+   * <p>
+   * Built using {@link Throwable#initCause} on freshly-constructed
+   * exceptions (no-arg ctor leaves cause uninitialized) so the test
+   * does not depend on JDK-internal reflective access to the
+   * {@code cause} field, which fails on JDK 16+ without
+   * {@code --add-opens java.base/java.lang=ALL-UNNAMED}.
+   */
+  @Test
+  public void testUnboundedChainOfNonMatchingTerminates() {
+    Throwable head = new RuntimeException();
+    Throwable cursor = head;
+    for (int i = 1; i < 1024; i++) {
+      Throwable next = new RuntimeException();
+      cursor.initCause(next);
+      cursor = next;
+    }
+    assertFalse(ConnectionFailureUtils.isConnectionFailure(head));
+  }
+}

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHAManagerImpl.java

@@ -118,9 +118,16 @@ public void start() throws IOException {
       final boolean success = HAUtils.addSCM(ozoneConf,
           new AddSCMRequest.Builder().setClusterId(scm.getClusterId())
               .setScmId(scm.getScmId())
-              .setRatisAddr(nodeDetails
-                  // TODO : Should we use IP instead of hostname??
-                  .getRatisHostPortStr()).build(), scm.getSCMNodeId());
+              // Pass the configured host:port string verbatim. Do NOT
+              // resolve it into an InetSocketAddress first -- that bakes
+              // a resolved IP into Ratis's peer address for the channel's
+              // lifetime. With the string passed through, gRPC's
+              // DnsNameResolver re-resolves hostname addresses on
+              // connection failure (peer pod restarts recover
+              // automatically), and IP-literal configs are still honored
+              // exactly as configured. See HDDS-15514.
+              .setRatisAddr(nodeDetails.getRatisHostPortStr())
+              .build(), scm.getSCMNodeId());
       if (!success) {
         throw new IOException("Adding SCM to existing HA group failed");
       } else {

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMRatisServerImpl.java

@@ -396,8 +396,18 @@ private static RaftGroup buildRaftGroup(SCMNodeDetails details,
     final RaftGroupId groupId = buildRaftGroupId(clusterId);
     RaftPeerId selfPeerId = getSelfPeerId(scmId);
 
+    // Pass the configured host:port string through verbatim. The
+    // invariant is "do not pre-resolve" -- never construct an
+    // InetSocketAddress from the configured address and hand the
+    // resolved form to RaftPeer.setAddress, which would freeze the peer
+    // at one IP for the channel's lifetime. Ratis routes the string to
+    // gRPC's NettyChannelBuilder, whose default DnsNameResolver
+    // re-resolves hostname addresses on connection failure (peer pod
+    // restarts recover automatically in Kubernetes-style environments
+    // where DNS names are stable but IPs are not). IP-literal configs
+    // are still honored exactly as configured. See HDDS-15514
+    // (DNS-refresh-on-failure for all RPC paths).
     RaftPeer localRaftPeer = RaftPeer.newBuilder().setId(selfPeerId)
-        // TODO : Should we use IP instead of hostname??
         .setAddress(details.getRatisHostPortStr()).build();
 
     List<RaftPeer> raftPeers = new ArrayList<>();