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; + } + } +}