Merge branch 'cassandra-4.1' into cassandra-5.0

* cassandra-4.1:
  Backport CASSANDRA-16418 to 3.x

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

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<Range<Token>> allRanges = replicas.ranges();

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;
 

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;

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)
+        try (Cluster cluster = init(builder().withNodes(2)
-                                        .withTokenSupplier(evenlyDistributedTokens(2))
+                                             .withTokenSupplier(evenlyDistributedTokens(2))
-                                        .withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(2, "dc0", "rack0"))
+                                             .withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(2, "dc0", "rack0"))
-                                        .withConfig(config -> config.with(NETWORK, GOSSIP))
+                                             .withConfig(config -> config.with(NETWORK, GOSSIP))
-                                        .start())
+                                             .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;
+            int numRows = 100;
-            populate(cluster, 0, NUM_ROWS, 1, 1, ConsistencyLevel.ONE);
+            cluster.schemaChange("CREATE TABLE IF NOT EXISTS " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
-            cluster.forEach(c -> c.flush(KEYSPACE));
+            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)
+        try (Cluster cluster = init(builder().withNodes(originalNodeCount)
-                                        .withTokenSupplier(evenlyDistributedTokens(expandedNodeCount))
+                                             .withTokenSupplier(evenlyDistributedTokens(expandedNodeCount))
-                                        .withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(expandedNodeCount, "dc0", "rack0"))
+                                             .withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(expandedNodeCount, "dc0", "rack0"))
-                                        .withConfig(config -> config.with(NETWORK, GOSSIP))
+                                             .withConfig(config -> config.with(NETWORK, GOSSIP))
-                                        .start())
+                                             .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;
+            int numRows = 100;
-            populate(cluster, 0, NUM_ROWS, 1, 2, ConsistencyLevel.ONE);
+            cluster.schemaChange("CREATE TABLE IF NOT EXISTS " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
-            cluster.forEach(c -> c.flush(KEYSPACE));
+            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);
         }
     }
-}
+
+    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));
+    }
+}

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<Token> 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<Token> 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);