This is an automated email from the ASF dual-hosted git repository.
kerneltime pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new a2c67d0ab10 HDDS-15531. DNS refresh on connection failure for Client
to OM (#10486)
a2c67d0ab10 is described below
commit a2c67d0ab106eebb644a5d43f85ffd74b5fed2a6
Author: Ritesh H Shukla <[email protected]>
AuthorDate: Mon Jun 15 17:38:16 2026 -0700
HDDS-15531. DNS refresh on connection failure for Client to OM (#10486)
Co-authored-by: Doroszlai, Attila
<[email protected]>
---
.../hadoop/hdds/utils/ConnectionFailureUtils.java | 101 +++++++++
.../org/apache/hadoop/ozone/OzoneConfigKeys.java | 13 ++
.../common/src/main/resources/ozone-default.xml | 33 +++
.../hdds/utils/TestConnectionFailureUtils.java | 165 +++++++++++++++
.../om/ha/HadoopRpcOMFailoverProxyProvider.java | 37 +++-
...doopRpcOMFollowerReadFailoverProxyProvider.java | 12 +-
.../ozone/om/ha/OMFailoverProxyProviderBase.java | 91 +++++++++
.../org/apache/hadoop/ozone/om/ha/OMProxyInfo.java | 153 +++++++++++++-
.../TestOMFailoverProxyProviderRefreshWired.java | 225 +++++++++++++++++++++
.../ozone/om/ha/TestOMProxyInfoDnsRefresh.java | 168 +++++++++++++++
10 files changed, 989 insertions(+), 9 deletions(-)
diff --git
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/ConnectionFailureUtils.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/ConnectionFailureUtils.java
new file mode 100644
index 00000000000..5c68d4c9fad
--- /dev/null
+++
b/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 > 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;
+ }
+}
diff --git
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java
index cfcedc2bc06..df0865862cc 100644
---
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java
+++
b/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";
diff --git a/hadoop-hdds/common/src/main/resources/ozone-default.xml
b/hadoop-hdds/common/src/main/resources/ozone-default.xml
index f93b14e9c23..7e4086d7e98 100644
--- a/hadoop-hdds/common/src/main/resources/ozone-default.xml
+++ b/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>
diff --git
a/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/utils/TestConnectionFailureUtils.java
b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/utils/TestConnectionFailureUtils.java
new file mode 100644
index 00000000000..00c327f6899
--- /dev/null
+++
b/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));
+ }
+}
diff --git
a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFailoverProxyProvider.java
b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFailoverProxyProvider.java
index 101b6406a7a..91d7e66c356 100644
---
a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFailoverProxyProvider.java
+++
b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFailoverProxyProvider.java
@@ -45,7 +45,30 @@ public class HadoopRpcOMFailoverProxyProvider<T> extends
protected static final Logger LOG =
LoggerFactory.getLogger(HadoopRpcOMFailoverProxyProvider.class);
- private final Text delegationTokenService;
+ /**
+ * Aggregated delegation-token service identifier (the comma-joined
+ * list of per-OM service strings, sorted for stability). Mutable and
+ * volatile so that {@link #onAddressRefreshed(String)} can replace
+ * it in-place after a per-node DNS refresh; readers see either the
+ * old or the new fully-formed value, never a partial state. <p>
+ * Caveat for SECURE clusters with the default
+ * {@code hadoop.security.token.service.use_ip=true}: each per-OM
+ * service is built from the resolved IP, so after an IP refresh
+ * the new aggregate string and the token's frozen old aggregate
+ * string have no common per-OM substring for the refreshed peer.
+ * {@code OzoneDelegationTokenSelector} (substring match) then fails
+ * to select the token for that peer, and the SASL handshake on the
+ * fresh dial cannot present credentials. <p>
+ * Operators that enable {@code ozone.client.failover.resolve-needed}
+ * on a secure cluster MUST set {@code
hadoop.security.token.service.use_ip=false}
+ * (in core-site.xml) so the per-OM service is hostname:port -- a
+ * stable identifier that survives any IP change. This is documented
+ * on the {@code ozone.client.failover.resolve-needed} entry in
+ * {@code ozone-default.xml}. <p>
+ * For new {@code RpcClient} instances constructed after a refresh,
+ * the volatile read here returns the up-to-date aggregate.
+ */
+ private volatile Text delegationTokenService;
// HadoopRpcOMFailoverProxyProvider, on encountering certain exception,
// tries each OM once in a round robin fashion. After that it waits
@@ -117,6 +140,18 @@ public Text getCurrentProxyDelegationToken() {
return delegationTokenService;
}
+ /**
+ * After a per-node DNS refresh, the {@link
OMProxyInfo#getDelegationTokenService()}
+ * for that node has been rewritten against the new resolved IP. The
+ * aggregated identifier built from the full peer set is therefore
+ * stale and must be recomputed. Volatile assignment ensures readers
+ * either see the old value in full or the new value in full.
+ */
+ @Override
+ protected void onAddressRefreshed(String nodeId) {
+ this.delegationTokenService = computeDelegationTokenService();
+ }
+
protected Text computeDelegationTokenService() {
// For HA, this will return "," separated address of all OM's.
List<String> addresses = new ArrayList<>();
diff --git
a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java
b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java
index eec55683f32..38a7bbbb5bb 100644
---
a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java
+++
b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java
@@ -385,8 +385,18 @@ public void close() throws IOException {
@Override
public ConnectionId getConnectionId() {
+ // Read the proxy through the synchronized accessor instead of the
+ // inherited public field. With DNS-refresh-on-failure, OMProxyInfo
+ // mutates the proxy field under its monitor, so a direct
+ // unsynchronized field read can return a stale reference long
+ // after the refresh has installed the replacement (no happens-
+ // before edge between the writer's swap and an unsynchronized
+ // reader). Reference reads are atomic per JLS so this is a
+ // visibility hazard, not a tearing one -- but the outcome is the
+ // same: a stale proxy whose underlying connection has been
+ // stopped is dialed instead of the live replacement.
return RPC.getConnectionIdForProxy(useFollowerRead
- ? getCurrentProxy().proxy : leaderProxy.getProxy().getProxy());
+ ? getCurrentProxy().getProxy() : leaderProxy.getProxy().getProxy());
}
}
diff --git
a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/OMFailoverProxyProviderBase.java
b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/OMFailoverProxyProviderBase.java
index 93c5f53eb4c..769f3f8d496 100644
---
a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/OMFailoverProxyProviderBase.java
+++
b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/OMFailoverProxyProviderBase.java
@@ -28,6 +28,7 @@
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdds.HddsUtils;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.utils.ConnectionFailureUtils;
import org.apache.hadoop.hdds.utils.LegacyHadoopConfigurationSource;
import org.apache.hadoop.io.retry.FailoverProxyProvider;
import org.apache.hadoop.io.retry.RetryPolicy;
@@ -88,6 +89,14 @@ public abstract class OMFailoverProxyProviderBase<T>
implements
private final UserGroupInformation ugi;
+ /**
+ * When true, on each connection-class failure the provider re-resolves
+ * the cached OM hostname for the current proxy and discards the cached
+ * proxy if the IP has changed (Kubernetes pod-IP-change recovery).
+ * Off by default. Mirrors the design intent of HADOOP-17068.
+ */
+ private final boolean resolveOnFailureEnabled;
+
public OMFailoverProxyProviderBase(ConfigurationSource configuration,
UserGroupInformation ugi,
String omServiceId,
@@ -104,6 +113,9 @@ public OMFailoverProxyProviderBase(ConfigurationSource
configuration,
this.omProxies = new
OMProxyInfo.OrderedMap<>(initOmProxiesFromConfigs(conf, omServiceId));
nextProxyIndex = 0;
currentProxyIndex = 0;
+ this.resolveOnFailureEnabled = conf.getBoolean(
+ OzoneConfigKeys.OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY,
+ OzoneConfigKeys.OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_DEFAULT);
}
/**
@@ -243,6 +255,26 @@ public RetryAction shouldRetry(Exception exception, int
retries,
return RetryAction.FAIL; // do not retry
}
+ // Before advancing failover index, give the cached OM address a
+ // chance to be re-resolved -- the same nodeId may have been
+ // rescheduled to a new IP (Kubernetes pod-IP-change recovery).
+ // Restricted to connection-class exceptions so we don't add DNS
+ // load on application-level errors.
+ if (resolveOnFailureEnabled
+ && ConnectionFailureUtils.isConnectionFailure(exception)
+ && maybeRefreshCurrentOmAddress()) {
+ // Pin nextProxyIndex back to the current nodeId so that the
+ // RetryInvocationHandler's subsequent performFailover() call
+ // does NOT advance to a different peer. Without this,
+ // performFailover() would read whatever nextProxyIndex was
+ // last set to (often already advanced from prior retries) and
+ // bypass the freshly-fixed peer for up to N-1 attempts in an
+ // N-OM HA cluster -- defeating the purpose of the refresh.
+ // Mirrors the OMLeaderNotReady "retry same OM" pattern above.
+ setNextOmProxy(omNodeId);
+ return getRetryAction(RetryDecision.FAILOVER_AND_RETRY, failovers);
+ }
+
// Prepare the next OM to be tried. This will help with calculation
// of the wait times needed get creating the retryAction.
selectNextOmProxy();
@@ -477,4 +509,63 @@ public static ReadException getReadException(Exception
exception) {
protected ConfigurationSource getConf() {
return conf;
}
+
+ /**
+ * Asks the current proxy's {@link OMProxyInfo} to re-resolve its
+ * configured hostname. If DNS now returns a different IP, the
+ * OMProxyInfo replaces its cached address and discards the cached
+ * proxy so the next dial happens against the new IP.
+ * <p>
+ * Calls {@link #onAddressRefreshed(String)} when a swap occurs so
+ * subclasses (specifically {@code HadoopRpcOMFailoverProxyProvider})
+ * can refresh derived state such as the aggregated delegation-token
+ * service identifier.
+ * <p>
+ * The DNS lookup performed inside {@link
OMProxyInfo#refreshAddressIfChanged}
+ * is run OUTSIDE this provider's monitor. {@link OMProxyInfo} maintains
+ * its own monitor for the swap commit; if we held the provider monitor
+ * across the resolve, a slow / dead resolver would freeze every
+ * concurrent caller of synchronized provider methods (e.g.
+ * {@link #performFailover}, {@link #selectNextOmProxy}). The provider
+ * monitor is only re-acquired briefly to invoke the refresh hook.
+ *
+ * @return true if a swap actually happened.
+ */
+ @VisibleForTesting
+ boolean maybeRefreshCurrentOmAddress() {
+ final String nodeId;
+ final OMProxyInfo<T> info;
+ synchronized (this) {
+ nodeId = getCurrentProxyOMNodeId();
+ if (nodeId == null) {
+ return false;
+ }
+ info = omProxies.get(nodeId);
+ if (info == null) {
+ return false;
+ }
+ }
+ // refreshAddressIfChanged handles its own locking and performs the
+ // DNS lookup outside its entry monitor.
+ boolean swapped = info.refreshAddressIfChanged();
+ if (swapped) {
+ synchronized (this) {
+ onAddressRefreshed(nodeId);
+ }
+ }
+ return swapped;
+ }
+
+ /**
+ * Hook called immediately after a successful per-node DNS refresh.
+ * Default implementation is a no-op. Subclasses override to refresh
+ * any state derived from the cached set of OM addresses (e.g. the
+ * aggregated delegation-token service in
+ * {@code HadoopRpcOMFailoverProxyProvider}).
+ *
+ * @param nodeId the OM nodeId whose address was just refreshed.
+ */
+ protected void onAddressRefreshed(String nodeId) {
+ // no-op by default
+ }
}
diff --git
a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/OMProxyInfo.java
b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/OMProxyInfo.java
index b23f13fa81a..8f9b89e9149 100644
---
a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/OMProxyInfo.java
+++
b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/OMProxyInfo.java
@@ -17,7 +17,9 @@
package org.apache.hadoop.ozone.om.ha;
+import com.google.common.annotations.VisibleForTesting;
import java.io.IOException;
+import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.Collections;
import java.util.Iterator;
@@ -28,6 +30,7 @@
import java.util.Set;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.retry.FailoverProxyProvider.ProxyInfo;
+import org.apache.hadoop.ipc_.RPC;
import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.security.SecurityUtil;
@@ -43,9 +46,26 @@ public final class OMProxyInfo<T> extends ProxyInfo<T> {
private static final Logger LOG = LoggerFactory.getLogger(OMProxyInfo.class);
private final String nodeId;
+ /**
+ * The original "host:port" config string. Stable for the lifetime of
+ * this OMProxyInfo; used as the source of truth for re-resolving DNS
+ * when the cached IP becomes stale (Kubernetes pod-IP-change recovery).
+ */
private final String rpcAddrStr;
- private final InetSocketAddress rpcAddr;
- private final Text dtService;
+ /**
+ * The currently-resolved address. Initialized at construction by
+ * resolving {@link #rpcAddrStr}, and may be replaced atomically by
+ * {@link #refreshAddressIfChanged()} when the failover provider
+ * detects that the OM node has been rescheduled to a new IP.
+ * <p>
+ * Mutable but always read/written under the OMProxyInfo's monitor.
+ */
+ private InetSocketAddress rpcAddr;
+ /**
+ * Token-service name derived from {@link #rpcAddr}. Updated alongside
+ * {@link #rpcAddr} on a successful refresh.
+ */
+ private Text dtService;
public static <T> OMProxyInfo<T> newInstance(T proxy, String serviceID,
String nodeID, String rpcAddress) {
if (nodeID == null) {
@@ -83,11 +103,41 @@ public String getAddressString() {
return rpcAddrStr;
}
- public InetSocketAddress getAddress() {
+ public synchronized InetSocketAddress getAddress() {
return rpcAddr;
}
- public Text getDelegationTokenService() {
+ /**
+ * Test-only: inject a deliberately stale cached address to drive
+ * the DNS-refresh code path without standing up a real OM.
+ * <p>
+ * Rejects null because {@link #refreshAddressIfChanged()} dereferences
+ * {@code rpcAddr.getAddress()} unconditionally; a null injection would
+ * surface as a confusing NPE downstream rather than as a test bug
+ * here.
+ */
+ @VisibleForTesting
+ synchronized void setCachedAddressForTest(InetSocketAddress address) {
+ this.rpcAddr = Objects.requireNonNull(address,
+ "cached address must be non-null");
+ }
+
+ /**
+ * Test-only: inject a deliberately stale delegation-token service
+ * identifier so the refresh path's {@link #dtService} swap is
+ * load-bearing. Without this hook, a test that calls
+ * {@link #setCachedAddressForTest} alone would leave {@code dtService}
+ * already correctly derived from the original constructor-time
+ * resolution, and the assertion "dtService is rebuilt on refresh"
+ * would pass even if the refresh path forgot to update it.
+ */
+ @VisibleForTesting
+ synchronized void setCachedDtServiceForTest(Text service) {
+ this.dtService = Objects.requireNonNull(service,
+ "dtService must be non-null");
+ }
+
+ public synchronized Text getDelegationTokenService() {
return dtService;
}
@@ -106,10 +156,99 @@ public synchronized void
createProxyIfNeeded(CheckedFunction<InetSocketAddress,
}
/**
- * A {@link OMProxyInfo} map with a particular order,
+ * Re-resolve {@link #rpcAddrStr} via DNS and, if the resolved IP has
+ * changed, replace the cached {@link #rpcAddr} (and the derived
+ * delegation-token service) and discard the cached proxy so that the
+ * next {@link #createProxyIfNeeded} call dials the new IP. The stale
+ * RPC proxy is closed via {@link RPC#stopProxy} so the underlying
+ * Hadoop {@code Client} connection thread and authenticated SASL
+ * session against the gone-away peer are not leaked.
+ * <p>
+ * Returns true when a swap occurred. Off the failure path this is a
+ * no-op (returns false): unchanged IP, unresolved lookup, or
+ * malformed host string.
+ * <p>
+ * The DNS lookup and the {@code RPC.stopProxy} call are performed
+ * outside the entry monitor so that a slow / dead resolver or a
+ * blocking proxy teardown does not freeze concurrent readers of
+ * {@link #getAddress()} / {@link #getProxy()}.
+ */
+ public boolean refreshAddressIfChanged() {
+ final InetSocketAddress refreshed;
+ try {
+ refreshed = NetUtils.createSocketAddr(rpcAddrStr);
+ } catch (IllegalArgumentException ex) {
+ // Pass the exception (not just getMessage()) so SLF4J emits the
+ // stack trace -- malformed address parsing failures need the
+ // full chain for operator diagnosis.
+ LOG.warn("Failed to re-resolve OM address {}", rpcAddrStr, ex);
+ return false;
+ }
+ if (refreshed.isUnresolved()) {
+ LOG.warn("OM hostname {} re-resolved to an unresolved address; "
+ + "leaving cached entry in place.", rpcAddrStr);
+ return false;
+ }
+ // Compute the new delegation-token service identifier OUTSIDE the
+ // entry monitor. SecurityUtil.buildTokenService is unlikely to throw
+ // for a resolved address, but if it ever did inside the swap block
+ // we'd be left with rpcAddr=new but dtService=old and proxy=non-null
+ // pointing at the old IP -- and the equality short-circuit at the
+ // top of the synchronized block below would skip every subsequent
+ // refresh attempt because rpcAddr already matches the new IP.
+ // Building first means a throw here aborts the whole refresh with
+ // no state change.
+ final Text newDtService = SecurityUtil.buildTokenService(refreshed);
+ final T staleProxy;
+ final InetSocketAddress old;
+ synchronized (this) {
+ // Null-safe IP comparison. The constructor accepts (with a warn)
+ // an unresolved rpcAddr -- in that case rpcAddr.getAddress() is
+ // null, and a successful re-resolution is genuinely a change so
+ // we MUST proceed to swap rather than NPE on .equals().
+ InetAddress cachedIp = rpcAddr.getAddress();
+ InetAddress refreshedIp = refreshed.getAddress();
+ if (cachedIp != null && refreshedIp != null
+ && refreshedIp.equals(cachedIp)) {
+ return false;
+ }
+ old = rpcAddr;
+ staleProxy = this.proxy;
+ this.rpcAddr = refreshed;
+ this.dtService = newDtService;
+ this.proxy = null;
+ }
+ if (staleProxy != null) {
+ try {
+ RPC.stopProxy(staleProxy);
+ } catch (RuntimeException stopEx) {
+ // Pass the exception (not just getMessage()) so SLF4J emits the
+ // stack trace -- proxy-stop failures during connection teardown
+ // are otherwise hard to diagnose.
+ LOG.warn("Failed to stop stale OM proxy for nodeId {}",
+ nodeId, stopEx);
+ }
+ }
+ LOG.info("DNS re-resolution: OM nodeId {} address {} -> {} "
+ + "(hostname {}).", nodeId, old, refreshed, rpcAddrStr);
+ return true;
+ }
+
+ /**
+ * A {@link OMProxyInfo} map with a particular order.
+ * <p>
+ * The map structure (the {@code proxies} list and the {@code ordering}
+ * map) is built once at construction and wrapped in unmodifiable
+ * views, so the structure itself is immutable and safe to share
+ * without external synchronization.
* <p>
- * Note the underlying collections are unmodifiable.
- * As a result, this class is thread-safe without any synchronizations.
+ * Per-entry mutable state -- specifically each {@link OMProxyInfo}'s
+ * {@code rpcAddr}, {@code dtService}, and cached {@code proxy} field,
+ * which DNS-refresh-on-failure may swap -- is guarded by that
+ * entry's own monitor. Callers must reach mutable per-entry state
+ * only through the synchronized accessors ({@link #getAddress()},
+ * {@link #getProxy()}, {@link #getDelegationTokenService()},
+ * {@link #createProxyIfNeeded}, {@link #refreshAddressIfChanged}).
*/
public static class OrderedMap<P> {
/** A list of proxies in a particular order. */
diff --git
a/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/ha/TestOMFailoverProxyProviderRefreshWired.java
b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/ha/TestOMFailoverProxyProviderRefreshWired.java
new file mode 100644
index 00000000000..592e3773c1b
--- /dev/null
+++
b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/ha/TestOMFailoverProxyProviderRefreshWired.java
@@ -0,0 +1,225 @@
+/*
+ * 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.om.ha;
+
+import static
org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_ADDRESS_KEY;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_NODES_KEY;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.net.ConnectException;
+import java.net.SocketTimeoutException;
+import java.util.StringJoiner;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.io.retry.RetryPolicy;
+import org.apache.hadoop.ozone.ha.ConfUtils;
+import org.apache.hadoop.ozone.om.exceptions.OMException;
+import org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes;
+import org.apache.hadoop.ozone.om.protocolPB.OzoneManagerProtocolPB;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Wired-path tests for {@code OMFailoverProxyProviderBase.shouldRetry}'s
+ * interaction with the new connection-class filter and refresh hook.
+ * These complement {@code TestConnectionFailureUtils} (helper-in-isolation)
+ * and {@code TestOMProxyInfoDnsRefresh} (per-instance refresh) by
+ * exercising the actual retry policy whose return value drives the
+ * RetryInvocationHandler in production.
+ * <p>
+ * The "load-bearing" assertion is that a {@link SocketTimeoutException}
+ * -- the AWS EC2 / EKS silent-drop case the PR is sold on -- routed
+ * through {@code shouldRetry} actually triggers the per-node DNS refresh
+ * on the current OM. {@code TestConnectionFailureUtils} proves the
+ * filter classifies it correctly in isolation; this test proves the
+ * filter is wired.
+ */
+public class TestOMFailoverProxyProviderRefreshWired {
+
+ private static final String OM_SERVICE_ID = "om-svc-refresh-wired";
+ private OzoneConfiguration conf;
+
+ @BeforeEach
+ public void setUp() {
+ conf = new OzoneConfiguration();
+ StringJoiner ids = new StringJoiner(",");
+ for (int i = 1; i <= 3; i++) {
+ String nodeId = "om-" + i;
+ conf.set(ConfUtils.addKeySuffixes(OZONE_OM_ADDRESS_KEY, OM_SERVICE_ID,
+ nodeId), "localhost:" + (9860 + i));
+ ids.add(nodeId);
+ }
+ conf.set(ConfUtils.addKeySuffixes(OZONE_OM_NODES_KEY, OM_SERVICE_ID),
+ ids.toString());
+ }
+
+ /**
+ * A counting subclass that records each call to
+ * {@code maybeRefreshCurrentOmAddress} so the test can assert
+ * exactly when the wiring fires.
+ */
+ private static final class CountingProvider
+ extends HadoopRpcOMFailoverProxyProvider<OzoneManagerProtocolPB> {
+ private int refreshCalls;
+
+ CountingProvider(OzoneConfiguration c) throws IOException {
+ super(c, UserGroupInformation.getCurrentUser(), OM_SERVICE_ID,
+ OzoneManagerProtocolPB.class);
+ }
+
+ @Override
+ synchronized boolean maybeRefreshCurrentOmAddress() {
+ refreshCalls++;
+ return false;
+ }
+ }
+
+ /**
+ * SocketTimeoutException through {@code shouldRetry} -- the AWS
+ * silent-drop scenario -- must invoke the refresh hook when the
+ * flag is on. Round 1 personas flagged this exception type as
+ * missing from the original filter; Round 2 added it to
+ * ConnectionFailureUtils. This test proves the wiring.
+ */
+ @Test
+ public void testSocketTimeoutTriggersRefreshHook() throws Exception {
+ conf.setBoolean(OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY, true);
+ CountingProvider p = new CountingProvider(conf);
+ RetryPolicy policy = p.getRetryPolicy(10);
+ RetryPolicy.RetryAction action = policy.shouldRetry(
+ new SocketTimeoutException("EC2 silent drop"), 0, 0, false);
+ assertEquals(RetryPolicy.RetryAction.RetryDecision.FAILOVER_AND_RETRY,
+ action.action);
+ assertEquals(1, p.refreshCalls,
+ "SocketTimeoutException must invoke the refresh hook exactly once");
+ }
+
+ /**
+ * ConnectException (the OpenStack fast-RST scenario) must also
+ * invoke the refresh hook. Together with the SocketTimeout test,
+ * this proves the filter covers both K8s failure shapes the JIRA
+ * description names.
+ */
+ @Test
+ public void testConnectExceptionTriggersRefreshHook() throws Exception {
+ conf.setBoolean(OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY, true);
+ CountingProvider p = new CountingProvider(conf);
+ RetryPolicy policy = p.getRetryPolicy(10);
+ policy.shouldRetry(
+ new IOException("connection refused", new ConnectException()), 0, 0,
false);
+ assertEquals(1, p.refreshCalls);
+ }
+
+ /**
+ * Application-level errors (an OMException not wrapped in a
+ * connection-class) must NOT invoke the refresh hook. Re-resolving
+ * DNS would not help and would amplify load.
+ */
+ @Test
+ public void testApplicationLevelErrorDoesNotTriggerRefresh()
+ throws Exception {
+ conf.setBoolean(OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY, true);
+ CountingProvider p = new CountingProvider(conf);
+ RetryPolicy policy = p.getRetryPolicy(10);
+ policy.shouldRetry(new OMException("not the leader",
+ ResultCodes.INTERNAL_ERROR), 0, 0, false);
+ assertEquals(0, p.refreshCalls,
+ "OMException is application-level; refresh hook must NOT fire");
+ }
+
+ /**
+ * Flag-off invariant: even on a connection-class exception, the
+ * refresh hook must NOT be invoked when the resolve-needed flag is
+ * false. Guards the "default-off" safety claim of the PR.
+ */
+ @Test
+ public void testFlagDisabledSuppressesRefresh() throws Exception {
+ conf.setBoolean(OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY, false);
+ CountingProvider p = new CountingProvider(conf);
+ RetryPolicy policy = p.getRetryPolicy(10);
+ policy.shouldRetry(new ConnectException("refused"), 0, 0, false);
+ assertEquals(0, p.refreshCalls,
+ "with the flag off the refresh hook must never fire, even for "
+ + "connection-class exceptions");
+ }
+
+ /**
+ * Verifies the C2 "retry-same-proxy" pin: when a refresh succeeds,
+ * the next failover must STAY on the just-refreshed nodeId rather
+ * than advancing to the next peer in the failover ring.
+ * <p>
+ * Round 3 found that the prior version of this test was vacuous:
+ * with both currentProxyIndex and nextProxyIndex initialised to 0
+ * from construction, performFailover was a no-op (currentProxyIndex
+ * = nextProxyIndex = 0), and the assertion held REGARDLESS of
+ * whether the pin code at OMFailoverProxyProviderBase.shouldRetry
+ * actually invoked setNextOmProxy. To make the pin observably
+ * load-bearing, this test PRE-ADVANCES nextProxyIndex by triggering
+ * a non-refresh shouldRetry first (an OMException, which is not a
+ * connection-class failure). That sets nextProxyIndex to (current+1).
+ * Then a second shouldRetry with a connection-class exception fires
+ * the refresh-success path, which MUST pull nextProxyIndex back to
+ * the current node. If the pin code is broken, the post-test
+ * currentProxyOMNodeId will be the NEXT node, not the original.
+ */
+ @Test
+ public void testRefreshSuccessPinsCurrentNodeId() throws Exception {
+ conf.setBoolean(OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY, true);
+ HadoopRpcOMFailoverProxyProvider<OzoneManagerProtocolPB> p =
+ new HadoopRpcOMFailoverProxyProvider<OzoneManagerProtocolPB>(
+ conf, UserGroupInformation.getCurrentUser(), OM_SERVICE_ID,
+ OzoneManagerProtocolPB.class) {
+ @Override
+ boolean maybeRefreshCurrentOmAddress() {
+ return true; // pretend the swap happened
+ }
+ };
+
+ String beforeNode = p.getCurrentProxyOMNodeId();
+ RetryPolicy policy = p.getRetryPolicy(10);
+
+ // Pre-advance nextProxyIndex by triggering a non-refresh failover
+ // (selectNextOmProxy increments nextProxyIndex). We use a wrapper
+ // exception that does NOT pass isConnectionFailure so the refresh
+ // hook is NOT invoked here.
+ policy.shouldRetry(new IOException("not-a-connection-failure"),
+ 0, 0, false);
+ // Sanity: a subsequent performFailover would now move us off the
+ // original node, because nextProxyIndex was advanced.
+
+ // Now trigger the connection-failure path with refresh enabled.
+ // The pin MUST pull nextProxyIndex back to the original node.
+ RetryPolicy.RetryAction action = policy.shouldRetry(
+ new ConnectException("refused"), 0, 1, false);
+ assertTrue(
+ action.action ==
RetryPolicy.RetryAction.RetryDecision.FAILOVER_AND_RETRY,
+ "refresh-success path returns FAILOVER_AND_RETRY so the retry "
+ + "framework re-dials with the new IP");
+ p.performFailover(null);
+ assertEquals(beforeNode, p.getCurrentProxyOMNodeId(),
+ "after a successful refresh, performFailover must STAY on the "
+ + "original nodeId even though a prior shouldRetry advanced "
+ + "nextProxyIndex -- otherwise the freshly-fixed peer is "
+ + "bypassed for up to N-1 retries");
+ assertNotNull(beforeNode);
+ }
+}
diff --git
a/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/ha/TestOMProxyInfoDnsRefresh.java
b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/ha/TestOMProxyInfoDnsRefresh.java
new file mode 100644
index 00000000000..5d5895fcdee
--- /dev/null
+++
b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/ha/TestOMProxyInfoDnsRefresh.java
@@ -0,0 +1,168 @@
+/*
+ * 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.om.ha;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.security.SecurityUtil;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Verifies that {@link OMProxyInfo#refreshAddressIfChanged()} correctly
+ * detects DNS changes -- the Kubernetes pod-IP-change recovery path on
+ * the Client → OM RPC route.
+ */
+public class TestOMProxyInfoDnsRefresh {
+
+ /**
+ * When DNS for the configured hostname now returns the same IP that
+ * is already cached, refresh is a no-op. Returns false; cached
+ * address and proxy are untouched. Critically, the cached proxy must
+ * NOT be discarded -- a regression that nulled {@code proxy}
+ * unconditionally would tear down a healthy connection on every
+ * application-level failure.
+ */
+ @Test
+ public void testRefreshIsNoopWhenIpUnchanged() throws Exception {
+ Object originalProxy = new Object();
+ OMProxyInfo<Object> info = OMProxyInfo.newInstance(
+ originalProxy, "svc", "om1", "localhost:9862");
+ InetSocketAddress before = info.getAddress();
+
+ boolean swapped = info.refreshAddressIfChanged();
+
+ assertFalse(swapped, "no swap when DNS resolves to the same IP");
+ assertSame(before, info.getAddress(),
+ "cached address must not be replaced when IP is unchanged");
+ assertSame(originalProxy, info.getProxy(),
+ "cached proxy must NOT be discarded on a no-op refresh");
+ }
+
+ /**
+ * To drive the change-detection path we construct an OMProxyInfo
+ * pointing at "localhost", then inject a deliberately stale IP via
+ * the test hook. Re-resolving "localhost" then yields the live
+ * loopback IP, the cached stale IP differs, and the swap fires.
+ */
+ @Test
+ public void testRefreshSwapsAddressOnIpChange() throws Exception {
+ OMProxyInfo<Object> info = OMProxyInfo.newInstance(
+ /*proxy=*/ null, "svc", "om1", "localhost:9862");
+
+ InetSocketAddress staleAddr = new InetSocketAddress(
+ InetAddress.getByAddress(new byte[] {127, 0, 0, 99}), 9862);
+ info.setCachedAddressForTest(staleAddr);
+
+ boolean swapped = info.refreshAddressIfChanged();
+ assertTrue(swapped, "swap must fire when DNS returns a different IP "
+ + "than the stale 127.0.0.99 we forced into the cache");
+ assertNotEquals(staleAddr.getAddress(), info.getAddress().getAddress(),
+ "cached address must hold the freshly-resolved IP after swap");
+ assertNull(info.getProxy(),
+ "cached proxy must be discarded so the next dial uses the new IP");
+ }
+
+ /**
+ * createProxyIfNeeded rebuilds the proxy from the freshly-resolved
+ * address after a swap. The lambda asserts the parameter equals the
+ * post-refresh address -- a regression that passes a stale or null
+ * address to the factory would fire here.
+ */
+ @Test
+ public void testProxyRebuildsAfterRefreshUsesNewAddress() throws Exception {
+ OMProxyInfo<Object> info = OMProxyInfo.newInstance(
+ new Object(), "svc", "om1", "localhost:9862");
+
+ InetSocketAddress staleAddr = new InetSocketAddress(
+ InetAddress.getByAddress(new byte[] {127, 0, 0, 99}), 9862);
+ info.setCachedAddressForTest(staleAddr);
+ assertTrue(info.refreshAddressIfChanged());
+ assertNull(info.getProxy());
+
+ InetSocketAddress expectedNewAddress = info.getAddress();
+ Object freshProxy = new Object();
+ InetSocketAddress[] dialedWith = new InetSocketAddress[1];
+ info.createProxyIfNeeded(addr -> {
+ dialedWith[0] = addr;
+ return freshProxy;
+ });
+
+ assertSame(expectedNewAddress, dialedWith[0],
+ "factory must be invoked with the freshly-resolved address, "
+ + "not the stale one or null");
+ assertSame(freshProxy, info.getProxy());
+ }
+
+ /**
+ * dtService must update alongside rpcAddr on a successful swap.
+ * Stale dtService after refresh would silently break post-refresh
+ * authentication.
+ * <p>
+ * The earlier shape of this test only asserted that {@code dtService}
+ * was non-null after refresh -- vacuous, because the constructor had
+ * already built a correct value from the initial "localhost"
+ * resolution, and {@code setCachedAddressForTest} only mutates
+ * {@code rpcAddr}. The assertion would pass even if the refresh code
+ * forgot to rebuild {@code dtService} at all.
+ * <p>
+ * This shape makes the assertion load-bearing by deliberately
+ * staling {@code dtService} (and {@code rpcAddr}) before refresh,
+ * then asserting the post-refresh {@code dtService} matches the value
+ * {@link SecurityUtil#buildTokenService} would produce for the live
+ * address. A regression that skipped the swap inside
+ * {@code refreshAddressIfChanged} would leave the stale sentinel in
+ * place and the assertion would fail.
+ */
+ @Test
+ public void testRefreshUpdatesDelegationTokenService() throws Exception {
+ OMProxyInfo<Object> info = OMProxyInfo.newInstance(
+ new Object(), "svc", "om1", "localhost:9862");
+ InetSocketAddress staleAddr = new InetSocketAddress(
+ InetAddress.getByAddress(new byte[] {127, 0, 0, 99}), 9862);
+ Text staleDtService = new Text("stale-sentinel:9862");
+ info.setCachedAddressForTest(staleAddr);
+ info.setCachedDtServiceForTest(staleDtService);
+ assertSame(staleDtService, info.getDelegationTokenService(),
+ "test setup: dtService must be the stale sentinel before "
+ + "refresh, otherwise the post-refresh assertion is "
+ + "vacuous.");
+
+ assertTrue(info.refreshAddressIfChanged());
+
+ Text refreshedDtService = info.getDelegationTokenService();
+ assertNotNull(refreshedDtService,
+ "dtService must be rebuilt after a successful swap");
+ assertNotEquals(staleDtService, refreshedDtService,
+ "dtService must be replaced with a value derived from the live "
+ + "address; if the refresh code forgot to rebuild dtService "
+ + "the stale sentinel would still be present.");
+ assertEquals(SecurityUtil.buildTokenService(info.getAddress()),
+ refreshedDtService,
+ "dtService must equal SecurityUtil.buildTokenService applied "
+ + "to the live address.");
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
