This is an automated email from the ASF dual-hosted git repository.
smengcl 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 7693b1979f8 HDDS-15895. Make ServerNotLeaderException suggested leader
flow IPv6-safe (#10807)
7693b1979f8 is described below
commit 7693b1979f8165ebebcdda33a92d30e061f8a63e
Author: Chi-Hsuan Huang <[email protected]>
AuthorDate: Thu Aug 13 07:07:52 2026 +0800
HDDS-15895. Make ServerNotLeaderException suggested leader flow IPv6-safe
(#10807)
---
.../hdds/ratis/ServerNotLeaderException.java | 5 +-
...TestServerNotLeaderExceptionMessageParsing.java | 22 +++
.../scm/proxy/SCMFailoverProxyProviderBase.java | 95 ++++++++++--
.../TestSCMFailoverProxyProviderRefreshWired.java | 167 ++++++++++++++++++++-
4 files changed, 277 insertions(+), 12 deletions(-)
diff --git
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/ratis/ServerNotLeaderException.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/ratis/ServerNotLeaderException.java
index ecf5333b80b..685b42aa852 100644
---
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/ratis/ServerNotLeaderException.java
+++
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/ratis/ServerNotLeaderException.java
@@ -32,7 +32,7 @@ public class ServerNotLeaderException extends IOException {
private static final Pattern CURRENT_PEER_ID_PATTERN =
Pattern.compile(".* Server:(.*?) is not the leader[.]+.*",
Pattern.DOTALL);
private static final Pattern SUGGESTED_LEADER_PATTERN =
- Pattern.compile(".*Suggested leader is Server:([^:]*)(:[0-9]+).*",
+ Pattern.compile(".*Suggested leader is
Server:(\\[[^\\]]+\\]|[^:]*)(:[0-9]+).*",
Pattern.DOTALL);
public ServerNotLeaderException(RaftPeerId currentPeerId, String hostname,
@@ -99,7 +99,8 @@ public static ServerNotLeaderException
convertToNotLeaderException(
null;
ServerNotLeaderException serverNotLeaderException;
if (suggestedLeader != null) {
- String suggestedLeaderHostPort = suggestedLeader + ":" + port;
+ String suggestedLeaderHostPort =
+ HddsUtils.getHostPortString(suggestedLeader, Integer.parseInt(port));
serverNotLeaderException =
new ServerNotLeaderException(currentPeer, suggestedLeaderHostPort,
hostname, roleType);
} else {
diff --git
a/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/ratis/TestServerNotLeaderExceptionMessageParsing.java
b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/ratis/TestServerNotLeaderExceptionMessageParsing.java
index f19fa67a9c5..07203529261 100644
---
a/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/ratis/TestServerNotLeaderExceptionMessageParsing.java
+++
b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/ratis/TestServerNotLeaderExceptionMessageParsing.java
@@ -19,7 +19,12 @@
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import org.apache.ratis.protocol.RaftPeer;
+import org.apache.ratis.protocol.RaftPeerId;
+import org.apache.ratis.protocol.exceptions.NotLeaderException;
import org.junit.jupiter.api.Test;
/** Class to test {@link ServerNotLeaderException} parsing. **/
@@ -75,4 +80,21 @@ public void testServerNotLeaderException() {
assertNull(snle.getSuggestedLeader());
}
+ @Test
+ public void testIpv6SuggestedLeader() {
+ RaftPeer suggestedLeader = RaftPeer.newBuilder()
+ .setId("scm2")
+ .setAddress("[2001:db8::1]:9894")
+ .build();
+ NotLeaderException notLeaderException = mock(NotLeaderException.class);
+ when(notLeaderException.getSuggestedLeader()).thenReturn(suggestedLeader);
+
+ ServerNotLeaderException converted =
ServerNotLeaderException.convertToNotLeaderException(
+ notLeaderException, RaftPeerId.valueOf("scm1"), "9863", "2001:db8::2",
"SCM");
+ assertEquals("[2001:db8::1]:9863", converted.getSuggestedLeader());
+
+ ServerNotLeaderException parsed = new
ServerNotLeaderException(converted.getMessage());
+ assertEquals("[2001:db8::1]:9863", parsed.getSuggestedLeader());
+ }
+
}
diff --git
a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/proxy/SCMFailoverProxyProviderBase.java
b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/proxy/SCMFailoverProxyProviderBase.java
index 4daf3144261..05bcbf6bffa 100644
---
a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/proxy/SCMFailoverProxyProviderBase.java
+++
b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/proxy/SCMFailoverProxyProviderBase.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.hdds.scm.proxy;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.net.InetAddresses;
import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
@@ -29,8 +30,10 @@
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.OptionalInt;
import java.util.stream.Collectors;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.HddsUtils;
import org.apache.hadoop.hdds.conf.ConfigurationException;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.ratis.ServerNotLeaderException;
@@ -249,24 +252,98 @@ public synchronized void
performFailoverToAssignedLeader(String newLeader,
Exception e) {
ServerNotLeaderException snle =
(ServerNotLeaderException) SCMHAUtils.getServerNotLeaderException(e);
- if (snle != null && snle.getSuggestedLeader() != null) {
- Optional<SCMProxyInfo> matchedProxyInfo =
- scmProxyInfoMap.values().stream().filter(
- proxyInfo -> NetUtils.getHostPortString(proxyInfo.getAddress())
- .equals(snle.getSuggestedLeader())).findFirst();
- if (matchedProxyInfo.isPresent()) {
- newLeader = matchedProxyInfo.get().getNodeId();
+ String suggestedLeader = snle != null ? snle.getSuggestedLeader() : null;
+ if (suggestedLeader != null) {
+ Optional<String> matchedNodeId =
findSuggestedLeaderNodeId(suggestedLeader);
+ if (matchedNodeId.isPresent()) {
+ newLeader = matchedNodeId.get();
getLogger().debug("Performing failover to suggested leader {}, nodeId
{}",
- snle.getSuggestedLeader(), newLeader);
+ suggestedLeader, newLeader);
} else {
getLogger().debug("Suggested leader {} does not match with any of the
" +
- "proxyInfo address {}", snle.getSuggestedLeader(),
+ "proxyInfo address {}", suggestedLeader,
Arrays.toString(scmProxyInfoMap.values().toArray()));
}
}
assignLeaderToNode(newLeader);
}
+ /**
+ * Find the SCM nodeId whose address matches the suggested-leader authority.
Callers hold
+ * the provider monitor, so this must never resolve a name.
+ */
+ private Optional<String> findSuggestedLeaderNodeId(String suggestedLeader) {
+ final Optional<String> host;
+ final OptionalInt port;
+ try {
+ host = HddsUtils.getHostName(suggestedLeader);
+ port = HddsUtils.getHostPort(suggestedLeader);
+ } catch (IllegalArgumentException ex) {
+ getLogger().warn("Ignoring unparseable suggested leader {}",
suggestedLeader, ex);
+ return Optional.empty();
+ }
+ if (!host.isPresent() || !port.isPresent()) {
+ return Optional.empty();
+ }
+ return scmProxyInfoMap.values().stream()
+ .filter(proxyInfo -> matchesAuthority(proxyInfo.getAddress(),
host.get(), port.getAsInt()))
+ .findFirst()
+ .map(SCMProxyInfo::getNodeId);
+ }
+
+ private static boolean matchesAuthority(InetSocketAddress address, String
host, int port) {
+ if (address.getPort() != port) {
+ return false;
+ }
+ // Both getHostString() and getAddress() read what was resolved when the
proxy info was
+ // built, so neither triggers a lookup here.
+ String addressHost = address.getHostString();
+ if (sameHost(addressHost, host)) {
+ return true;
+ }
+ InetAddress resolved = address.getAddress();
+ return resolved != null && sameIpLiteral(resolved.getHostAddress(), host);
+ }
+
+ /**
+ * Whether two authority hosts denote the same SCM. Names are matched
case-insensitively as DNS
+ * requires, while IP literals go through {@link #sameIpLiteral} so that an
IPv6 zone keeps its
+ * case: eth0 and ETH0 are different interfaces.
+ */
+ private static boolean sameHost(String addressHost, String host) {
+ if (isIpLiteral(addressHost) || isIpLiteral(host)) {
+ return sameIpLiteral(addressHost, host);
+ }
+ return addressHost.equalsIgnoreCase(host);
+ }
+
+ /**
+ * Whether two IP literals denote the same address, allowing for different
spellings of one
+ * IPv6 address. Scope is compared as text, unlike {@link
InetAddress#equals} which drops it,
+ * so fe80::1%1 and fe80::1%2 differ, and so do %2 and %eth0 for the same
interface.
+ */
+ private static boolean sameIpLiteral(String host, String otherHost) {
+ String ip = stripScope(host);
+ String otherIp = stripScope(otherHost);
+ return scopeOf(host).equals(scopeOf(otherHost))
+ && InetAddresses.isInetAddress(ip) &&
InetAddresses.isInetAddress(otherIp)
+ &&
InetAddresses.forString(ip).equals(InetAddresses.forString(otherIp));
+ }
+
+ private static boolean isIpLiteral(String host) {
+ return InetAddresses.isInetAddress(stripScope(host));
+ }
+
+ private static String stripScope(String host) {
+ int mark = host.indexOf('%');
+ return mark < 0 ? host : host.substring(0, mark);
+ }
+
+ private static String scopeOf(String host) {
+ int mark = host.indexOf('%');
+ return mark < 0 ? "" : host.substring(mark + 1);
+ }
+
@Override
public Class<T> getInterface() {
return protocolClass;
diff --git
a/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/scm/proxy/TestSCMFailoverProxyProviderRefreshWired.java
b/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/scm/proxy/TestSCMFailoverProxyProviderRefreshWired.java
index e082ac0f7db..83a08a7d8df 100644
---
a/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/scm/proxy/TestSCMFailoverProxyProviderRefreshWired.java
+++
b/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/scm/proxy/TestSCMFailoverProxyProviderRefreshWired.java
@@ -21,16 +21,22 @@
import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_NODES_KEY;
import static
org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_SERVICE_IDS_KEY;
import static
org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import java.io.IOException;
import java.net.ConnectException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
import java.net.SocketTimeoutException;
+import java.util.ArrayList;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.ratis.ServerNotLeaderException;
import org.apache.hadoop.io.retry.RetryPolicy;
import org.apache.hadoop.ozone.ha.ConfUtils;
+import org.apache.ozone.test.GenericTestUtils.LogCapturer;
+import org.apache.ratis.protocol.RaftPeerId;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -41,13 +47,15 @@
* Complements {@code TestConnectionFailureUtils} (helper-in-isolation)
* and {@code TestSCMFailoverProxyProviderRefresh} (per-instance refresh)
* by exercising the actual retry policy whose return value drives the
- * RetryInvocationHandler in production.
+ * RetryInvocationHandler in production. It also verifies wired
+ * suggested-leader failover target selection.
*/
public class TestSCMFailoverProxyProviderRefreshWired {
private static final String SCM_SERVICE_ID = "scmservice";
private static final String SCM_NODE_1 = "scm1";
private static final String SCM_NODE_2 = "scm2";
+ private static final String SCM_NODE_3 = "scm3";
private OzoneConfiguration conf;
@@ -121,6 +129,163 @@ public void
testApplicationLevelErrorDoesNotTriggerRefresh() throws Exception {
"ServerNotLeaderException is application-level; refresh must NOT
fire");
}
+ @Test
+ public void testFailoverToIpv6SuggestedLeader() {
+ SCMBlockLocationFailoverProxyProvider provider = newThreeNodeProvider();
+ SCMProxyInfo leaderProxy = proxyInfoOf(provider, SCM_NODE_3);
+ int port = leaderProxy.getAddress().getPort();
+ provider.replaceProxyInfoForTest(SCM_NODE_3,
+ new SCMProxyInfo(leaderProxy.getServiceId(), SCM_NODE_3,
+ new InetSocketAddress("2001:db8::1", port)));
+
+ // The cached address expands to 2001:db8:0:0:0:0:0:1, so the compressed
hint only
+ // matches through the canonical comparison.
+ failoverWith(provider, notLeader("[2001:db8::1]:" + port));
+
+ assertEquals(SCM_NODE_3, provider.getCurrentProxySCMNodeId(),
+ "suggested leader must override the next round-robin SCM");
+ }
+
+ @Test
+ public void testFailoverToIpv4SuggestedLeader() {
+ SCMBlockLocationFailoverProxyProvider provider = newThreeNodeProvider();
+ SCMProxyInfo leaderProxy = proxyInfoOf(provider, SCM_NODE_3);
+ // Give scm3 a distinct IPv4 address so its authority is unambiguous among
the
+ // localhost nodes.
+ int port = leaderProxy.getAddress().getPort();
+ provider.replaceProxyInfoForTest(SCM_NODE_3,
+ new SCMProxyInfo(leaderProxy.getServiceId(), SCM_NODE_3,
+ new InetSocketAddress("127.0.0.2", port)));
+
+ failoverWith(provider, notLeader("127.0.0.2:" + port));
+
+ assertEquals(SCM_NODE_3, provider.getCurrentProxySCMNodeId(),
+ "an IPv4 suggested leader must keep matching");
+ }
+
+ @Test
+ public void testFailoverToSuggestedLeaderMatchesWithoutResolution() {
+ SCMBlockLocationFailoverProxyProvider provider = newThreeNodeProvider();
+ SCMProxyInfo leaderProxy = proxyInfoOf(provider, SCM_NODE_3);
+ // An address that was never resolved: matching it can only be done on the
host text,
+ // and getAddress() is null.
+ int port = leaderProxy.getAddress().getPort();
+ provider.replaceProxyInfoForTest(SCM_NODE_3,
+ new SCMProxyInfo(leaderProxy.getServiceId(), SCM_NODE_3,
+ InetSocketAddress.createUnresolved("scm3.example.com", port)));
+
+ failoverWith(provider, notLeader("scm3.example.com:" + port));
+
+ assertEquals(SCM_NODE_3, provider.getCurrentProxySCMNodeId(),
+ "suggested leader must match by authority without DNS resolution");
+ }
+
+ @Test
+ public void testSuggestedLeaderDistinguishesIpv6ScopeIds() {
+ SCMBlockLocationFailoverProxyProvider provider = newThreeNodeProvider();
+ // Every node gets the same IPv6 address and port; only the scope tells
scm3 apart, so a
+ // scope-blind comparison matches one of the other two whatever the
iteration order.
+ int port = proxyInfoOf(provider, SCM_NODE_3).getAddress().getPort();
+ for (SCMProxyInfo proxyInfo : new
ArrayList<>(provider.getSCMProxyInfoList())) {
+ String host = SCM_NODE_3.equals(proxyInfo.getNodeId()) ? "fe80::1%2" :
"fe80::1%1";
+ provider.replaceProxyInfoForTest(proxyInfo.getNodeId(),
+ new SCMProxyInfo(proxyInfo.getServiceId(), proxyInfo.getNodeId(),
+ new InetSocketAddress(host, port)));
+ }
+
+ failoverWith(provider, notLeader("[fe80::1%2]:" + port));
+
+ assertEquals(SCM_NODE_3, provider.getCurrentProxySCMNodeId(),
+ "a differing IPv6 scope must not be treated as the same SCM");
+ }
+
+ @Test
+ public void testSuggestedLeaderDistinguishesIpv6ZoneNameCase() {
+ SCMBlockLocationFailoverProxyProvider provider = newThreeNodeProvider();
+ // As above, but the scope is an interface name. Interface names are
case-sensitive, so a
+ // case-insensitive host comparison matches one of the ETH0 nodes instead
of scm3.
+ int port = proxyInfoOf(provider, SCM_NODE_3).getAddress().getPort();
+ for (SCMProxyInfo proxyInfo : new
ArrayList<>(provider.getSCMProxyInfoList())) {
+ String host = SCM_NODE_3.equals(proxyInfo.getNodeId()) ? "fe80::1%eth0"
: "fe80::1%ETH0";
+ provider.replaceProxyInfoForTest(proxyInfo.getNodeId(),
+ new SCMProxyInfo(proxyInfo.getServiceId(), proxyInfo.getNodeId(),
+ InetSocketAddress.createUnresolved(host, port)));
+ }
+
+ failoverWith(provider, notLeader("[fe80::1%eth0]:" + port));
+
+ assertEquals(SCM_NODE_3, provider.getCurrentProxySCMNodeId(),
+ "a differing IPv6 zone name must not be treated as the same SCM");
+ }
+
+ @Test
+ public void testSuggestedLeaderMatchesResolvedAddressOfHostname() throws
Exception {
+ SCMBlockLocationFailoverProxyProvider provider = newThreeNodeProvider();
+ SCMProxyInfo leaderProxy = proxyInfoOf(provider, SCM_NODE_3);
+ int port = leaderProxy.getAddress().getPort();
+ // scm3 is configured by hostname but already resolved to an IPv6 address,
which is what
+ // Ratis reports in the hint. getByAddress attaches the hostname without
any lookup.
+ InetAddress resolved = InetAddress.getByAddress("scm3.example.com",
+ InetAddress.getByName("2001:db8::1").getAddress());
+ provider.replaceProxyInfoForTest(SCM_NODE_3,
+ new SCMProxyInfo(leaderProxy.getServiceId(), SCM_NODE_3,
+ new InetSocketAddress(resolved, port)));
+
+ failoverWith(provider, notLeader("[2001:db8::1]:" + port));
+
+ assertEquals(SCM_NODE_3, provider.getCurrentProxySCMNodeId(),
+ "a hint carrying the resolved IP must match a hostname-configured
SCM");
+ }
+
+ @Test
+ public void testUnparseableSuggestedLeaderKeepsRoundRobin() {
+ SCMBlockLocationFailoverProxyProvider provider = newThreeNodeProvider();
+ int port = proxyInfoOf(provider, SCM_NODE_3).getAddress().getPort();
+ // The message parser's non-bracketed alternative accepts a malformed
authority such as
+ // [a]b:9863, which no longer parses as host and port on the client side.
+ String malformed = "[a]b:" + port;
+ ServerNotLeaderException parsed =
+ new ServerNotLeaderException(notLeader(malformed).getMessage());
+ assertEquals(malformed, parsed.getSuggestedLeader());
+
+ LogCapturer log =
LogCapturer.captureLogs(SCMBlockLocationFailoverProxyProvider.class);
+ try {
+ failoverWith(provider, parsed);
+ } finally {
+ log.stopCapturing();
+ }
+
+ assertThat(log.getOutput()).contains("Ignoring unparseable suggested
leader " + malformed);
+ assertEquals(SCM_NODE_2, provider.getCurrentProxySCMNodeId(),
+ "an unparseable suggested leader must fall back to round-robin
failover");
+ }
+
+ private SCMBlockLocationFailoverProxyProvider newThreeNodeProvider() {
+ conf.set(OZONE_SCM_NODES_KEY + "." + SCM_SERVICE_ID,
+ SCM_NODE_1 + "," + SCM_NODE_2 + "," + SCM_NODE_3);
+ conf.set(ConfUtils.addKeySuffixes(OZONE_SCM_ADDRESS_KEY,
+ SCM_SERVICE_ID, SCM_NODE_3), "localhost");
+ return new SCMBlockLocationFailoverProxyProvider(conf);
+ }
+
+ private static SCMProxyInfo proxyInfoOf(
+ SCMBlockLocationFailoverProxyProvider provider, String nodeId) {
+ return provider.getSCMProxyInfoList().stream()
+ .filter(proxyInfo -> nodeId.equals(proxyInfo.getNodeId()))
+ .findFirst().get();
+ }
+
+ private static ServerNotLeaderException notLeader(String suggestedLeader) {
+ return new ServerNotLeaderException(RaftPeerId.valueOf(SCM_NODE_1),
+ suggestedLeader, "localhost", "SCM");
+ }
+
+ private static void failoverWith(SCMBlockLocationFailoverProxyProvider
provider,
+ ServerNotLeaderException e) {
+ provider.performFailoverToAssignedLeader(null, e);
+ provider.performFailover(null);
+ }
+
@Test
public void testFlagDisabledSuppressesRefresh() throws Exception {
conf.setBoolean(OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY, false);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
