From ab99b876ddbd670de01fc93c80346f161edf9100 Mon Sep 17 00:00:00 2001 From: Di Wang Date: Tue, 13 Oct 2026 14:23:57 +0800 Subject: [PATCH] Fix ConfigNode client failover with stale candidates 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 +- .../client/AbstractConfigNodeClient.java | 12 +- ...bstractConfigNodeClientConnectionTest.java | 340 ++++++++++++++++++ .../client/AbstractConfigNodeClientTest.java | 312 ++++++++++++++++ 4 files changed, 672 insertions(+), 13 deletions(-) create mode 100644 iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/client/AbstractConfigNodeClientConnectionTest.java create mode 100644 iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/client/AbstractConfigNodeClientTest.java 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 570432bcb9fe..13dd39c7a318 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 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 @@ /** 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 void tearDown() { EnvFactory.getEnv().cleanClusterEnvironment(); } - // @Test + @Test public void testGracefulLeaderShutdownPreservesStoppedAndRemoving() throws Exception { init(3); int stopped = stoppedFixture(0); @@ -114,7 +113,7 @@ public void testGracefulLeaderShutdownPreservesStoppedAndRemoving() throws Excep assertStatus(1, removing, NodeStatus.Removing); } - // @Test + @Test public void testGracefulSeedShutdownReportsStopped() throws Exception { init(1); transfer(1); @@ -137,7 +136,7 @@ public void testGracefulSeedShutdownReportsStopped() throws Exception { awaitPersisted(2, seedId, null); } - // @Test + @Test public void testForcedLeaderShutdownWithTwoDataNodes() throws Exception { init(2); int stopped = stoppedFixture(0); @@ -157,7 +156,7 @@ public void testForcedLeaderShutdownWithTwoDataNodes() throws Exception { assertStatus(0, stopped, NodeStatus.Stopped); } - // @Test + @Test public void testNewLeaderChangesSurviveReturnToOriginalLeader() throws Exception { init(3); int stopped = stoppedFixture(0); @@ -181,7 +180,7 @@ public void testNewLeaderChangesSurviveReturnToOriginalLeader() throws Exception assertStatus(1, removing, NodeStatus.Unknown); } - // @Test + @Test public void testReadOnlyReasonPersistsAcrossLeaderTransferAndClearsAfterRestart() throws Exception { init(2); @@ -230,7 +229,7 @@ public void testReadOnlyReasonPersistsAcrossLeaderTransferAndClearsAfterRestart( } } - // @Test + @Test public void testLiveOldLeaderPartitionAndQuorumRecovery() throws Exception { init(3); int stopped = stoppedFixture(0); @@ -299,7 +298,7 @@ private void init(int dataNodes) throws Exception { } } - // @Test + @Test public void testNoQuorumTransitionMatrix() throws Exception { init(2); int nodeId = stoppedFixture(0); @@ -362,7 +361,7 @@ public void testNoQuorumTransitionMatrix() throws Exception { } } - // @Test + @Test public void testSnapshotAndFollowingClearLogRecoverTogether() throws Exception { init(3); int stopped = stoppedFixture(0); @@ -458,7 +457,7 @@ private void awaitConfigNodeRunning(int index) { 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 86bc416653e2..2d20166fbb56 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 @@ private void tryToConnect(int timeoutMs) throws TException { } } + // 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 @@ private void tryToConnect(int timeoutMs) throws TException { } public void syncLatestConfigNodeList() { - configNodes = getConfigNodeInfo().getLatestConfigNodes(); - cursor = 0; + final List 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 000000000000..6b99634c3ae6 --- /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 nodes = Collections.singletonList(server.endpoint()); + when(nodeInfo.getLatestConfigNodes()).thenReturn(nodes); + ClientManager manager = newManager(nodes, nodeInfo); + TestConfigNodeClient client = null; + try { + client = manager.borrowClient(AbstractConfigNodeInfo.CONFIG_REGION_ID); + TTransport transport = client.getTransport(); + List 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 clients = new ArrayList<>(); + List allTransports = new ArrayList<>(); + try (SocketAcceptor first = new SocketAcceptor(); + SocketAcceptor second = new SocketAcceptor()) { + AtomicReference> 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 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> 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 transports, int expected) { + assertEquals(expected, transports.stream().filter(TTransport::isOpen).count()); + } + + private static ClientManager newManager( + List nodes, AbstractConfigNodeInfo nodeInfo) { + return new ClientManager<>( + manager -> + new GenericKeyedObjectPool<>( + new BaseKeyedPooledObjectFactory() { + @Override + public TestConfigNodeClient create(ConfigRegionId key) { + return mock( + TestConfigNodeClient.class, + withSettings() + .useConstructor(nodes, nodeInfo, manager) + .defaultAnswer(CALLS_REAL_METHODS)); + } + + @Override + public PooledObject wrap(TestConfigNodeClient client) { + return new DefaultPooledObject<>(client); + } + + @Override + public void destroyObject( + ConfigRegionId key, PooledObject client) { + client.getObject().invalidate(); + } + })); + } + + private static void assertPeerConnectionsClosed(List 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 connections = new LinkedBlockingQueue<>(); + private final List 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 awaitConnections(int count) throws InterruptedException { + List 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 { + + private final AbstractConfigNodeInfo nodeInfo; + + TestConfigNodeClient(List nodes, AbstractConfigNodeInfo nodeInfo) throws TException { + this(nodes, nodeInfo, null); + } + + TestConfigNodeClient( + List nodes, + AbstractConfigNodeInfo nodeInfo, + ClientManager 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 000000000000..53c8cd4e0b0b --- /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 expectedNodes = Arrays.asList(FIRST, THIRD); + List 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 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 latestNodes) throws TException { + AbstractConfigNodeInfo nodeInfo = mock(AbstractConfigNodeInfo.class); + when(nodeInfo.getLatestConfigNodes()).thenReturn(latestNodes); + return newClient(Collections.singletonList(FIRST), nodeInfo); + } + + private static TestConfigNodeClient newClient( + List nodes, AbstractConfigNodeInfo nodeInfo) { + return mock( + TestConfigNodeClient.class, + withSettings().useConstructor(nodes, nodeInfo).defaultAnswer(CALLS_REAL_METHODS)); + } + + private static void executeResponses( + TestConfigNodeClient client, List expectedNodes, List responses) + throws TException { + List 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 { + + private final AbstractConfigNodeInfo nodeInfo; + + TestConfigNodeClient(List 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; + } + } +}