maedhroz commented on a change in pull request #1111: URL: https://github.com/apache/cassandra/pull/1111#discussion_r672616626
########## File path: test/distributed/org/apache/cassandra/distributed/upgrade/MixedModeMessageForwardTest.java ########## @@ -0,0 +1,127 @@ +/* + * 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.cassandra.distributed.upgrade; + +import org.apache.cassandra.distributed.UpgradeableCluster; +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.api.Feature; +import org.apache.cassandra.distributed.api.NodeToolResult; +import org.apache.cassandra.distributed.shared.Shared; +import org.apache.cassandra.distributed.shared.Versions; +import org.awaitility.Awaitility; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Arrays; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +import static org.apache.cassandra.distributed.shared.AssertUtils.*; + +@Shared +public class MixedModeMessageForwardTest extends UpgradeTestBase +{ + private static final Logger logger = LoggerFactory.getLogger(MixedModeMessageForwardTest.class); + private static int nextKey = 1; + private static String TABLE = "tbl"; + private static String INSERT_QUERY = String.format("INSERT INTO %s.%s(pk) VALUES (?)", KEYSPACE, TABLE); + private static String CHECK_QUERY = String.format("SELECT pk FROM %s.%s WHERE pk = ?", KEYSPACE, TABLE); + + private boolean checkClusterUp(UpgradeableCluster cluster, int coordId) + { + NodeToolResult result; + result = cluster.get(coordId).nodetoolResult("ring"); + + // Must have an Up line for each node + long upCount = Arrays.stream(result.getStdout().split("\\r?\\n")).filter(line -> line.contains(" Up ")).count(); + if (upCount < cluster.size()) + { + logger.info("Only {}/{} are up.", upCount, cluster.size()); + return false; + } + + return true; + } + + private void writeReadTest(UpgradeableCluster cluster) + { + // Coordinate a write from each node and then check present on all replicas + int readKey = nextKey; + for (int coordId = 1; coordId <= cluster.size(); coordId++) + { + final int checkNodeId = coordId; + Awaitility.await("Cluster Up").atMost(1, TimeUnit.MINUTES).until(() -> checkClusterUp(cluster, checkNodeId)); + + logger.info("Coordinating CL.ALL Insert from node{} ", coordId); + cluster.get(coordId).coordinator().execute(INSERT_QUERY, ConsistencyLevel.ALL, nextKey++); + } + + for (int coordId = 1; coordId <= cluster.size(); coordId++) + { + for (int nodeId = 1; nodeId <= cluster.size(); nodeId++) { + Object[][] results = cluster.get(nodeId).executeInternal(CHECK_QUERY, readKey); + assertRows(results, row(readKey)); + } + readKey++; + } + } + + /* Verify that messages sent with sendToHintedReplicas to non-local DCs + * are forwarded on to the hosts there. + * + * 1) creates a mixed cluster with multiple datacenters and a keyspace + * configured to write to all replicas in the datacenter + * 2) check the original single-version cluster by issuing an INSERT + * mutation from a coordinator on each node, then check that value + * has locally been written to each of the nodes. + * 3) Upgrade nodes one at a time, rechecking that all writes are forwarded. + */ + @Test + public void checkWritesForwardedToOtherDcTest() throws Throwable + { + int numDCs = 2; + int nodesPerDc = 3; + String ntsArgs = IntStream.range(1, numDCs + 1) + .mapToObj(dc -> String.format("'datacenter%d' : %d", dc, nodesPerDc)) + .collect(Collectors.joining(",")); + + new TestCase() + .withConfig(c -> c.with(Feature.GOSSIP, Feature.NETWORK).set("request_timeout_in_ms", 30000)) + .withBuilder(b -> b.withRacks(numDCs, 1, nodesPerDc)) + .nodes(numDCs * nodesPerDc) + .upgrade(Versions.Major.v30, Versions.Major.v4) + .setup(cluster -> { + cluster.schemaChange("ALTER KEYSPACE " + KEYSPACE + + " WITH replication = {'class': 'NetworkTopologyStrategy', " + ntsArgs + " };"); + + cluster.schemaChange("CREATE TABLE "+ KEYSPACE + "." + TABLE + " (pk int, PRIMARY KEY(pk))"); + + logger.info("Testing after setup, all nodes running {}", cluster.get(1).getReleaseVersionString()); + writeReadTest(cluster); + }) + .runAfterNodeUpgrade((UpgradeableCluster cluster, int nodeId) -> { + // Should be able to coordinate a write to any node and have a copy appear locally on all others + logger.info("Testing after upgrading node{} to {}", nodeId, cluster.get(nodeId).getReleaseVersionString()); + writeReadTest(cluster); + }) + .run(); Review comment: nit: This is very similar to the framework in `MixedModeReplicationTestBase`, although it clearly adds some new points of configurability. This test is already written, so it probably doesn't make sense to rework, but I figured I'd mention it at least to share some tribal knowledge. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
