diff --git a/CHANGES.txt b/CHANGES.txt index 409a8c4f5b..1549165f6c 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -23,6 +23,7 @@ Merged from 4.0: * Revert unnecessary read lock acquisition when reading ring version in TokenMetadata introduced in CASSANDRA-16286 (CASSANDRA-19107) Merged from 3.11: Merged from 3.0: + * Backport CASSANDRA-16418 to 3.x (CASSANDRA-18824) 5.0-beta1 diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java index 8f7bf6946c..791a8e3c67 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java @@ -616,7 +616,11 @@ public class CompactionManager implements CompactionManagerMBean, ICompactionMan { assert !cfStore.isIndex(); Keyspace keyspace = cfStore.keyspace; - + if (!StorageService.instance.getTokenMetadata().getPendingRanges(keyspace.getName(), FBUtilities.getBroadcastAddressAndPort()).isEmpty()) + { + logger.info("Cleanup cannot run while node has pending ranges for keyspace {} table {}, wait for node addition/decommission to complete and try again", cfStore.keyspace.getName(), cfStore.getTableName()); + return AllSSTableOpStatus.ABORTED; + } // if local ranges is empty, it means no data should remain final RangesAtEndpoint replicas = StorageService.instance.getLocalReplicas(keyspace.getName()); final Set> allRanges = replicas.ranges(); diff --git a/src/java/org/apache/cassandra/locator/SimpleStrategy.java b/src/java/org/apache/cassandra/locator/SimpleStrategy.java index 488b601ce7..c8cf4e27ae 100644 --- a/src/java/org/apache/cassandra/locator/SimpleStrategy.java +++ b/src/java/org/apache/cassandra/locator/SimpleStrategy.java @@ -45,7 +45,7 @@ import org.apache.cassandra.service.StorageService; */ public class SimpleStrategy extends AbstractReplicationStrategy { - private static final String REPLICATION_FACTOR = "replication_factor"; + public static final String REPLICATION_FACTOR = "replication_factor"; private static final Logger logger = LoggerFactory.getLogger(SimpleStrategy.class); private final ReplicationFactor rf; diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index db78db2e86..78be739301 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -4065,7 +4065,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE if (SchemaConstants.isLocalSystemKeyspace(keyspaceName)) throw new RuntimeException("Cleanup of the system keyspace is neither necessary nor wise"); - if (tokenMetadata.getPendingRanges(keyspaceName, getBroadcastAddressAndPort()).size() > 0) + if (!tokenMetadata.getPendingRanges(keyspaceName, getBroadcastAddressAndPort()).isEmpty()) throw new RuntimeException("Node is involved in cluster membership changes. Not safe to run cleanup."); CompactionManager.AllSSTableOpStatus status = CompactionManager.AllSSTableOpStatus.SUCCESSFUL; diff --git a/test/distributed/org/apache/cassandra/distributed/test/ring/CleanupFailureTest.java b/test/distributed/org/apache/cassandra/distributed/test/ring/CleanupFailureTest.java index 8ec3630983..8a2ff3dccc 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/ring/CleanupFailureTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/ring/CleanupFailureTest.java @@ -33,7 +33,6 @@ import static org.junit.Assert.assertEquals; import static org.apache.cassandra.distributed.action.GossipHelper.statusToBootstrap; import static org.apache.cassandra.distributed.action.GossipHelper.statusToDecommission; import static org.apache.cassandra.distributed.action.GossipHelper.withProperty; -import static org.apache.cassandra.distributed.test.ring.BootstrapTest.populate; import static org.apache.cassandra.distributed.api.Feature.GOSSIP; import static org.apache.cassandra.distributed.api.Feature.NETWORK; import static org.apache.cassandra.distributed.api.TokenSupplier.evenlyDistributedTokens; @@ -43,11 +42,11 @@ public class CleanupFailureTest extends TestBaseImpl @Test public void cleanupDuringDecommissionTest() throws Throwable { - try (Cluster cluster = builder().withNodes(2) - .withTokenSupplier(evenlyDistributedTokens(2)) - .withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(2, "dc0", "rack0")) - .withConfig(config -> config.with(NETWORK, GOSSIP)) - .start()) + try (Cluster cluster = init(builder().withNodes(2) + .withTokenSupplier(evenlyDistributedTokens(2)) + .withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(2, "dc0", "rack0")) + .withConfig(config -> config.with(NETWORK, GOSSIP)) + .start(), 1)) { IInvokableInstance nodeToDecommission = cluster.get(1); IInvokableInstance nodeToRemainInCluster = cluster.get(2); @@ -56,9 +55,9 @@ public class CleanupFailureTest extends TestBaseImpl cluster.forEach(statusToDecommission(nodeToDecommission)); // Add data to cluster while node is decomissioning - int NUM_ROWS = 100; - populate(cluster, 0, NUM_ROWS, 1, 1, ConsistencyLevel.ONE); - cluster.forEach(c -> c.flush(KEYSPACE)); + int numRows = 100; + cluster.schemaChange("CREATE TABLE IF NOT EXISTS " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))"); + insertData(cluster, 1, numRows, ConsistencyLevel.ONE); // Check data before cleanup on nodeToRemainInCluster assertEquals(100, nodeToRemainInCluster.executeInternal("SELECT * FROM " + KEYSPACE + ".tbl").length); @@ -79,11 +78,11 @@ public class CleanupFailureTest extends TestBaseImpl int originalNodeCount = 1; int expandedNodeCount = originalNodeCount + 1; - try (Cluster cluster = builder().withNodes(originalNodeCount) - .withTokenSupplier(evenlyDistributedTokens(expandedNodeCount)) - .withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(expandedNodeCount, "dc0", "rack0")) - .withConfig(config -> config.with(NETWORK, GOSSIP)) - .start()) + try (Cluster cluster = init(builder().withNodes(originalNodeCount) + .withTokenSupplier(evenlyDistributedTokens(expandedNodeCount)) + .withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(expandedNodeCount, "dc0", "rack0")) + .withConfig(config -> config.with(NETWORK, GOSSIP)) + .start(), 2)) { IInstanceConfig config = cluster.newInstanceConfig(); IInvokableInstance bootstrappingNode = cluster.bootstrap(config); @@ -94,19 +93,28 @@ public class CleanupFailureTest extends TestBaseImpl cluster.forEach(statusToBootstrap(bootstrappingNode)); // Add data to cluster while node is bootstrapping - int NUM_ROWS = 100; - populate(cluster, 0, NUM_ROWS, 1, 2, ConsistencyLevel.ONE); - cluster.forEach(c -> c.flush(KEYSPACE)); + int numRows = 100; + cluster.schemaChange("CREATE TABLE IF NOT EXISTS " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))"); + insertData(cluster, 1, numRows, ConsistencyLevel.ONE); // Check data before cleanup on bootstrappingNode - assertEquals(NUM_ROWS, bootstrappingNode.executeInternal("SELECT * FROM " + KEYSPACE + ".tbl").length); + assertEquals(numRows, bootstrappingNode.executeInternal("SELECT * FROM " + KEYSPACE + ".tbl").length); // Run cleanup on bootstrappingNode NodeToolResult result = bootstrappingNode.nodetoolResult("cleanup"); result.asserts().stderrContains("Node is involved in cluster membership changes. Not safe to run cleanup."); // Check data after cleanup on bootstrappingNode - assertEquals(NUM_ROWS, bootstrappingNode.executeInternal("SELECT * FROM " + KEYSPACE + ".tbl").length); + assertEquals(numRows, bootstrappingNode.executeInternal("SELECT * FROM " + KEYSPACE + ".tbl").length); } } -} \ No newline at end of file + + private void insertData(Cluster cluster, int node, int numberOfRows, ConsistencyLevel cl) + { + for (int i = 0; i < numberOfRows; i++) + { + cluster.coordinator(node).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (?, ?, ?)", cl, i, i, i); + } + cluster.forEach(c -> c.flush(KEYSPACE)); + } +} diff --git a/test/unit/org/apache/cassandra/db/CleanupTest.java b/test/unit/org/apache/cassandra/db/CleanupTest.java index 5d176859c2..23fb0a3a62 100644 --- a/test/unit/org/apache/cassandra/db/CleanupTest.java +++ b/test/unit/org/apache/cassandra/db/CleanupTest.java @@ -32,26 +32,33 @@ import java.util.UUID; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; +import com.google.common.collect.ImmutableMap; import com.google.common.collect.Sets; import org.junit.BeforeClass; import org.junit.Test; import org.apache.cassandra.SchemaLoader; import org.apache.cassandra.Util; -import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.config.DatabaseDescriptor; -import org.apache.cassandra.schema.ColumnMetadata; -import org.apache.cassandra.schema.KeyspaceMetadata; import org.apache.cassandra.cql3.Operator; import org.apache.cassandra.db.compaction.CompactionManager; import org.apache.cassandra.db.filter.RowFilter; import org.apache.cassandra.dht.ByteOrderedPartitioner.BytesToken; import org.apache.cassandra.dht.Range; +import org.apache.cassandra.dht.Token; import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.io.sstable.format.SSTableReader; -import org.apache.cassandra.dht.Token; import org.apache.cassandra.locator.AbstractNetworkTopologySnitch; +import org.apache.cassandra.locator.AbstractReplicationStrategy; +import org.apache.cassandra.locator.IEndpointSnitch; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.locator.PendingRangeMaps; +import org.apache.cassandra.locator.PropertyFileSnitch; +import org.apache.cassandra.locator.Replica; +import org.apache.cassandra.locator.SimpleStrategy; import org.apache.cassandra.locator.TokenMetadata; +import org.apache.cassandra.schema.ColumnMetadata; +import org.apache.cassandra.schema.KeyspaceMetadata; import org.apache.cassandra.schema.KeyspaceParams; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.utils.ByteBufferUtil; @@ -407,6 +414,43 @@ public class CleanupTest assertEquals(testCase.getKey(), CompactionManager.needsCleanup(ssTable, testCase.getValue())); } } + + @Test + public void testCleanupIsAbortedWhenNodeHasPendingRanges() throws ExecutionException, InterruptedException, UnknownHostException + { + // given + StorageService.instance.getTokenMetadata().clearUnsafe(); + + Keyspace keyspace = Keyspace.open(KEYSPACE1); + ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF_STANDARD1); + + fillCF(cfs, "val", LOOPS); + assertEquals(LOOPS, Util.getAll(Util.cmd(cfs).build()).size()); + + Range range = range(new BytesToken(new byte[]{0}), new BytesToken(new byte[]{1})); + givenPendingRange(cfs, range); + + // when + CompactionManager.AllSSTableOpStatus status = CompactionManager.instance.performCleanup(cfs, 2); + + // then + assertEquals("cleanup should be aborted", CompactionManager.AllSSTableOpStatus.ABORTED, status); + } + + private void givenPendingRange(ColumnFamilyStore cfs, Range range) throws UnknownHostException + { + StorageService.instance.getTokenMetadata().calculatePendingRanges(createStrategy(cfs.keyspace.getName()), cfs.keyspace.getName()); + PendingRangeMaps ranges = StorageService.instance.getTokenMetadata().getPendingRanges(cfs.keyspace.getName()); + ranges.addPendingRange(range, Replica.fullReplica(InetAddressAndPort.getByName("127.0.0.1"), range)); + } + + private AbstractReplicationStrategy createStrategy(String keyspace) + { + IEndpointSnitch snitch = new PropertyFileSnitch(); + DatabaseDescriptor.setEndpointSnitch(snitch); + return new SimpleStrategy(keyspace, new TokenMetadata(), DatabaseDescriptor.getEndpointSnitch(), ImmutableMap.of(SimpleStrategy.REPLICATION_FACTOR, "1")); + } + private static BytesToken token(byte ... value) { return new BytesToken(value);