diff --git a/CHANGES.txt b/CHANGES.txt index b53bc55d26..c49019041c 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 3.0.30 + * Backport CASSANDRA-16418 to 3.x (CASSANDRA-18824) * Suppress CVE-2023-6378 (CASSANDRA-19142) * Do not set RPC_READY to false on transports shutdown in order to not fail counter updates for deployments with coordinator and storage nodes with transports turned off (CASSANDRA-18935) * Suppress CVE-2023-44487 (CASSANDRA-18943) @@ -26,6 +27,7 @@ * Pass down all contact points to driver for cassandra-stress (CASSANDRA-18025) * Validate the existence of a datacenter in nodetool rebuild (CASSANDRA-14319) + 3.0.29 * Suppress CVE-2023-2251 (CASSANDRA-18497) * Do not remove SSTables when cause of FSReadError is OutOfMemoryError while using best_effort disk failure policy (CASSANDRA-18336) @@ -176,6 +178,7 @@ Merged from 2.2: * Remove OpenJDK log warning (CASSANDRA-15563) * Fix the histogram merge of the table metrics (CASSANDRA-16259) + 3.0.23: * Fix OOM when terminating repair session (CASSANDRA-15902) * Avoid marking shutting down nodes as up after receiving gossip shutdown message (CASSANDRA-16094) @@ -188,6 +191,7 @@ Merged from 2.2: * Package tools/bin scripts as executable (CASSANDRA-16151) * Fixed a NullPointerException when calling nodetool enablethrift (CASSANDRA-16127) + 3.0.22: * Fix gossip shutdown order (CASSANDRA-15816) * Remove broken 'defrag-on-read' optimization (CASSANDRA-15432) @@ -203,6 +207,7 @@ Merged from 2.2: Merged from 2.1: * Only allow strings to be passed to JMX authentication (CASSANDRA-16077) + 3.0.21 * Backport CASSANDRA-12189: escape string literals (CASSANDRA-15948) * Avoid hinted handoff per-host throttle being arounded to 0 in large cluster (CASSANDRA-15859) @@ -957,6 +962,7 @@ Merged from 2.1: * InvalidateKeys should have a weak ref to key cache (CASSANDRA-11176) * COPY FROM on large datasets: fix progress report and debug performance (CASSANDRA-11053) + 3.0.4 * Preserve order for preferred SSL cipher suites (CASSANDRA-11164) * MV should only query complex columns included in the view (CASSANDRA-11069) @@ -5011,7 +5017,7 @@ Merged from 0.8: * improve ignoring of obsolete mutations in index maintenance (CASSANDRA-2401) * recognize attempt to drop just the index while leaving the column definition alone (CASSANDRA-2619) - + 0.8.0-beta1 * remove Avro RPC support (CASSANDRA-926) @@ -5646,7 +5652,7 @@ Merged from 0.8: * Allow using DynamicEndpointSnitch with RackAwareStrategy (CASSANDRA-1429) * remove the remaining vestiges of the unfinished DatacenterShardStrategy (replaced by NetworkTopologyStrategy in 0.7) - + 0.6.5 * fix key ordering in range query results with RandomPartitioner @@ -5887,7 +5893,7 @@ Merged from 0.8: * change streaming chunk size to 32MB to accomodate Windows XP limitations (was 64MB) (CASSANDRA-795) * fix get_range_slice returning results in the wrong order (CASSANDRA-781) - + 0.5.0 final * avoid attempting to delete temporary bootstrap files twice (CASSANDRA-681) @@ -6014,7 +6020,7 @@ Merged from 0.8: * optimized local-node writes (CASSANDRA-558) * added get_range_slice, deprecating get_key_range (CASSANDRA-344) * expose TimedOutException to thrift (CASSANDRA-563) - + 0.4.2 * Add validation disallowing null keys (CASSANDRA-486) @@ -6080,7 +6086,6 @@ Merged from 0.8: - Similarly, merged batch_insert_super into batch_insert. - 0.4.0 beta * On-disk data format has changed to allow billions of keys/rows per node instead of only millions @@ -6117,7 +6122,6 @@ Merged from 0.8: * Rename configuration "table" to "keyspace" * Moved to crash-only design; no more shutdown (just kill the process) * Lots of bug fixes - Full list of issues resolved in 0.4 is at https://issues.apache.org/jira/secure/IssueNavigator.jspa?reset=true&&pid=12310865&fixfor=12313862&resolution=1&sorter/field=issuekey&sorter/order=DESC @@ -6148,3 +6152,5 @@ Full list of issues resolved in 0.4 is at https://issues.apache.org/jira/secure/ * Combined blocking and non-blocking versions of insert APIs * Added FlushPeriodInMinutes configuration parameter to force flushing of infrequently-updated ColumnFamilies + + diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java index 2b9ee508b1..e7245ba06d 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java @@ -453,9 +453,9 @@ public class CompactionManager implements CompactionManagerMBean { assert !cfStore.isIndex(); Keyspace keyspace = cfStore.keyspace; - if (!StorageService.instance.isJoined()) + if (!StorageService.instance.getTokenMetadata().getPendingRanges(keyspace.getName(), FBUtilities.getBroadcastAddress()).isEmpty()) { - logger.info("Cleanup cannot run before a node has joined the ring"); + 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 diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index a1848f80f0..080a22c6c7 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -2996,6 +2996,9 @@ public class StorageService extends NotificationBroadcasterSupport implements IE if (Schema.isLocalSystemKeyspace(keyspaceName)) throw new RuntimeException("Cleanup of the system keyspace is neither necessary nor wise"); + if (!tokenMetadata.getPendingRanges(keyspaceName, FBUtilities.getBroadcastAddress()).isEmpty()) + throw new RuntimeException("Node is involved in cluster membership changes. Not safe to run cleanup."); + CompactionManager.AllSSTableOpStatus status = CompactionManager.AllSSTableOpStatus.SUCCESSFUL; for (ColumnFamilyStore cfStore : getValidColumnFamilies(false, false, keyspaceName, tables)) { diff --git a/test/distributed/org/apache/cassandra/distributed/action/GossipHelper.java b/test/distributed/org/apache/cassandra/distributed/action/GossipHelper.java index 75eb5d4186..1737035dc1 100644 --- a/test/distributed/org/apache/cassandra/distributed/action/GossipHelper.java +++ b/test/distributed/org/apache/cassandra/distributed/action/GossipHelper.java @@ -52,6 +52,22 @@ public class GossipHelper }; } + public static InstanceAction statusToDecommission(IInvokableInstance newNode) + { + return (instance) -> + { + changeGossipState(instance, + newNode, + Arrays.asList(tokens(newNode), + statusLeaving(newNode))); + }; + } + + public static VersionedApplicationState statusLeaving(IInvokableInstance instance) + { + return versionedToken(instance, ApplicationState.STATUS, (partitioner, tokens) -> new VersionedValue.VersionedValueFactory(partitioner).leaving(tokens)); + } + public static void withProperty(String prop, String value, Runnable r) { String before = System.getProperty(prop); diff --git a/test/distributed/org/apache/cassandra/distributed/test/CleanupFailureTest.java b/test/distributed/org/apache/cassandra/distributed/test/CleanupFailureTest.java new file mode 100644 index 0000000000..778423d07f --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/CleanupFailureTest.java @@ -0,0 +1,118 @@ +/* + * 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.test; + +import org.junit.Test; + +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.api.IInstanceConfig; +import org.apache.cassandra.distributed.api.IInvokableInstance; +import org.apache.cassandra.distributed.api.NodeToolResult; +import org.apache.cassandra.distributed.shared.NetworkTopology; + +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.api.Feature.GOSSIP; +import static org.apache.cassandra.distributed.api.Feature.NETWORK; +import static org.apache.cassandra.distributed.api.TokenSupplier.evenlyDistributedTokens; +import static org.junit.Assert.assertEquals; + +public class CleanupFailureTest extends TestBaseImpl +{ + @Test + public void cleanupDuringDecommissionTest() throws Throwable + { + 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); + + // Start decomission on nodeToDecommission + cluster.forEach(statusToDecommission(nodeToDecommission)); + + // Add data to cluster while node is decomissioning + 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); + + // Run cleanup on nodeToRemainInCluster + NodeToolResult result = nodeToRemainInCluster.nodetoolResult("cleanup"); + result.asserts().failure(); + result.asserts().stderrContains("Node is involved in cluster membership changes. Not safe to run cleanup."); + + // Check data after cleanup on nodeToRemainInCluster + assertEquals(100, nodeToRemainInCluster.executeInternal("SELECT * FROM " + KEYSPACE + ".tbl").length); + } + } + + @Test + public void cleanupDuringBootstrapTest() throws Throwable + { + int originalNodeCount = 1; + int expandedNodeCount = originalNodeCount + 1; + + 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); + withProperty("cassandra.join_ring", "false", + () -> bootstrappingNode.startup(cluster)); + + // Start decomission on bootstrappingNode + cluster.forEach(statusToBootstrap(bootstrappingNode)); + + // Add data to cluster while node is bootstrapping + 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(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(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)); + } +} diff --git a/test/unit/org/apache/cassandra/db/CleanupTest.java b/test/unit/org/apache/cassandra/db/CleanupTest.java index d4c613d237..dd433e83a6 100644 --- a/test/unit/org/apache/cassandra/db/CleanupTest.java +++ b/test/unit/org/apache/cassandra/db/CleanupTest.java @@ -41,6 +41,11 @@ import org.apache.cassandra.SchemaLoader; import org.apache.cassandra.Util; import org.apache.cassandra.config.ColumnDefinition; import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.locator.AbstractReplicationStrategy; +import org.apache.cassandra.locator.IEndpointSnitch; +import org.apache.cassandra.locator.PendingRangeMaps; +import org.apache.cassandra.locator.PropertyFileSnitch; +import org.apache.cassandra.locator.SimpleStrategy; import org.apache.cassandra.schema.KeyspaceMetadata; import org.apache.cassandra.cql3.Operator; import org.apache.cassandra.db.compaction.CompactionManager; @@ -116,9 +121,11 @@ public class CleanupTest } @Test - public void testCleanup() throws ExecutionException, InterruptedException + public void testCleanup() throws ExecutionException, InterruptedException, UnknownHostException { - StorageService.instance.getTokenMetadata().clearUnsafe(); + TokenMetadata tmd = StorageService.instance.getTokenMetadata(); + tmd.clearUnsafe(); + tmd.updateNormalToken(token(new byte[]{ 50 }), InetAddress.getByName("127.0.0.1")); Keyspace keyspace = Keyspace.open(KEYSPACE1); ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF_STANDARD1); @@ -330,6 +337,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, InetAddress.getByName("127.0.0.1")); + } + + private AbstractReplicationStrategy createStrategy(String keyspace) + { + IEndpointSnitch snitch = new PropertyFileSnitch(); + DatabaseDescriptor.setEndpointSnitch(snitch); + return new SimpleStrategy(keyspace, new TokenMetadata(), DatabaseDescriptor.getEndpointSnitch(), Collections.emptyMap()); + } + private static BytesToken token(byte ... value) { return new BytesToken(value);