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 8e9d4e9e098b..ee68d422a251 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_THRESHOLD = + "hdds.heartbeat.address.refresh.threshold"; + public static final int HDDS_HEARTBEAT_ADDRESS_REFRESH_THRESHOLD_DEFAULT = 3; public static final String HDDS_NODE_REPORT_INTERVAL = "hdds.node.report.interval"; public static final String HDDS_NODE_REPORT_INTERVAL_DEFAULT = diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/net/HostAndPort.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/net/HostAndPort.java index e89ad73d31b1..2a142ea8a112 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 eec93686a6b6..2e9c6b27618b 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 @@ + + hdds.heartbeat.address.refresh.threshold + 3 + OZONE, DATANODE, HA + 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. + + + ozone.directory.deleting.service.interval 1m 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 000000000000..9be015f5ef92 --- /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 b4a9a6e1a267..60bf26a2dc0f 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 25be82cd0414..2252d0c6c9d6 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 7cb24558c7cb..4364a50883de 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_THRESHOLD; +import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_ADDRESS_REFRESH_THRESHOLD_DEFAULT; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_PIPELINE_ACTION_MAX_LIMIT; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_PIPELINE_ACTION_MAX_LIMIT_DEFAULT; +import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_DEFAULT; +import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY; import static org.apache.hadoop.ozone.container.upgrade.UpgradeUtils.toLayoutVersionProto; import com.google.common.base.Preconditions; @@ -44,6 +48,7 @@ import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMHeartbeatRequestProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMHeartbeatResponseProto; import org.apache.hadoop.hdds.upgrade.HDDSLayoutVersionManager; +import org.apache.hadoop.hdds.utils.ConnectionFailureUtils; import org.apache.hadoop.hdfs.util.EnumCounters; import org.apache.hadoop.ozone.container.common.helpers.DeletedContainerBlocksSummary; import org.apache.hadoop.ozone.container.common.statemachine.EndpointStateMachine; @@ -77,6 +82,8 @@ public class HeartbeatEndpointTask private int maxContainerActionsPerHB; private int maxPipelineActionsPerHB; private HDDSLayoutVersionManager layoutVersionManager; + private final boolean resolveOnFailureEnabled; + private final int refreshThreshold; /** * Constructs a SCM heart beat. @@ -100,6 +107,10 @@ public HeartbeatEndpointTask(EndpointStateMachine rpcEndpoint, } else { this.layoutVersionManager = context.getParent().getLayoutVersionManager(); } + this.resolveOnFailureEnabled = conf.getBoolean(OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY, + OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_DEFAULT); + this.refreshThreshold = Math.max(1, conf.getInt(HDDS_HEARTBEAT_ADDRESS_REFRESH_THRESHOLD, + HDDS_HEARTBEAT_ADDRESS_REFRESH_THRESHOLD_DEFAULT)); } /** @@ -157,12 +168,33 @@ public EndpointStateMachine.EndPointStates call() throws Exception { // put back the reports which failed to be sent putBackIncrementalReports(requestBuilder); rpcEndpoint.logIfNeeded(ex); + maybeRefreshScmAddress(ex); } finally { rpcEndpoint.unlock(); } return rpcEndpoint.getState(); } + /** + * On a connection-class heartbeat failure past the threshold (and when resolve-needed is on), + * asks the connection manager to re-resolve this SCM peer and rebuild the endpoint. + */ + private void maybeRefreshScmAddress(IOException heartbeatFailure) { + if (!resolveOnFailureEnabled + || rpcEndpoint.isPassive() + || rpcEndpoint.getMissedCount() < refreshThreshold + || !ConnectionFailureUtils.isConnectionFailure(heartbeatFailure)) { + return; + } + try { + context.getParent().getConnectionManager() + .refreshSCMServer(rpcEndpoint.getAddress(), context.getThreadNamePrefix()); + } catch (IOException ex) { + LOG.warn("Failed to refresh SCM address {} after {} missed heartbeats", + rpcEndpoint.getAddress(), rpcEndpoint.getMissedCount(), ex); + } + } + // TODO: Make it generic. private void putBackIncrementalReports( SCMHeartbeatRequestProto.Builder requestBuilder) { diff --git a/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 0225b6aa873f..ba86a52dd4e3 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 000000000000..550d5dae969d --- /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_THRESHOLD; +import static org.apache.hadoop.hdds.upgrade.HDDSLayoutVersionManager.maxLayoutVersion; +import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY; +import static org.mockito.Mockito.any; +import static org.mockito.Mockito.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.net.ConnectException; +import java.util.UUID; +import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.hdds.protocol.DatanodeDetails; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto; +import org.apache.hadoop.hdds.scm.net.HostAndPort; +import org.apache.hadoop.hdds.upgrade.HDDSLayoutVersionManager; +import org.apache.hadoop.hdfs.util.EnumCounters; +import org.apache.hadoop.ozone.container.common.statemachine.DatanodeStateMachine; +import org.apache.hadoop.ozone.container.common.statemachine.DatanodeStateMachine.DatanodeStates; +import org.apache.hadoop.ozone.container.common.statemachine.EndpointStateMachine; +import org.apache.hadoop.ozone.container.common.statemachine.SCMConnectionManager; +import org.apache.hadoop.ozone.container.common.statemachine.StateContext; +import org.apache.hadoop.ozone.protocolPB.StorageContainerDatanodeProtocolClientSideTranslatorPB; +import org.junit.jupiter.api.Test; + +/** + * Verifies a connection-class heartbeat failure past the threshold triggers a DNS re-resolution of + * the SCM peer (HDDS-15533); flag-off, application errors, and below-threshold do not. + */ +public class TestHeartbeatEndpointTaskDnsRefresh { + + private static final HostAndPort SCM = new HostAndPort("test-scm-1", 9861); + + @Test + public void connectionFailureAtThresholdTriggersRefresh() throws Exception { + OzoneConfiguration conf = new OzoneConfiguration(); + conf.setBoolean(OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY, true); + conf.setInt(HDDS_HEARTBEAT_ADDRESS_REFRESH_THRESHOLD, 2); + SCMConnectionManager cm = runHeartbeat(conf, 3, new ConnectException("refused")); + verify(cm, times(1)).refreshSCMServer(eq(SCM), any()); + } + + @Test + public void flagOffSuppressesRefresh() throws Exception { + OzoneConfiguration conf = new OzoneConfiguration(); + conf.setBoolean(OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY, false); + SCMConnectionManager cm = runHeartbeat(conf, 5, new ConnectException("refused")); + verify(cm, never()).refreshSCMServer(any(), any()); + } + + @Test + public void applicationErrorDoesNotTriggerRefresh() throws Exception { + OzoneConfiguration conf = new OzoneConfiguration(); + conf.setBoolean(OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY, true); + SCMConnectionManager cm = runHeartbeat(conf, 5, new IOException("application-level")); + verify(cm, never()).refreshSCMServer(any(), any()); + } + + @Test + public void belowThresholdDoesNotTriggerRefresh() throws Exception { + OzoneConfiguration conf = new OzoneConfiguration(); + conf.setBoolean(OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY, true); + conf.setInt(HDDS_HEARTBEAT_ADDRESS_REFRESH_THRESHOLD, 5); + SCMConnectionManager cm = runHeartbeat(conf, 1, new ConnectException("refused")); + verify(cm, never()).refreshSCMServer(any(), any()); + } + + /** + * Drives one heartbeat that fails with {@code failure}, with the endpoint reporting + * {@code missedCount} missed heartbeats, and returns the mocked connection manager to verify. + */ + private SCMConnectionManager runHeartbeat(OzoneConfiguration conf, long missedCount, + IOException failure) throws Exception { + StorageContainerDatanodeProtocolClientSideTranslatorPB proxy = + mock(StorageContainerDatanodeProtocolClientSideTranslatorPB.class); + when(proxy.sendHeartbeat(any())).thenThrow(failure); + + EndpointStateMachine endpoint = mock(EndpointStateMachine.class); + when(endpoint.getEndPoint()).thenReturn(proxy); + when(endpoint.getAddress()).thenReturn(SCM); + when(endpoint.getMissedCount()).thenReturn(missedCount); + when(endpoint.isPassive()).thenReturn(false); + + SCMConnectionManager connectionManager = mock(SCMConnectionManager.class); + DatanodeStateMachine dsm = mock(DatanodeStateMachine.class); + when(dsm.getConnectionManager()).thenReturn(connectionManager); + when(dsm.getQueuedCommandCount()) + .thenReturn(new EnumCounters<>(SCMCommandProto.Type.class)); + StateContext context = new StateContext(conf, DatanodeStates.RUNNING, dsm, ""); + + HDDSLayoutVersionManager lvm = mock(HDDSLayoutVersionManager.class); + when(lvm.getSoftwareLayoutVersion()).thenReturn(maxLayoutVersion()); + when(lvm.getMetadataLayoutVersion()).thenReturn(maxLayoutVersion()); + + DatanodeDetails dn = DatanodeDetails.newBuilder() + .setUuid(UUID.randomUUID()) + .setHostName("localhost") + .setIpAddress("127.0.0.1") + .build(); + + HeartbeatEndpointTask.newBuilder() + .setConfig(conf) + .setDatanodeDetails(dn) + .setContext(context) + .setLayoutVersionManager(lvm) + .setEndpointStateMachine(endpoint) + .build() + .call(); + return connectionManager; + } +}