This is an automated email from the ASF dual-hosted git repository.
adoroszlai 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 273bf2f39b8 HDDS-15533. DNS refresh on heartbeat failure for DN to SCM
(#10844)
273bf2f39b8 is described below
commit 273bf2f39b85ba1dfadac4530e1c5b81cb0a67d6
Author: Ritesh H Shukla <[email protected]>
AuthorDate: Tue Jul 28 02:23:09 2026 -0700
HDDS-15533. DNS refresh on heartbeat failure for DN to SCM (#10844)
Generated-by: Claude Code (Fable 5)
---
.../org/apache/hadoop/hdds/HddsConfigKeys.java | 4 +
.../apache/hadoop/hdds/scm/net/HostAndPort.java | 20 +++-
.../common/src/main/resources/ozone-default.xml | 10 ++
.../hadoop/hdds/scm/net/TestHostAndPort.java | 56 +++++++++
.../common/statemachine/EndpointStateMachine.java | 10 +-
.../common/statemachine/SCMConnectionManager.java | 98 ++++++++++-----
.../states/endpoint/HeartbeatEndpointTask.java | 32 +++++
.../statemachine/TestSCMConnectionManager.java | 91 ++++++++++++++
.../TestHeartbeatEndpointTaskDnsRefresh.java | 133 +++++++++++++++++++++
9 files changed, 422 insertions(+), 32 deletions(-)
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
index 8e9d4e9e098..4804bfdebac 100644
---
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_MISSED_COUNT_THRESHOLD =
+ "hdds.heartbeat.address.refresh.missed-count-threshold";
+ public static final int
HDDS_HEARTBEAT_ADDRESS_REFRESH_MISSED_COUNT_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
index e89ad73d31b..2a142ea8a11 100644
---
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,7 +17,9 @@
package org.apache.hadoop.hdds.scm.net;
+import java.net.InetAddress;
import java.net.InetSocketAddress;
+import java.util.Objects;
import org.apache.hadoop.net.NetUtils;
/**
@@ -30,14 +32,13 @@ public class HostAndPort {
private final String hostAndPortString;
private final int hash;
/** The address can be updated from time to time. */
- private final InetSocketAddress address;
+ private volatile InetSocketAddress address;
public HostAndPort(String host, int port) {
this.host = host;
this.port = port;
this.hostAndPortString = host + ":" + port;
this.hash = host.hashCode() ^ Integer.hashCode(port);
- // TODO: HDDS-15533 change the address resolution logic and make
this.address threadsafe.
this.address = NetUtils.createSocketAddr(hostAndPortString);
}
@@ -57,6 +58,21 @@ public InetSocketAddress getAddress() {
return address;
}
+ /** Re-resolves host:port and returns the new address if its IP changed,
else null. No mutation. */
+ public InetSocketAddress resolveLatest() {
+ final InetSocketAddress latest =
NetUtils.createSocketAddr(hostAndPortString);
+ final InetAddress latestIp = latest.getAddress();
+ if (latestIp == null || latestIp.equals(address.getAddress())) {
+ return null;
+ }
+ return latest;
+ }
+
+ /** Commits an address re-resolved via {@link #resolveLatest()}. */
+ public void setAddress(InetSocketAddress newAddress) {
+ this.address = Objects.requireNonNull(newAddress, "newAddress == null");
+ }
+
@Override
public int hashCode() {
return hash;
diff --git a/hadoop-hdds/common/src/main/resources/ozone-default.xml
b/hadoop-hdds/common/src/main/resources/ozone-default.xml
index eec93686a6b..1033e639f1c 100644
--- 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.missed-count-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/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/scm/net/TestHostAndPort.java
b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/scm/net/TestHostAndPort.java
new file mode 100644
index 00000000000..9be015f5ef9
--- /dev/null
+++
b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/scm/net/TestHostAndPort.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdds.scm.net;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for {@link HostAndPort} address refresh (HDDS-15533).
+ */
+public class TestHostAndPort {
+
+ @Test
+ public void resolveLatestReturnsNullWhenIpUnchanged() {
+ HostAndPort address = new HostAndPort("127.0.0.1", 9861);
+ assertNull(address.resolveLatest());
+ }
+
+ @Test
+ public void setAddressRejectsNull() {
+ HostAndPort address = new HostAndPort("127.0.0.1", 9861);
+ assertThrows(NullPointerException.class, () -> address.setAddress(null));
+ }
+
+ @Test
+ public void setAddressDoesNotChangeIdentity() throws Exception {
+ HostAndPort address = new HostAndPort("127.0.0.1", 9861);
+ InetSocketAddress refreshed =
+ new InetSocketAddress(InetAddress.getByAddress(new byte[]{10, 0, 0,
7}), 9861);
+ address.setAddress(refreshed);
+ assertEquals(refreshed, address.getAddress());
+ // equals/hashCode stay keyed on host:port so the instance remains a
stable map key.
+ assertEquals(new HostAndPort("127.0.0.1", 9861), address);
+ assertEquals(new HostAndPort("127.0.0.1", 9861).hashCode(),
address.hashCode());
+ }
+}
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/EndpointStateMachine.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/EndpointStateMachine.java
index b4a9a6e1a26..60bf26a2dc0 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/EndpointStateMachine.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/EndpointStateMachine.java
@@ -152,10 +152,14 @@ public ExecutorService getExecutorService() {
*/
@Override
public void close() {
- if (endPoint != null) {
- endPoint.close();
+ try {
+ if (endPoint != null) {
+ endPoint.close();
+ }
+ } finally {
+ // Always release the executor thread, even if closing the RPC proxy
throws.
+ executorService.shutdown();
}
- executorService.shutdown();
}
/**
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/SCMConnectionManager.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/SCMConnectionManager.java
index 25be82cd041..2252d0c6c9d 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/SCMConnectionManager.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/SCMConnectionManager.java
@@ -22,8 +22,10 @@
import static
org.apache.hadoop.hdds.utils.HddsServerUtil.getScmRpcRetryInterval;
import static
org.apache.hadoop.hdds.utils.HddsServerUtil.getScmRpcTimeOutInMilliseconds;
+import com.google.common.annotations.VisibleForTesting;
import java.io.Closeable;
import java.io.IOException;
+import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
@@ -139,33 +141,7 @@ public void addSCMServer(HostAndPort address,
"Ignoring the request.");
return;
}
-
- Configuration hadoopConfig =
- LegacyHadoopConfigurationSource.asHadoopConfiguration(this.conf);
- RPC.setProtocolEngine(
- hadoopConfig,
- StorageContainerDatanodeProtocolPB.class,
- ProtobufRpcEngine.class);
- long version =
- RPC.getProtocolVersion(StorageContainerDatanodeProtocolPB.class);
-
- RetryPolicy retryPolicy =
- RetryPolicies.retryUpToMaximumCountWithFixedSleep(
- getScmRpcRetryCount(conf), getScmRpcRetryInterval(conf),
- TimeUnit.MILLISECONDS);
-
- StorageContainerDatanodeProtocolPB rpcProxy = RPC.getProtocolProxy(
- StorageContainerDatanodeProtocolPB.class, version,
- address.getAddress(), UserGroupInformation.getCurrentUser(),
hadoopConfig,
- NetUtils.getDefaultSocketFactory(hadoopConfig), getRpcTimeout(),
- retryPolicy).getProxy();
-
- StorageContainerDatanodeProtocolClientSideTranslatorPB rpcClient =
- new StorageContainerDatanodeProtocolClientSideTranslatorPB(
- rpcProxy);
-
- EndpointStateMachine endPoint = new EndpointStateMachine(address,
- rpcClient, this.conf, threadNamePrefix);
+ EndpointStateMachine endPoint = buildScmEndpoint(address,
address.getAddress(), threadNamePrefix);
endPoint.setPassive(false);
scmMachines.put(address, endPoint);
} finally {
@@ -173,6 +149,74 @@ public void addSCMServer(HostAndPort address,
}
}
+ @VisibleForTesting
+ EndpointStateMachine buildScmEndpoint(HostAndPort address, InetSocketAddress
dialAddress,
+ String threadNamePrefix) throws IOException {
+ Configuration hadoopConfig =
+ LegacyHadoopConfigurationSource.asHadoopConfiguration(this.conf);
+ RPC.setProtocolEngine(hadoopConfig,
StorageContainerDatanodeProtocolPB.class,
+ ProtobufRpcEngine.class);
+ long version =
RPC.getProtocolVersion(StorageContainerDatanodeProtocolPB.class);
+ RetryPolicy retryPolicy =
RetryPolicies.retryUpToMaximumCountWithFixedSleep(
+ getScmRpcRetryCount(conf), getScmRpcRetryInterval(conf),
TimeUnit.MILLISECONDS);
+ StorageContainerDatanodeProtocolPB rpcProxy = RPC.getProtocolProxy(
+ StorageContainerDatanodeProtocolPB.class, version,
+ dialAddress, UserGroupInformation.getCurrentUser(), hadoopConfig,
+ NetUtils.getDefaultSocketFactory(hadoopConfig), getRpcTimeout(),
+ retryPolicy).getProxy();
+ StorageContainerDatanodeProtocolClientSideTranslatorPB rpcClient =
+ new StorageContainerDatanodeProtocolClientSideTranslatorPB(rpcProxy);
+ return new EndpointStateMachine(address, rpcClient, this.conf,
threadNamePrefix);
+ }
+
+ /**
+ * Re-resolves the active SCM endpoint at {@code address}; on an IP change,
rebuilds it under the
+ * same key and closes the stale proxy. Returns true if rebuilt.
+ */
+ public boolean refreshSCMServer(HostAndPort address, String threadNamePrefix)
+ throws IOException {
+ final EndpointStateMachine current;
+ readLock();
+ try {
+ current = scmMachines.get(address);
+ if (current == null || current.isPassive()) {
+ return false;
+ }
+ } finally {
+ readUnlock();
+ }
+ // Resolve outside the lock, but commit the new address only after the
replacement is built,
+ // so a build failure or a lost race never leaves the cached address ahead
of the live proxy.
+ final InetSocketAddress latest = address.resolveLatest();
+ if (latest == null) {
+ return false;
+ }
+ final EndpointStateMachine stale;
+ final InetSocketAddress previous;
+ writeLock();
+ try {
+ if (scmMachines.get(address) != current) {
+ return false;
+ }
+ EndpointStateMachine rebuilt = buildScmEndpoint(address, latest,
threadNamePrefix);
+ rebuilt.setPassive(false);
+ previous = address.getAddress();
+ address.setAddress(latest);
+ scmMachines.put(address, rebuilt);
+ stale = current;
+ } finally {
+ writeUnlock();
+ }
+ // The swap is committed; failing to close the stale proxy is
cleanup-only, not a refresh failure.
+ try {
+ stale.close();
+ } catch (RuntimeException e) {
+ LOG.warn("Failed to close stale endpoint for {}", address, e);
+ }
+ LOG.info("SCM endpoint {} re-resolved: {} -> {}",
address.getHostAndPortString(), previous, latest);
+ return true;
+ }
+
/**
* Adds a new Recon server to the set of endpoints.
*
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java
index 7cb24558c7c..6c03d140d84 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java
+++
b/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_MISSED_COUNT_THRESHOLD;
+import static
org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_ADDRESS_REFRESH_MISSED_COUNT_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_MISSED_COUNT_THRESHOLD,
+ HDDS_HEARTBEAT_ADDRESS_REFRESH_MISSED_COUNT_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/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestSCMConnectionManager.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestSCMConnectionManager.java
index 0225b6aa873..ba86a52dd4e 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestSCMConnectionManager.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestSCMConnectionManager.java
@@ -18,7 +18,14 @@
package org.apache.hadoop.ozone.container.common.statemachine;
import static
org.apache.hadoop.ozone.container.common.statemachine.EndpointStateMachine.EndPointStates.HEARTBEAT;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.spy;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.scm.net.HostAndPort;
import org.junit.jupiter.api.Assertions;
@@ -29,6 +36,8 @@
*/
public class TestSCMConnectionManager {
+ private static final InetSocketAddress NEW_IP = newIp();
+
@Test
public void testRemoveSCMServerDoesNotMarkEndpointShutdown()
throws Exception {
@@ -46,4 +55,86 @@ public void testRemoveSCMServerDoesNotMarkEndpointShutdown()
Assertions.assertEquals(HEARTBEAT, endpoint.getState());
}
}
+
+ @Test
+ public void refreshRebuildsEndpointWhenIpChanges() throws Exception {
+ try (SCMConnectionManager cm =
+ new SCMConnectionManager(new OzoneConfiguration())) {
+ final HostAndPort address = spy(new HostAndPort("127.0.0.1", 9861));
+ cm.addSCMServer(address, "");
+ final EndpointStateMachine original = cm.getValues().iterator().next();
+ doReturn(NEW_IP).when(address).resolveLatest();
+
+ Assertions.assertTrue(cm.refreshSCMServer(address, ""));
+ Assertions.assertNotSame(original, cm.getValues().iterator().next());
+ Assertions.assertEquals(NEW_IP, address.getAddress());
+ }
+ }
+
+ @Test
+ public void refreshBuildFailureLeavesEndpointAndAddressUnchanged()
+ throws Exception {
+ try (FailingConnectionManager cm =
+ new FailingConnectionManager(new OzoneConfiguration())) {
+ final HostAndPort address = spy(new HostAndPort("127.0.0.1", 9861));
+ cm.addSCMServer(address, "");
+ final EndpointStateMachine original = cm.getValues().iterator().next();
+ final InetSocketAddress before = address.getAddress();
+ doReturn(NEW_IP).when(address).resolveLatest();
+ cm.failBuild = true;
+
+ // Build fails after DNS returned a new IP: the live endpoint and the
cached address must both
+ // stay unchanged, otherwise the DN dials the stale proxy forever while
getAddress() reports the
+ // new IP -- the "stuck until restart" state this feature removes.
+ Assertions.assertThrows(IOException.class, () ->
cm.refreshSCMServer(address, ""));
+ Assertions.assertSame(original, cm.getValues().iterator().next());
+ Assertions.assertEquals(before, address.getAddress());
+ }
+ }
+
+ @Test
+ public void refreshAbandonedWhenEndpointRemovedDuringResolve() throws
Exception {
+ try (SCMConnectionManager cm =
+ new SCMConnectionManager(new OzoneConfiguration())) {
+ final HostAndPort address = spy(new HostAndPort("127.0.0.1", 9861));
+ cm.addSCMServer(address, "");
+ final InetSocketAddress before = address.getAddress();
+ // resolveLatest runs in the unlocked window between the endpoint
snapshot and the write lock;
+ // removing the endpoint right there exercises the lost-race guard
deterministically.
+ doAnswer(inv -> {
+ cm.removeSCMServer(address);
+ return NEW_IP;
+ }).when(address).resolveLatest();
+
+ Assertions.assertFalse(cm.refreshSCMServer(address, ""));
+ Assertions.assertTrue(cm.getValues().isEmpty());
+ Assertions.assertEquals(before, address.getAddress());
+ }
+ }
+
+ private static InetSocketAddress newIp() {
+ try {
+ return new InetSocketAddress(InetAddress.getByAddress(new byte[]{10, 0,
0, 7}), 9861);
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ /** SCMConnectionManager whose endpoint build can be forced to fail, to
exercise rollback. */
+ private static final class FailingConnectionManager extends
SCMConnectionManager {
+ private boolean failBuild;
+
+ FailingConnectionManager(ConfigurationSource conf) {
+ super(conf);
+ }
+
+ @Override
+ EndpointStateMachine buildScmEndpoint(HostAndPort address,
InetSocketAddress dialAddress,
+ String threadNamePrefix) throws IOException {
+ if (failBuild) {
+ throw new IOException("simulated build failure");
+ }
+ return super.buildScmEndpoint(address, dialAddress, threadNamePrefix);
+ }
+ }
}
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTaskDnsRefresh.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTaskDnsRefresh.java
new file mode 100644
index 00000000000..6d9a4406c5d
--- /dev/null
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/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_MISSED_COUNT_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_MISSED_COUNT_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_MISSED_COUNT_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;
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
