This is an automated email from the ASF dual-hosted git repository.
CRZbulabula pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new bacbd5a6dc4 Fix ConfigNode client failover with stale candidates
(#18847)
bacbd5a6dc4 is described below
commit bacbd5a6dc4b677cfe0b84b4f8ab3065c69ffc13
Author: d-wang-commit <[email protected]>
AuthorDate: Sat Oct 17 10:20:27 2026 +0800
Fix ConfigNode client failover with stale candidates (#18847)
Refresh candidates before fallback selection and resume after the current
endpoint.
Preserve candidates and cursor on empty refreshes.
Add routing and connection tests and re-enable node status failover IT.
---
.../it/cluster/IoTDBNodeStatusFailoverIT.java | 21 +-
.../commons/client/AbstractConfigNodeClient.java | 12 +-
.../AbstractConfigNodeClientConnectionTest.java | 340 +++++++++++++++++++++
.../client/AbstractConfigNodeClientTest.java | 312 +++++++++++++++++++
4 files changed, 672 insertions(+), 13 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/IoTDBNodeStatusFailoverIT.java
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/IoTDBNodeStatusFailoverIT.java
index 570432bcb9f..13dd39c7a31 100644
---
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/IoTDBNodeStatusFailoverIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/IoTDBNodeStatusFailoverIT.java
@@ -37,8 +37,8 @@ import org.apache.iotdb.rpc.TSStatusCode;
import com.sun.tools.attach.VirtualMachine;
import org.apache.tsfile.utils.Pair;
import org.junit.After;
-import org.junit.Ignore;
import org.junit.Rule;
+import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.rules.ErrorCollector;
import org.junit.rules.TemporaryFolder;
@@ -77,7 +77,6 @@ import static org.junit.Assert.fail;
/** Process-level status persistence checks with three actual ConfigRegion
voters. */
@RunWith(IoTDBTestRunner.class)
@Category(ClusterIT.class)
-@Ignore("Temporarily skip IoTDBNodeStatusFailoverIT")
public class IoTDBNodeStatusFailoverIT {
@Rule public TemporaryFolder temporary = new TemporaryFolder();
@Rule public ErrorCollector failures = new ErrorCollector();
@@ -89,7 +88,7 @@ public class IoTDBNodeStatusFailoverIT {
EnvFactory.getEnv().cleanClusterEnvironment();
}
- // @Test
+ @Test
public void testGracefulLeaderShutdownPreservesStoppedAndRemoving() throws
Exception {
init(3);
int stopped = stoppedFixture(0);
@@ -114,7 +113,7 @@ public class IoTDBNodeStatusFailoverIT {
assertStatus(1, removing, NodeStatus.Removing);
}
- // @Test
+ @Test
public void testGracefulSeedShutdownReportsStopped() throws Exception {
init(1);
transfer(1);
@@ -137,7 +136,7 @@ public class IoTDBNodeStatusFailoverIT {
awaitPersisted(2, seedId, null);
}
- // @Test
+ @Test
public void testForcedLeaderShutdownWithTwoDataNodes() throws Exception {
init(2);
int stopped = stoppedFixture(0);
@@ -157,7 +156,7 @@ public class IoTDBNodeStatusFailoverIT {
assertStatus(0, stopped, NodeStatus.Stopped);
}
- // @Test
+ @Test
public void testNewLeaderChangesSurviveReturnToOriginalLeader() throws
Exception {
init(3);
int stopped = stoppedFixture(0);
@@ -181,7 +180,7 @@ public class IoTDBNodeStatusFailoverIT {
assertStatus(1, removing, NodeStatus.Unknown);
}
- // @Test
+ @Test
public void
testReadOnlyReasonPersistsAcrossLeaderTransferAndClearsAfterRestart()
throws Exception {
init(2);
@@ -230,7 +229,7 @@ public class IoTDBNodeStatusFailoverIT {
}
}
- // @Test
+ @Test
public void testLiveOldLeaderPartitionAndQuorumRecovery() throws Exception {
init(3);
int stopped = stoppedFixture(0);
@@ -299,7 +298,7 @@ public class IoTDBNodeStatusFailoverIT {
}
}
- // @Test
+ @Test
public void testNoQuorumTransitionMatrix() throws Exception {
init(2);
int nodeId = stoppedFixture(0);
@@ -362,7 +361,7 @@ public class IoTDBNodeStatusFailoverIT {
}
}
- // @Test
+ @Test
public void testSnapshotAndFollowingClearLogRecoverTogether() throws
Exception {
init(3);
int stopped = stoppedFixture(0);
@@ -458,7 +457,7 @@ public class IoTDBNodeStatusFailoverIT {
Collections.singletonList(NodeStatus.Running));
}
- // @Test
+ @Test
public void testRealRemovalPersistsIntentAndDeletesRegistration() throws
Exception {
EnvFactory.getEnv()
.getConfig()
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/AbstractConfigNodeClient.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/AbstractConfigNodeClient.java
index 86bc416653e..2d20166fbb5 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/AbstractConfigNodeClient.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/AbstractConfigNodeClient.java
@@ -322,6 +322,11 @@ public abstract class AbstractConfigNodeClient<C extends
AbstractConfigNodeClien
}
}
+ // Refresh stale candidates before retrying after an RPC or connection
failure.
+ syncLatestConfigNodeList();
+ if (configNode != null) {
+ cursor = configNodes.indexOf(configNode);
+ }
for (int tryHostNum = 0; tryHostNum < configNodes.size(); tryHostNum++) {
cursor = (cursor + 1) % configNodes.size();
TEndPoint tryEndpoint = configNodes.get(cursor);
@@ -345,8 +350,11 @@ public abstract class AbstractConfigNodeClient<C extends
AbstractConfigNodeClien
}
public void syncLatestConfigNodeList() {
- configNodes = getConfigNodeInfo().getLatestConfigNodes();
- cursor = 0;
+ final List<TEndPoint> latestConfigNodes =
getConfigNodeInfo().getLatestConfigNodes();
+ if (!latestConfigNodes.isEmpty()) {
+ configNodes = latestConfigNodes;
+ cursor = 0;
+ }
}
@Override
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/client/AbstractConfigNodeClientConnectionTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/client/AbstractConfigNodeClientConnectionTest.java
new file mode 100644
index 00000000000..6b99634c3ae
--- /dev/null
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/client/AbstractConfigNodeClientConnectionTest.java
@@ -0,0 +1,340 @@
+/*
+ * 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.iotdb.commons.client;
+
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.client.property.ThriftClientProperty;
+import org.apache.iotdb.commons.conf.CommonConfig;
+import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.consensus.ConfigRegionId;
+
+import org.apache.commons.pool2.BaseKeyedPooledObjectFactory;
+import org.apache.commons.pool2.PooledObject;
+import org.apache.commons.pool2.impl.DefaultPooledObject;
+import org.apache.commons.pool2.impl.GenericKeyedObjectPool;
+import org.apache.thrift.TException;
+import org.apache.thrift.transport.TTransport;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.SocketException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNotSame;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.withSettings;
+
+public class AbstractConfigNodeClientConnectionTest {
+
+ private static final int CLIENT_COUNT = 100;
+ private static final int RECONNECT_ROUNDS = 3;
+ private static final int TIMEOUT_SECONDS = 30;
+
+ @Test(timeout = 60_000)
+ public void testCloseReturnsConnectionToPoolUntilManagerCloses() throws
Exception {
+ CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig();
+ boolean internalSslEnabled = commonConfig.isEnableInternalSSL();
+ commonConfig.setEnableInternalSSL(false);
+ try (SocketAcceptor server = new SocketAcceptor()) {
+ AbstractConfigNodeInfo nodeInfo = mock(AbstractConfigNodeInfo.class);
+ List<TEndPoint> nodes = Collections.singletonList(server.endpoint());
+ when(nodeInfo.getLatestConfigNodes()).thenReturn(nodes);
+ ClientManager<ConfigRegionId, TestConfigNodeClient> manager =
newManager(nodes, nodeInfo);
+ TestConfigNodeClient client = null;
+ try {
+ client = manager.borrowClient(AbstractConfigNodeInfo.CONFIG_REGION_ID);
+ TTransport transport = client.getTransport();
+ List<Socket> sockets = server.awaitConnections(1);
+ assertTrue(transport.isOpen());
+ assertEquals(1, manager.getPool().getNumActive());
+
+ client.close();
+ assertTrue(transport.isOpen());
+ assertEquals(0, manager.getPool().getNumActive());
+ assertEquals(1, manager.getPool().getNumIdle());
+
+ try (TestConfigNodeClient borrowed =
+ manager.borrowClient(AbstractConfigNodeInfo.CONFIG_REGION_ID)) {
+ assertSame(client, borrowed);
+ assertSame(transport, borrowed.getTransport());
+ assertEquals(1, manager.getPool().getCreatedCount());
+ assertEquals(1, manager.getPool().getNumActive());
+ assertEquals(0, manager.getPool().getNumIdle());
+ }
+ manager.close();
+ assertFalse(transport.isOpen());
+ assertEquals(0, manager.getPool().getNumActive());
+ assertEquals(0, manager.getPool().getNumIdle());
+ assertPeerConnectionsClosed(sockets);
+ } finally {
+ manager.close();
+ if (client != null) {
+ client.invalidate();
+ }
+ }
+ } finally {
+ commonConfig.setEnableInternalSSL(internalSslEnabled);
+ }
+ }
+
+ /** Exercises 100 clients with real TCP transports, not 100 complete
DataNode processes. */
+ @Test(timeout = 120_000)
+ public void testConcurrentReconnectReleasesPreviousConnections() throws
Exception {
+ CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig();
+ boolean internalSslEnabled = commonConfig.isEnableInternalSSL();
+ commonConfig.setEnableInternalSSL(false);
+ ExecutorService workers = Executors.newFixedThreadPool(CLIENT_COUNT);
+ List<TestConfigNodeClient> clients = new ArrayList<>();
+ List<TTransport> allTransports = new ArrayList<>();
+ try (SocketAcceptor first = new SocketAcceptor();
+ SocketAcceptor second = new SocketAcceptor()) {
+ AtomicReference<List<TEndPoint>> latestNodes =
+ new AtomicReference<>(Collections.singletonList(first.endpoint()));
+ AbstractConfigNodeInfo nodeInfo = mock(AbstractConfigNodeInfo.class);
+ when(nodeInfo.getLatestConfigNodes())
+ .thenAnswer(invocation -> new ArrayList<>(latestNodes.get()));
+
+ for (int i = 0; i < CLIENT_COUNT; i++) {
+ TestConfigNodeClient client =
+ mock(
+ TestConfigNodeClient.class,
+ withSettings()
+ .useConstructor(latestNodes.get(), nodeInfo)
+ .defaultAnswer(CALLS_REAL_METHODS));
+ clients.add(client);
+ allTransports.add(client.getTransport());
+ }
+ List<Socket> previousSockets = first.awaitConnections(CLIENT_COUNT);
+ assertOpenTransportCount(allTransports, CLIENT_COUNT);
+
+ for (int round = 0; round < RECONNECT_ROUNDS; round++) {
+ SocketAcceptor target = round % 2 == 0 ? second : first;
+ latestNodes.set(Collections.singletonList(target.endpoint()));
+ CountDownLatch ready = new CountDownLatch(CLIENT_COUNT);
+ CountDownLatch start = new CountDownLatch(1);
+ List<Future<?>> reconnects = new ArrayList<>();
+ try {
+ for (TestConfigNodeClient client : clients) {
+ reconnects.add(
+ workers.submit(
+ () -> {
+ ready.countDown();
+ assertTrue(start.await(TIMEOUT_SECONDS,
TimeUnit.SECONDS));
+ TTransport previousTransport = client.getTransport();
+ client.configLeader = null;
+ // Use real fallback selection and real TCP connection
replacement.
+ client.connectAndSync();
+ assertEquals(target.endpoint(), client.configNode);
+ assertNotSame(previousTransport, client.getTransport());
+ assertFalse(previousTransport.isOpen());
+ assertTrue(client.getTransport().isOpen());
+ return null;
+ }));
+ }
+ assertTrue(ready.await(TIMEOUT_SECONDS, TimeUnit.SECONDS));
+ } finally {
+ start.countDown();
+ }
+ for (Future<?> reconnect : reconnects) {
+ reconnect.get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ }
+ for (TestConfigNodeClient client : clients) {
+ allTransports.add(client.getTransport());
+ }
+ assertOpenTransportCount(allTransports, CLIENT_COUNT);
+ assertPeerConnectionsClosed(previousSockets);
+ previousSockets = target.awaitConnections(CLIENT_COUNT);
+ }
+
+ clients.forEach(TestConfigNodeClient::invalidate);
+ assertOpenTransportCount(allTransports, 0);
+ assertPeerConnectionsClosed(previousSockets);
+ assertEquals(CLIENT_COUNT * (RECONNECT_ROUNDS + 1),
allTransports.size());
+ } finally {
+ workers.shutdownNow();
+ try {
+ assertTrue(workers.awaitTermination(TIMEOUT_SECONDS,
TimeUnit.SECONDS));
+ } finally {
+ clients.forEach(TestConfigNodeClient::invalidate);
+ commonConfig.setEnableInternalSSL(internalSslEnabled);
+ }
+ }
+ }
+
+ private static void assertOpenTransportCount(List<TTransport> transports,
int expected) {
+ assertEquals(expected,
transports.stream().filter(TTransport::isOpen).count());
+ }
+
+ private static ClientManager<ConfigRegionId, TestConfigNodeClient>
newManager(
+ List<TEndPoint> nodes, AbstractConfigNodeInfo nodeInfo) {
+ return new ClientManager<>(
+ manager ->
+ new GenericKeyedObjectPool<>(
+ new BaseKeyedPooledObjectFactory<ConfigRegionId,
TestConfigNodeClient>() {
+ @Override
+ public TestConfigNodeClient create(ConfigRegionId key) {
+ return mock(
+ TestConfigNodeClient.class,
+ withSettings()
+ .useConstructor(nodes, nodeInfo, manager)
+ .defaultAnswer(CALLS_REAL_METHODS));
+ }
+
+ @Override
+ public PooledObject<TestConfigNodeClient>
wrap(TestConfigNodeClient client) {
+ return new DefaultPooledObject<>(client);
+ }
+
+ @Override
+ public void destroyObject(
+ ConfigRegionId key, PooledObject<TestConfigNodeClient>
client) {
+ client.getObject().invalidate();
+ }
+ }));
+ }
+
+ private static void assertPeerConnectionsClosed(List<Socket> sockets) throws
IOException {
+ for (Socket socket : sockets) {
+ // connect() sends no RPC payload: EOF proves the client closed the
actual TCP connection.
+ socket.setSoTimeout((int) TimeUnit.SECONDS.toMillis(TIMEOUT_SECONDS));
+ assertEquals(-1, socket.getInputStream().read());
+ socket.close();
+ }
+ }
+
+ private static final class SocketAcceptor implements AutoCloseable {
+
+ private final ServerSocket server;
+ private final ExecutorService acceptor =
Executors.newSingleThreadExecutor();
+ private final BlockingQueue<Socket> connections = new
LinkedBlockingQueue<>();
+ private final List<Socket> allConnections =
Collections.synchronizedList(new ArrayList<>());
+ private final Future<?> accepting;
+
+ private SocketAcceptor() throws IOException {
+ server = new ServerSocket(0, CLIENT_COUNT * 2,
InetAddress.getByName("127.0.0.1"));
+ accepting =
+ acceptor.submit(
+ () -> {
+ while (!server.isClosed()) {
+ try {
+ Socket socket = server.accept();
+ allConnections.add(socket);
+ connections.add(socket);
+ } catch (SocketException e) {
+ if (!server.isClosed()) {
+ throw e;
+ }
+ }
+ }
+ return null;
+ });
+ }
+
+ private TEndPoint endpoint() {
+ return new TEndPoint("127.0.0.1", server.getLocalPort());
+ }
+
+ private List<Socket> awaitConnections(int count) throws
InterruptedException {
+ List<Socket> result = new ArrayList<>();
+ for (int i = 0; i < count; i++) {
+ Socket socket = connections.poll(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ assertNotNull(socket);
+ result.add(socket);
+ }
+ return result;
+ }
+
+ @Override
+ public void close() throws Exception {
+ try {
+ server.close();
+ accepting.get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ } finally {
+ acceptor.shutdownNow();
+ try {
+ assertTrue(acceptor.awaitTermination(TIMEOUT_SECONDS,
TimeUnit.SECONDS));
+ } finally {
+ for (Socket socket : allConnections) {
+ socket.close();
+ }
+ }
+ }
+ }
+ }
+
+ abstract static class TestConfigNodeClient
+ extends AbstractConfigNodeClient<TestConfigNodeClient> {
+
+ private final AbstractConfigNodeInfo nodeInfo;
+
+ TestConfigNodeClient(List<TEndPoint> nodes, AbstractConfigNodeInfo
nodeInfo) throws TException {
+ this(nodes, nodeInfo, null);
+ }
+
+ TestConfigNodeClient(
+ List<TEndPoint> nodes,
+ AbstractConfigNodeInfo nodeInfo,
+ ClientManager<ConfigRegionId, TestConfigNodeClient> manager)
+ throws TException {
+ super(
+ nodes,
+ new ThriftClientProperty.Builder()
+ .setConnectionTimeoutMs((int)
TimeUnit.SECONDS.toMillis(TIMEOUT_SECONDS))
+ .build(),
+ manager);
+ this.nodeInfo = nodeInfo;
+ }
+
+ @Override
+ protected String getNodeTypeName() {
+ return "DataNode";
+ }
+
+ @Override
+ protected TEndPoint getAddressAndPort() {
+ return new TEndPoint("127.0.0.1", 0);
+ }
+
+ @Override
+ protected AbstractConfigNodeInfo getConfigNodeInfo() {
+ return nodeInfo;
+ }
+ }
+}
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/client/AbstractConfigNodeClientTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/client/AbstractConfigNodeClientTest.java
new file mode 100644
index 00000000000..53c8cd4e0b0
--- /dev/null
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/client/AbstractConfigNodeClientTest.java
@@ -0,0 +1,312 @@
+/*
+ * 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.iotdb.commons.client;
+
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.property.ThriftClientProperty;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.thrift.TException;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.clearInvocations;
+import static org.mockito.Mockito.doThrow;
+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 static org.mockito.Mockito.withSettings;
+
+public class AbstractConfigNodeClientTest {
+
+ private static final TEndPoint FIRST = new TEndPoint("127.0.0.1", 10710);
+ private static final TEndPoint SECOND = new TEndPoint("127.0.0.1", 10720);
+ private static final TEndPoint THIRD = new TEndPoint("127.0.0.1", 10730);
+
+ @Test
+ public void testRefreshStaleListAfterRedirectWithoutLeader() throws
TException {
+ TestConfigNodeClient client = newClient(Arrays.asList(SECOND, FIRST,
THIRD));
+
+ // The latest list has a different order. Continue after the node that
just replied, rather
+ // than restarting at a fixed position that would select FIRST again.
+ executeResponses(
+ client, Arrays.asList(FIRST, THIRD),
Arrays.asList(redirectWithoutLeader(), success()));
+
+ assertEquals(Arrays.asList(SECOND, FIRST, THIRD), client.configNodes);
+ assertNull(client.configLeader);
+ }
+
+ @Test
+ public void testRepeatedRedirectsContinueThroughLatestList() throws
TException {
+ TestConfigNodeClient client = newClient(Arrays.asList(FIRST, SECOND,
THIRD));
+
+ // Refreshing the same list on every response must not reset the cursor
and pin retries to
+ // SECOND. Both non-leaders respond successfully at the transport layer.
+ executeResponses(
+ client,
+ Arrays.asList(FIRST, SECOND, THIRD),
+ Arrays.asList(redirectWithoutLeader(), redirectWithoutLeader(),
success()));
+ }
+
+ @Test
+ public void testResponseCandidateListCanReorderNodesAfterLeaderUpdate()
throws TException {
+ AbstractConfigNodeInfo nodeInfo = mock(AbstractConfigNodeInfo.class);
+ when(nodeInfo.getLatestConfigNodes()).thenReturn(Arrays.asList(FIRST,
SECOND, THIRD));
+ TestConfigNodeClient client = newClient(Collections.singletonList(FIRST),
nodeInfo);
+ List<TEndPoint> expectedNodes = Arrays.asList(FIRST, THIRD);
+ List<TEndPoint> requestedNodes = new ArrayList<>();
+
+ TSStatus result =
+ client.executeRemoteCallWithRetry(
+ () -> {
+ int requestIndex = requestedNodes.size();
+ assertTrue(requestIndex < expectedNodes.size());
+ assertEquals(expectedNodes.get(requestIndex), client.configNode);
+ requestedNodes.add(client.configNode);
+ return requestIndex == 0 ? redirectWithoutLeader() : success();
+ },
+ status -> {
+ if (!client.updateConfigNodeLeader(status)) {
+ return true;
+ }
+ // Registration updates both lists, but the shared set and
response list may have
+ // different iteration orders. Continue after FIRST in the
refreshed shared list.
+
when(nodeInfo.getLatestConfigNodes()).thenReturn(Arrays.asList(SECOND, FIRST,
THIRD));
+ client.configNodes = Arrays.asList(FIRST, THIRD, SECOND);
+ return false;
+ });
+
+ assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
result.getCode());
+ assertEquals(expectedNodes, requestedNodes);
+ }
+
+ @Test
+ public void testExplicitRedirectTakesPriorityOverCandidateList() throws
TException {
+ TestConfigNodeClient client = newClient(Arrays.asList(FIRST, SECOND));
+ TSStatus redirect = redirectWithoutLeader().setRedirectNode(THIRD);
+
+ executeResponses(client, Arrays.asList(FIRST, THIRD),
Arrays.asList(redirect, success()));
+
+ assertEquals(THIRD, client.configLeader);
+ }
+
+ @Test
+ public void testSingleCandidateCanBeRetried() throws TException {
+ TestConfigNodeClient client = newClient(Collections.singletonList(FIRST));
+
+ executeResponses(
+ client, Arrays.asList(FIRST, FIRST),
Arrays.asList(redirectWithoutLeader(), success()));
+ }
+
+ @Test
+ public void testRemovedCurrentNodeStartsAtFirstLatestCandidate() throws
TException {
+ TestConfigNodeClient client = newClient(Arrays.asList(SECOND, THIRD));
+
+ executeResponses(
+ client, Arrays.asList(FIRST, SECOND),
Arrays.asList(redirectWithoutLeader(), success()));
+ }
+
+ @Test
+ public void testEmptyLatestListPreservesKnownCandidates() throws TException {
+ AbstractConfigNodeInfo nodeInfo = mock(AbstractConfigNodeInfo.class);
+ when(nodeInfo.getLatestConfigNodes()).thenReturn(Collections.emptyList());
+ TestConfigNodeClient client = newClient(Arrays.asList(FIRST, SECOND),
nodeInfo);
+
+ executeResponses(
+ client, Arrays.asList(FIRST, SECOND),
Arrays.asList(redirectWithoutLeader(), success()));
+
+ assertEquals(Arrays.asList(FIRST, SECOND), client.configNodes);
+ }
+
+ @Test
+ public void
testRpcFailureRefreshesCandidatesEvenWhenOldNodeAcceptsConnections()
+ throws TException {
+ TestConfigNodeClient client = newClient(Arrays.asList(FIRST, SECOND,
THIRD));
+ List<TEndPoint> requestedNodes = new ArrayList<>();
+
+ TSStatus result =
+ client.executeRemoteCallWithRetry(
+ () -> {
+ requestedNodes.add(client.configNode);
+ if (requestedNodes.size() == 1) {
+ throw new TException("RPC failed while the old endpoint still
accepts connections");
+ }
+ assertEquals(SECOND, client.configNode);
+ return success();
+ },
+ status -> !client.updateConfigNodeLeader(status));
+
+ assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
result.getCode());
+ assertEquals(Arrays.asList(FIRST, SECOND), requestedNodes);
+ }
+
+ @Test
+ public void
testCachedLeaderConnectionFailureRefreshesBeforeTryingOldCandidates()
+ throws TException {
+ AbstractConfigNodeInfo nodeInfo = mock(AbstractConfigNodeInfo.class);
+ when(nodeInfo.getLatestConfigNodes()).thenReturn(Arrays.asList(SECOND,
THIRD));
+ TestConfigNodeClient client = newClient(Arrays.asList(FIRST, THIRD),
nodeInfo);
+ doThrow(new TException("Cached leader is unreachable"))
+ .when(client)
+ .connect(eq(FIRST), anyInt());
+
+ client.connectAndSync();
+
+ assertEquals(SECOND, client.configNode);
+ assertNull(client.configLeader);
+ assertEquals(Arrays.asList(SECOND, THIRD), client.configNodes);
+ verify(client, never()).connect(eq(THIRD), anyInt());
+ }
+
+ @Test
+ public void testUnavailableCandidatesHaveBoundedConnectionAttempts() throws
TException {
+ TestConfigNodeClient client = newClient(Arrays.asList(FIRST, SECOND,
THIRD));
+ client.configLeader = null;
+ clearInvocations(client);
+ doThrow(new TException("Candidate is
unreachable")).when(client).connect(any(), anyInt());
+
+ try {
+ client.connectAndSync();
+ fail("All candidates are unreachable");
+ } catch (TException expected) {
+ // connectAndSync allows two passes, with at most one attempt per
candidate in each pass.
+ verify(client, times(2)).connect(eq(FIRST), anyInt());
+ verify(client, times(2)).connect(eq(SECOND), anyInt());
+ verify(client, times(2)).connect(eq(THIRD), anyInt());
+ verify(client, times(6)).connect(any(), anyInt());
+ }
+ }
+
+ @Test
+ public void testSuccessfulRpcKeepsExistingConnection() throws TException {
+ AbstractConfigNodeInfo nodeInfo = mock(AbstractConfigNodeInfo.class);
+ TestConfigNodeClient client = newClient(Collections.singletonList(FIRST),
nodeInfo);
+ clearInvocations(client);
+
+ executeResponses(
+ client, Collections.singletonList(FIRST),
Collections.singletonList(success()));
+
+ verify(client, never()).connect(any(), anyInt());
+ verify(nodeInfo, never()).getLatestConfigNodes();
+ }
+
+ @Test
+ public void testWarmingUpKeepsKnownLeaderWithoutRefreshingCandidates()
throws TException {
+ AbstractConfigNodeInfo nodeInfo = mock(AbstractConfigNodeInfo.class);
+ TestConfigNodeClient client = newClient(Arrays.asList(FIRST, SECOND),
nodeInfo);
+
+ executeResponses(
+ client,
+ Arrays.asList(FIRST, FIRST),
+ Arrays.asList(
+ new
TSStatus(TSStatusCode.CONFIG_NODE_LEADER_WARMING_UP.getStatusCode()),
success()));
+
+ assertEquals(FIRST, client.configLeader);
+ verify(nodeInfo, never()).getLatestConfigNodes();
+ verify(client, never()).connect(eq(SECOND), anyInt());
+ }
+
+ private static TestConfigNodeClient newClient(List<TEndPoint> latestNodes)
throws TException {
+ AbstractConfigNodeInfo nodeInfo = mock(AbstractConfigNodeInfo.class);
+ when(nodeInfo.getLatestConfigNodes()).thenReturn(latestNodes);
+ return newClient(Collections.singletonList(FIRST), nodeInfo);
+ }
+
+ private static TestConfigNodeClient newClient(
+ List<TEndPoint> nodes, AbstractConfigNodeInfo nodeInfo) {
+ return mock(
+ TestConfigNodeClient.class,
+ withSettings().useConstructor(nodes,
nodeInfo).defaultAnswer(CALLS_REAL_METHODS));
+ }
+
+ private static void executeResponses(
+ TestConfigNodeClient client, List<TEndPoint> expectedNodes,
List<TSStatus> responses)
+ throws TException {
+ List<TEndPoint> requestedNodes = new ArrayList<>();
+ TSStatus result =
+ client.executeRemoteCallWithRetry(
+ () -> {
+ int requestIndex = requestedNodes.size();
+ assertTrue(requestIndex < expectedNodes.size());
+ assertEquals(expectedNodes.get(requestIndex), client.configNode);
+ requestedNodes.add(client.configNode);
+ return responses.get(requestIndex);
+ },
+ status -> !client.updateConfigNodeLeader(status));
+
+ assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
result.getCode());
+ assertEquals(expectedNodes, requestedNodes);
+ }
+
+ private static TSStatus redirectWithoutLeader() {
+ return new TSStatus(TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode());
+ }
+
+ private static TSStatus success() {
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ }
+
+ abstract static class TestConfigNodeClient
+ extends AbstractConfigNodeClient<TestConfigNodeClient> {
+
+ private final AbstractConfigNodeInfo nodeInfo;
+
+ TestConfigNodeClient(List<TEndPoint> nodes, AbstractConfigNodeInfo
nodeInfo) throws TException {
+ super(nodes, new ThriftClientProperty.Builder().build(), null);
+ this.nodeInfo = nodeInfo;
+ }
+
+ @Override
+ public void connect(TEndPoint endpoint, int timeoutMs) throws TException {
+ // Model successful TCP connections while exercising the real selection
and retry logic.
+ configNode = endpoint;
+ }
+
+ @Override
+ protected String getNodeTypeName() {
+ return "DataNode";
+ }
+
+ @Override
+ protected TEndPoint getAddressAndPort() {
+ return new TEndPoint("127.0.0.1", 10740);
+ }
+
+ @Override
+ protected AbstractConfigNodeInfo getConfigNodeInfo() {
+ return nodeInfo;
+ }
+ }
+}
