mirror of https://github.com/apache/cassandra
Backport CASSANDRA-16418 to 3.x
When a node is decommissioned, it triggers data transfer to other nodes. During this transfer process, receiving nodes temporarily hold token ranges in a pending state. However, the current cleanup process doesn't account for these pending ranges when calculating token ownership, leading to inadvertent cleanup of data already stored in SSTables. To address this issue, this patch introduces two changes. Firstly, it backports CASSANDRA-16418, introducing a preventive check in `StorageService#forceKeyspaceCleanup`. This check disallows the initiation of cleanup when a node contains any pending ranges for the requested keyspace. Secondly, it reintroduces a similar condition to test for the existence of pending ranges in `CompactionManager#performCleanup`. This ensures the safety of this API as well. Patch by Szymon Miezal; reviewed by Brandon Williams, Jacek Lewandowski for CASSANDRA-18824 Co-authored-by: Szymon Miezal <szymon.miezal@datastax.com> Co-authored-by: Jacek Lewandowski <lewandowski.jacek@gmail.com>
This commit is contained in:
Szymon Miężał
Jacek Lewandowski
parent
0bc95bbebd
commit
5be57829b0
18
CHANGES.txt
18
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
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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))
|
||||
{
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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));
|
||||
}
|
||||
}
|
||||
|
|
@ -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<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, 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);
|
||||
|
|
|
|||
Loading…
Reference in New Issue