diff --git a/driver-core/src/test/java/com/datastax/driver/core/CCMBridge.java b/driver-core/src/test/java/com/datastax/driver/core/CCMBridge.java index 344678c8149..1012033ffd4 100644 --- a/driver-core/src/test/java/com/datastax/driver/core/CCMBridge.java +++ b/driver-core/src/test/java/com/datastax/driver/core/CCMBridge.java @@ -430,7 +430,7 @@ protected CCMBridge( this.thriftPort = thriftPort; this.binaryPort = binaryPort; this.isDSE = dseVersion != null; - this.isScylla = (getGlobalScyllaVersion() != null); + this.isScylla = (scyllaVersion != null); this.jvmArgs = jvmArgs; this.nodes = nodes; this.ccmDir = Files.createTempDir(); @@ -733,24 +733,38 @@ public void add(int n) { public void add(int dc, int n) { logger.debug( String.format("Adding: node %s (%s%s:%s) to %s", n, ipPrefix, n, binaryPort, this)); - String thriftItf = ipOfNode(n) + ":" + thriftPort; String storageItf = ipOfNode(n) + ":" + storagePort; String binaryItf = ipOfNode(n) + ":" + binaryPort; String remoteLogItf = ipOfNode(n) + ":" + TestUtils.findAvailablePort(); - execute( - CCM_COMMAND - + " add node%d -d dc%s -i %s%d -t %s -l %s --binary-itf %s -j %d -r %s -s -b" - + (isDSE ? " --dse" : "") - + (isScylla ? " --scylla" : ""), - n, - dc, - ipPrefix, - n, - thriftItf, - storageItf, - binaryItf, - TestUtils.findAvailablePort(), - remoteLogItf); + if (isScylla) { + // scylla-ccm's `add` command has no thrift option: Scylla never had a Thrift interface. + execute( + CCM_COMMAND + + " add node%d -d dc%s -i %s%d -l %s --binary-itf %s -j %d -r %s -s -b --scylla", + n, + dc, + ipPrefix, + n, + storageItf, + binaryItf, + TestUtils.findAvailablePort(), + remoteLogItf); + } else { + String thriftItf = ipOfNode(n) + ":" + thriftPort; + execute( + CCM_COMMAND + + " add node%d -d dc%s -i %s%d -t %s -l %s --binary-itf %s -j %d -r %s -s -b" + + (isDSE ? " --dse" : ""), + n, + dc, + ipPrefix, + n, + thriftItf, + storageItf, + binaryItf, + TestUtils.findAvailablePort(), + remoteLogItf); + } } @Override diff --git a/driver-core/src/test/java/com/datastax/driver/core/TabletsIT.java b/driver-core/src/test/java/com/datastax/driver/core/TabletsTest.java similarity index 93% rename from driver-core/src/test/java/com/datastax/driver/core/TabletsIT.java rename to driver-core/src/test/java/com/datastax/driver/core/TabletsTest.java index f3ecdd362c0..4b4694ed2fe 100644 --- a/driver-core/src/test/java/com/datastax/driver/core/TabletsIT.java +++ b/driver-core/src/test/java/com/datastax/driver/core/TabletsTest.java @@ -24,8 +24,8 @@ }) @ScyllaOnly @ScyllaVersion(minOSS = "6.0.0", minEnterprise = "2024.2", description = "Needs to support tablets") -public class TabletsIT extends CCMTestsSupport { - private static final Logger LOG = LoggerFactory.getLogger(TabletsIT.class); +public class TabletsTest extends CCMTestsSupport { + private static final Logger LOG = LoggerFactory.getLogger(TabletsTest.class); private static final int INITIAL_TABLETS = 32; private static final int QUERIES = 1600; private static final int REPLICATION_FACTOR = 2; @@ -190,8 +190,13 @@ public void every_statement_should_deliver_tablet_info() { continue; } Session session = sessionEntry.getValue().get(); - // Empty out tablets information - session.getCluster().getMetadata().getTabletMap().removeTableMappings(KEYSPACE_NAME); + // Empty out tablets information. The mapping is keyed by the lowercased keyspace name, as + // reported by the server, so the key has to be lowercased here too or this is a no-op. + session + .getCluster() + .getMetadata() + .getTabletMap() + .removeTableMappings(KEYSPACE_NAME.toLowerCase()); Statement stmt; try { stmt = stmtEntry.getValue().apply(session); @@ -226,6 +231,10 @@ public void every_statement_should_deliver_tablet_info() { stmtEntry.getKey(), sessionEntry.getKey())); continue; } + // executeOnAllHostsAndReturnIfResultHasTabletsInfo pins the statement to a specific host + // while hunting for tablet info. Clear that pin, otherwise the routing check below always + // observes the pinned host and can never detect misrouting. + stmt.setHost(null); if (!checkIfRoutedProperly(session, stmt)) { testErrors.add( String.format( @@ -343,6 +352,11 @@ private static boolean checkIfRoutedProperly(Session session, Statement stmt) { int expectedNodesCount = stmt.isLWT() ? 1 : REPLICATION_FACTOR; Set nodes = new HashSet<>(); for (int i = 0; i < REPLICATION_FACTOR * 3; i++) { + // PagingOptimizingLoadBalancingPolicy returns Statement.getLastHost() ahead of the real query + // plan, and that field is set after every successful BoundStatement execution. Clearing it + // keeps the loop from being pinned to the first coordinator, which would let any routing + // behaviour satisfy the check below. + stmt.setLastHost(null); nodes.add(session.execute(stmt).getExecutionInfo().getQueriedHost()); } return nodes.size() <= expectedNodesCount; diff --git a/driver-core/src/test/java/com/datastax/driver/core/ZeroTokenNodesIT.java b/driver-core/src/test/java/com/datastax/driver/core/ZeroTokenNodesTest.java similarity index 96% rename from driver-core/src/test/java/com/datastax/driver/core/ZeroTokenNodesIT.java rename to driver-core/src/test/java/com/datastax/driver/core/ZeroTokenNodesTest.java index ebcf5bdf352..26c0041303e 100644 --- a/driver-core/src/test/java/com/datastax/driver/core/ZeroTokenNodesIT.java +++ b/driver-core/src/test/java/com/datastax/driver/core/ZeroTokenNodesTest.java @@ -14,7 +14,7 @@ import org.testng.annotations.DataProvider; import org.testng.annotations.Test; -public class ZeroTokenNodesIT { +public class ZeroTokenNodesTest { @DataProvider(name = "loadBalancingPolicies") public static Object[][] loadBalancingPolicies() { @@ -173,12 +173,17 @@ public void should_discover_zero_token_DC_when_option_is_enabled( queriedNodes.add(rs.getExecutionInfo().getQueriedHost().getEndPoint().resolve()); } + // containsOnly, not containsExactly: queriedNodes is a HashSet, whose iteration order is + // hash-derived rather than insertion order, so an order-sensitive assertion is a latent + // flake. + // AssertJ 1.7.1, pinned by this module, has no containsExactlyInAnyOrder; for a Set, + // containsOnly is equivalent to it. if (isDcAware) { assertThat(queriedNodes) - .containsExactly(ccmBridge.addressOfNode(1), ccmBridge.addressOfNode(2)); + .containsOnly(ccmBridge.addressOfNode(1), ccmBridge.addressOfNode(2)); } else { assertThat(queriedNodes) - .containsExactly( + .containsOnly( ccmBridge.addressOfNode(1), ccmBridge.addressOfNode(2), ccmBridge.addressOfNode(3), diff --git a/driver-core/src/test/java/com/datastax/driver/core/policies/LWTLoadBalancingIT.java b/driver-core/src/test/java/com/datastax/driver/core/policies/LWTLoadBalancingIT.java deleted file mode 100644 index eed462e1dc6..00000000000 --- a/driver-core/src/test/java/com/datastax/driver/core/policies/LWTLoadBalancingIT.java +++ /dev/null @@ -1,112 +0,0 @@ -/* - * 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 com.datastax.driver.core.policies; - -import static org.assertj.core.api.Assertions.assertThat; - -import com.datastax.driver.core.BoundStatement; -import com.datastax.driver.core.CCMConfig; -import com.datastax.driver.core.CCMTestsSupport; -import com.datastax.driver.core.Cluster; -import com.datastax.driver.core.ConsistencyLevel; -import com.datastax.driver.core.Host; -import com.datastax.driver.core.PreparedStatement; -import com.datastax.driver.core.ResultSet; -import com.datastax.driver.core.Session; -import com.datastax.driver.core.SimpleStatement; -import java.net.InetSocketAddress; -import java.util.HashSet; -import java.util.Set; -import org.testng.annotations.Test; - -/** - * Integration tests verifying that statements with SERIAL/LOCAL_SERIAL consistency level are routed - * through the LWT load-balancing path (PRESERVE_REPLICA_ORDER). - */ -@CCMConfig(numberOfNodes = 3) -public class LWTLoadBalancingIT extends CCMTestsSupport { - - @Override - public Cluster.Builder createClusterBuilder() { - return Cluster.builder() - .withLoadBalancingPolicy( - new TokenAwarePolicy(new RoundRobinPolicy(), TokenAwarePolicy.ReplicaOrdering.RANDOM)); - } - - @Override - public void onTestContextInitialized() { - execute("CREATE TABLE IF NOT EXISTS test_lwt (pk int, ck int, v int, PRIMARY KEY (pk, ck))"); - for (int i = 0; i < 10; i++) { - execute(String.format("INSERT INTO test_lwt (pk, ck, v) VALUES (%d, %d, %d)", i, 0, i)); - } - } - - @Test(groups = "short") - public void should_route_local_serial_select_through_lwt_path() { - Session session = session(); - - SimpleStatement simpleSelect = - new SimpleStatement("SELECT * FROM test_lwt WHERE pk = ? AND ck = ?", 1, 0); - simpleSelect.setConsistencyLevel(ConsistencyLevel.LOCAL_SERIAL); - - PreparedStatement preparedSelect = session.prepare(simpleSelect); - BoundStatement boundSelect = preparedSelect.bind(1, 0); - - // Verify statement properties - assertThat(simpleSelect.isLWT()).isFalse(); - assertThat(simpleSelect.getConsistencyLevel()).isEqualTo(ConsistencyLevel.LOCAL_SERIAL); - - // Execute multiple times and collect coordinators — with PRESERVE_REPLICA_ORDER routing, - // the same partition key should always be routed to the same first replica. - Set coordinators = new HashSet<>(); - for (int i = 0; i < 30; i++) { - ResultSet rs = session.execute(boundSelect); - Host coordinator = rs.getExecutionInfo().getQueriedHost(); - assertThat(coordinator).isNotNull(); - coordinators.add(coordinator.getEndPoint().resolve()); - } - - // With PRESERVE_REPLICA_ORDER, the first replica is deterministic for a given partition key, - // so all 30 executions should hit the same coordinator. - assertThat(coordinators).hasSize(1); - } - - @Test(groups = "short") - public void should_route_serial_select_through_lwt_path() { - Session session = session(); - - SimpleStatement simpleSelect = - new SimpleStatement("SELECT * FROM test_lwt WHERE pk = ? AND ck = ?", 2, 0); - simpleSelect.setConsistencyLevel(ConsistencyLevel.SERIAL); - - PreparedStatement preparedSelect = session.prepare(simpleSelect); - BoundStatement boundSelect = preparedSelect.bind(2, 0); - - // Execute multiple times and collect coordinators - Set coordinators = new HashSet<>(); - for (int i = 0; i < 30; i++) { - ResultSet rs = session.execute(boundSelect); - Host coordinator = rs.getExecutionInfo().getQueriedHost(); - assertThat(coordinator).isNotNull(); - coordinators.add(coordinator.getEndPoint().resolve()); - } - - // With PRESERVE_REPLICA_ORDER, the first replica is deterministic for a given partition key. - assertThat(coordinators).hasSize(1); - } -} diff --git a/driver-core/src/test/java/com/datastax/driver/core/policies/LWTLoadBalancingTest.java b/driver-core/src/test/java/com/datastax/driver/core/policies/LWTLoadBalancingTest.java new file mode 100644 index 00000000000..f9eb78bc544 --- /dev/null +++ b/driver-core/src/test/java/com/datastax/driver/core/policies/LWTLoadBalancingTest.java @@ -0,0 +1,188 @@ +/* + * 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 com.datastax.driver.core.policies; + +import static org.assertj.core.api.Assertions.assertThat; + +import com.datastax.driver.core.BoundStatement; +import com.datastax.driver.core.CCMConfig; +import com.datastax.driver.core.CCMTestsSupport; +import com.datastax.driver.core.Cluster; +import com.datastax.driver.core.ConsistencyLevel; +import com.datastax.driver.core.Host; +import com.datastax.driver.core.PreparedStatement; +import com.datastax.driver.core.ResultSet; +import com.datastax.driver.core.Session; +import com.datastax.driver.core.SimpleStatement; +import com.datastax.driver.core.TestUtils; +import com.google.common.base.Throwables; +import java.net.InetSocketAddress; +import java.util.HashSet; +import java.util.Objects; +import java.util.Set; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testng.annotations.Test; + +/** + * Integration tests verifying that statements with SERIAL/LOCAL_SERIAL consistency level are routed + * through the LWT load-balancing path (PRESERVE_REPLICA_ORDER). + */ +@CCMConfig(numberOfNodes = 3) +public class LWTLoadBalancingTest extends CCMTestsSupport { + + private static final Logger LOGGER = LoggerFactory.getLogger(LWTLoadBalancingTest.class); + + /** Equal to the node count, so that every node is a replica of every partition. */ + private static final int REPLICATION_FACTOR = 3; + + private static final int EXECUTIONS = 30; + + @Override + public Cluster.Builder createClusterBuilder() { + return Cluster.builder() + .withLoadBalancingPolicy( + new TokenAwarePolicy(new RoundRobinPolicy(), TokenAwarePolicy.ReplicaOrdering.RANDOM)); + } + + /** + * Override to create the keyspace with a replication factor greater than 1. The default test + * keyspace created by {@link CCMTestsSupport} is hardcoded to RF=1, and with a single replica per + * partition "the first replica" is trivially unique — every assertion below would hold under + * {@code REGULAR} routing too, so the tests could not tell {@code PRESERVE_REPLICA_ORDER} apart + * from {@code RANDOM}. + * + *

Tablets are disabled when running against Scylla: with tablets enabled, replica placement + * comes from the tablet map, which is empty until it has been learned from a misrouted query, and + * an empty replica list makes the LWT query plan fall back to the child policy — a non-replica + * coordinator on the first execution. Cassandra does not support the tablets property. + */ + @Override + protected void initTestKeyspace() { + try { + keyspace = TestUtils.generateIdentifier("ks_"); + LOGGER.debug("Using keyspace " + keyspace); + boolean isScylla = Objects.nonNull(ccm().getScyllaVersion()); + session() + .execute( + String.format( + "CREATE KEYSPACE %s WITH replication = {'class': 'NetworkTopologyStrategy'," + + " 'datacenter1': %d}" + + (isScylla ? " AND tablets = {'enabled': false}" : ""), + keyspace, + REPLICATION_FACTOR)); + useKeyspace(keyspace); + } catch (Exception e) { + errorOut(); + LOGGER.error("Could not create test keyspace", e); + Throwables.propagate(e); + } + } + + @Override + public void onTestContextInitialized() { + execute("CREATE TABLE IF NOT EXISTS test_lwt (pk int, ck int, v int, PRIMARY KEY (pk, ck))"); + for (int i = 0; i < 10; i++) { + execute(String.format("INSERT INTO test_lwt (pk, ck, v) VALUES (%d, %d, %d)", i, 0, i)); + } + } + + @Test(groups = "short") + public void should_route_local_serial_select_through_lwt_path() { + Session session = session(); + + SimpleStatement simpleSelect = + new SimpleStatement("SELECT * FROM test_lwt WHERE pk = ? AND ck = ?"); + simpleSelect.setConsistencyLevel(ConsistencyLevel.LOCAL_SERIAL); + + PreparedStatement preparedSelect = session.prepare(simpleSelect); + + // Verify statement properties + assertThat(simpleSelect.isLWT()).isFalse(); + assertThat(simpleSelect.getConsistencyLevel()).isEqualTo(ConsistencyLevel.LOCAL_SERIAL); + + // With PRESERVE_REPLICA_ORDER, the first replica is deterministic for a given partition key, + // so every execution should hit the same coordinator. Contrast with the non-serial control in + // should_spread_non_serial_select_across_replicas, which shares the same statement and policy. + assertThat(collectCoordinators(session, preparedSelect, 1)).hasSize(1); + } + + @Test(groups = "short") + public void should_route_serial_select_through_lwt_path() { + Session session = session(); + + SimpleStatement simpleSelect = + new SimpleStatement("SELECT * FROM test_lwt WHERE pk = ? AND ck = ?"); + simpleSelect.setConsistencyLevel(ConsistencyLevel.SERIAL); + + PreparedStatement preparedSelect = session.prepare(simpleSelect); + + // With PRESERVE_REPLICA_ORDER, the first replica is deterministic for a given partition key. + assertThat(collectCoordinators(session, preparedSelect, 2)).hasSize(1); + } + + /** + * Control for the two tests above. This is the same statement against the same table, executed by + * the same {@code TokenAwarePolicy(RoundRobinPolicy, RANDOM)} — only the consistency level + * differs. A non-serial level takes the {@code REGULAR} routing path, which shuffles the replicas + * on every query, so the coordinator must vary. If this test ever collapses to a single + * coordinator as well, the {@code hasSize(1)} assertions above have stopped proving anything + * about {@code PRESERVE_REPLICA_ORDER}. + */ + @Test(groups = "short") + public void should_spread_non_serial_select_across_replicas() { + Session session = session(); + + SimpleStatement simpleSelect = + new SimpleStatement("SELECT * FROM test_lwt WHERE pk = ? AND ck = ?"); + simpleSelect.setConsistencyLevel(ConsistencyLevel.ONE); + + PreparedStatement preparedSelect = session.prepare(simpleSelect); + BoundStatement boundSelect = preparedSelect.bind(3, 0); + + assertThat(boundSelect.isLWT()).isFalse(); + assertThat(boundSelect.getConsistencyLevel().isSerial()).isFalse(); + + // Uniform over REPLICATION_FACTOR replicas across EXECUTIONS queries, so the probability of a + // false failure here is REPLICATION_FACTOR^(1 - EXECUTIONS). + assertThat(collectCoordinators(session, preparedSelect, 3).size()).isGreaterThan(1); + } + + /** + * Executes {@code prepared} against partition {@code pk} {@link #EXECUTIONS} times and returns + * the distinct coordinators used. + * + *

A fresh {@link BoundStatement} is bound for every execution on purpose. {@link + * PagingOptimizingLoadBalancingPolicy}, which the driver wraps around the configured policy, + * returns {@code Statement.getLastHost()} ahead of the real query plan, and that field is set on + * every successful {@code BoundStatement} execution. Reusing a single instance would therefore + * pin the coordinator after the first query and make every assertion in this class hold + * regardless of how routing actually behaves. + */ + private static Set collectCoordinators( + Session session, PreparedStatement prepared, int pk) { + Set coordinators = new HashSet<>(); + for (int i = 0; i < EXECUTIONS; i++) { + ResultSet rs = session.execute(prepared.bind(pk, 0)); + Host coordinator = rs.getExecutionInfo().getQueriedHost(); + assertThat(coordinator).isNotNull(); + coordinators.add(coordinator.getEndPoint().resolve()); + } + return coordinators; + } +} diff --git a/driver-core/src/test/java/com/datastax/driver/core/schemabuilder/SchemaBuilderIT.java b/driver-core/src/test/java/com/datastax/driver/core/schemabuilder/SchemaBuilderTest.java similarity index 99% rename from driver-core/src/test/java/com/datastax/driver/core/schemabuilder/SchemaBuilderIT.java rename to driver-core/src/test/java/com/datastax/driver/core/schemabuilder/SchemaBuilderTest.java index 1a32eb2d88d..9e1b696ad4e 100644 --- a/driver-core/src/test/java/com/datastax/driver/core/schemabuilder/SchemaBuilderIT.java +++ b/driver-core/src/test/java/com/datastax/driver/core/schemabuilder/SchemaBuilderTest.java @@ -37,7 +37,7 @@ import java.util.Iterator; import org.testng.annotations.Test; -public class SchemaBuilderIT extends CCMTestsSupport { +public class SchemaBuilderTest extends CCMTestsSupport { // Test relies on existence of 'ks' keyspace, // but no such keyspace is created. If (fixed) created,