From 8d2c11ed2e9f886261c845636892acfcc5e1e782 Mon Sep 17 00:00:00 2001 From: Dmitry Konstantinov Date: Fri, 5 Sep 2025 13:47:39 +0100 Subject: [PATCH] Fix cleanup of old incremental repair sessions in case of owned token range changes or a table deleting Patch by Dmitry Konstantinov; reviewed by Marcus Eriksson, Jaydeepkumar Chovatia for CASSANDRA-20877 --- CHANGES.txt | 1 + src/java/org/apache/cassandra/dht/Range.java | 15 ++ .../repair/consistent/LocalSessions.java | 30 +++- ...entalRepairCleanupAfterNodeAddingTest.java | 145 ++++++++++++++++++ .../org/apache/cassandra/dht/RangeTest.java | 44 ++++++ .../repair/consistent/LocalSessionTest.java | 16 ++ 6 files changed, 249 insertions(+), 2 deletions(-) create mode 100644 test/distributed/org/apache/cassandra/distributed/test/IncrementalRepairCleanupAfterNodeAddingTest.java diff --git a/CHANGES.txt b/CHANGES.txt index ad5c3f688c..cc845d3588 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.0.20 + * Fix cleanup of old incremental repair sessions in case of owned token range changes or a table deleting (CASSANDRA-20877) * Fix memory leak in BufferPoolAllocator when a capacity needs to be extended (CASSANDRA-20753) * Leveled Compaction doesn't validate maxBytesForLevel when the table is altered/created (CASSANDRA-20570) * Updated dtest-api to 0.0.18 and removed JMX-related classes that now live in the dtest-api (CASSANDRA-20884) diff --git a/src/java/org/apache/cassandra/dht/Range.java b/src/java/org/apache/cassandra/dht/Range.java index 5b2f3d9fbf..9f77fbd9f2 100644 --- a/src/java/org/apache/cassandra/dht/Range.java +++ b/src/java/org/apache/cassandra/dht/Range.java @@ -352,6 +352,21 @@ public class Range> extends AbstractBounds implemen return result; } + public static > List> intersect(Collection> ranges1, Collection> ranges2) + { + Set> result = new HashSet<>(); + // note: O(n^2), simple but not very efficient + for (Range range1 : ranges1) + { + for (Range range2 : ranges2) + { + result.addAll(range1.intersectionWith(range2)); + } + } + return normalize(result); + } + + /** * Calculate set of the difference ranges of given two ranges * (as current (A, B] and rhs is (C, D]) diff --git a/src/java/org/apache/cassandra/repair/consistent/LocalSessions.java b/src/java/org/apache/cassandra/repair/consistent/LocalSessions.java index dd60ad4a21..fbc8b7d666 100644 --- a/src/java/org/apache/cassandra/repair/consistent/LocalSessions.java +++ b/src/java/org/apache/cassandra/repair/consistent/LocalSessions.java @@ -92,6 +92,7 @@ import org.apache.cassandra.repair.messages.RepairMessage; import org.apache.cassandra.repair.messages.StatusRequest; import org.apache.cassandra.repair.messages.StatusResponse; import org.apache.cassandra.schema.TableId; +import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.service.ActiveRepairService; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.utils.FBUtilities; @@ -239,14 +240,27 @@ public class LocalSessions */ private boolean isSuperseded(LocalSession session) { + // to reduce overheads of intersect calculation for tables within the same keyspace + Map>> rangesPerKeyspaceCache = new HashMap<>(); for (TableId tid : session.tableIds) { - RepairedState state = repairedStates.get(tid); + TableMetadata tableMetadata = getTableMetadata(tid); + if (tableMetadata == null) // if a table was removed - ignore it + continue; + RepairedState state = repairedStates.get(tid); if (state == null) return false; - long minRepaired = state.minRepairedAt(session.ranges); + Collection> actualRanges = rangesPerKeyspaceCache.computeIfAbsent(tableMetadata.keyspace, (keyspace) -> { + List> localRanges = getLocalRanges(tableMetadata.keyspace); + if (localRanges.isEmpty()) // to handle the case when we run before the information about owned ranges is properly populated + return session.ranges; + + // ignore token ranges which were moved to other nodes and not owned by the current one anymore + return Range.intersect(session.ranges, localRanges); + }); + long minRepaired = state.minRepairedAt(actualRanges); if (minRepaired <= session.repairedAt) return false; } @@ -254,6 +268,18 @@ public class LocalSessions return true; } + @VisibleForTesting + protected TableMetadata getTableMetadata(TableId tableId) + { + return Schema.instance.getTableMetadata(tableId); + } + + @VisibleForTesting + protected List> getLocalRanges(String keyspace) + { + return StorageService.instance.getLocalAndPendingRanges(keyspace); + } + public RepairedState.Stats getRepairedStats(TableId tid, Collection> ranges) { RepairedState state = repairedStates.get(tid); diff --git a/test/distributed/org/apache/cassandra/distributed/test/IncrementalRepairCleanupAfterNodeAddingTest.java b/test/distributed/org/apache/cassandra/distributed/test/IncrementalRepairCleanupAfterNodeAddingTest.java new file mode 100644 index 0000000000..a14f78c0be --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/IncrementalRepairCleanupAfterNodeAddingTest.java @@ -0,0 +1,145 @@ +/* + * 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 java.util.List; +import java.util.Map; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import org.junit.Assert; +import org.junit.Test; + +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.api.ICluster; +import org.apache.cassandra.distributed.api.IInstanceConfig; +import org.apache.cassandra.distributed.api.IInvokableInstance; +import org.apache.cassandra.distributed.api.TokenSupplier; +import org.apache.cassandra.distributed.shared.NetworkTopology; +import org.apache.cassandra.distributed.shared.WithProperties; +import org.apache.cassandra.repair.consistent.ConsistentSession; +import org.apache.cassandra.service.ActiveRepairService; +import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.utils.concurrent.SimpleCondition; +import org.apache.cassandra.utils.progress.ProgressEventType; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static java.util.concurrent.TimeUnit.MINUTES; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.apache.cassandra.distributed.api.Feature.GOSSIP; +import static org.apache.cassandra.distributed.api.Feature.NATIVE_PROTOCOL; +import static org.apache.cassandra.distributed.api.Feature.NETWORK; +import static org.apache.cassandra.distributed.test.ExecUtil.rethrow; +import static org.apache.cassandra.repair.consistent.LocalSessionInfo.STATE; +import static org.apache.cassandra.repair.messages.RepairOption.INCREMENTAL_KEY; +import static org.awaitility.Awaitility.await; +import static org.hamcrest.collection.IsCollectionWithSize.hasSize; + +public class IncrementalRepairCleanupAfterNodeAddingTest extends TestBaseImpl +{ + @Test + public void test() throws Exception + { + int originalNodeCount = 3; + try (WithProperties withProperties = new WithProperties()) + { + withProperties.setProperty("cassandra.repair_delete_timeout_seconds", "0"); + try (Cluster cluster = builder().withNodes(originalNodeCount) + .withTokenSupplier(TokenSupplier.evenlyDistributedTokens(originalNodeCount + 1, 1)) + .withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(4, "dc0", "rack0")) + .withConfig(config -> config.with(NETWORK, GOSSIP, NATIVE_PROTOCOL)) + .start()) + { + populate(cluster, 0, 100); + + repair(cluster, KEYSPACE, ImmutableMap.of(INCREMENTAL_KEY, "true")); + + Thread.sleep(1); // to ensure that we crossed LocalSessions.AUTO_DELETE_TIMEOUT + + // to check that the session is still here (it is not superseded yet) + cluster.get(1).runOnInstance(rethrow(() -> { + ActiveRepairService.instance.consistent.local.cleanup(); + List> sessions = ActiveRepairService.instance.getSessions(true, null); + Assert.assertThat(sessions, hasSize(1)); + })); + + addNode(cluster); + + repair(cluster, KEYSPACE, ImmutableMap.of(INCREMENTAL_KEY, "true")); + + Thread.sleep(1); // to ensure that we crossed LocalSessions.AUTO_DELETE_TIMEOUT + + cluster.get(1).runOnInstance(rethrow(() -> { + ActiveRepairService.instance.consistent.local.cleanup(); + List> sessions = ActiveRepairService.instance.getSessions(true, null); + Assert.assertThat(sessions, hasSize(1)); + })); + } + } + } + + protected void addNode(Cluster cluster) + { + IInstanceConfig config = cluster.newInstanceConfig(); + config.set("auto_bootstrap", true); + IInvokableInstance newInstance = cluster.bootstrap(config); + newInstance.startup(cluster); + } + + public static void populate(ICluster cluster, int from, int to) + { + populate(cluster, from, to, 1, 3, ConsistencyLevel.QUORUM); + } + + public static void populate(ICluster cluster, int from, int to, int coord, int rf, ConsistencyLevel cl) + { + cluster.schemaChange(withKeyspace("CREATE KEYSPACE IF NOT EXISTS %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor': " + rf + "};")); + cluster.schemaChange(withKeyspace("CREATE TABLE IF NOT EXISTS %s.repair_add_node_test (pk int, ck int, v int, PRIMARY KEY (pk, ck))")); + for (int i = from; i < to; i++) + { + cluster.coordinator(coord).execute(withKeyspace("INSERT INTO %s.repair_add_node_test (pk, ck, v) VALUES (?, ?, ?)"), + cl, i, i, i); + } + } + + static void repair(ICluster cluster, String keyspace, Map options) + { + cluster.get(1).runOnInstance(rethrow(() -> { + SimpleCondition await = new SimpleCondition(); + StorageService.instance.repair(keyspace, options, ImmutableList.of((tag, event) -> { + if (event.getType() == ProgressEventType.COMPLETE) + await.signalAll(); + })).right.get(); + await.await(1L, MINUTES); + + // local sessions finalization happens asynchronously + // so to avoid race condition and flakiness for the test we wait explicitly for local sessions to finalize + await().pollInterval(10, MILLISECONDS) + .atMost(60, SECONDS) + .until(() -> { + List> sessions = ActiveRepairService.instance.getSessions(true, null); + for (Map sessionInfo : sessions) + if (!sessionInfo.get(STATE).equals(ConsistentSession.State.FINALIZED.toString())) + return false; + return true; + }); + })); + } +} diff --git a/test/unit/org/apache/cassandra/dht/RangeTest.java b/test/unit/org/apache/cassandra/dht/RangeTest.java index 7cdb78869c..54d8f58d57 100644 --- a/test/unit/org/apache/cassandra/dht/RangeTest.java +++ b/test/unit/org/apache/cassandra/dht/RangeTest.java @@ -737,4 +737,48 @@ public class RangeTest assertEquals(ranges, Range.subtract(ranges, asList(r(6, 7), r(20, 25)))); assertEquals(Sets.newHashSet(r(1, 4), r(11, 15)), Range.subtract(ranges, asList(r(4, 7), r(8, 11)))); } + + @Test + public void testGroupIntersection() + { + assertEquals(Collections.emptyList(), + Range.intersect(asList(r(1, 5), r(10, 15)), + asList(r(6, 7), r(20, 25)) + )); + + assertEquals(asList(r(5, 6)), + Range.intersect(asList(r(1, 6), r(10, 15)), + asList(r(5, 10)) + )); + + assertEquals(asList(r(5, 6), r(10, 11)), + Range.intersect(asList(r(1, 6), r(10, 15)), + asList(r(5, 11)) + )); + + assertEquals(asList(r(5, 6), r(10, 11)), + Range.intersect(asList(r(1, 6), r(10, 15)), + asList(r(5, 11)) + )); + + assertEquals(asList(r(5, 6), r(10, 11), r(12, 15)), + Range.intersect(asList(r(1, 6), r(10, 15)), + asList(r(5, 11), r(12, 20)) + )); + + assertEquals(asList(r(5, 6), r(10, 15)), + Range.intersect(asList(r(1, 6), r(10, 15)), + asList(r(5, 11), r(11, 20)) + )); + + assertEquals(Collections.emptyList(), + Range.intersect(Collections.emptyList(), + asList(r(5, 11), r(11, 20)) + )); + + assertEquals(Collections.emptyList(), + Range.intersect(asList(r(1, 6), r(10, 15)), + Collections.emptyList() + )); + } } diff --git a/test/unit/org/apache/cassandra/repair/consistent/LocalSessionTest.java b/test/unit/org/apache/cassandra/repair/consistent/LocalSessionTest.java index 1003681414..fdb578956d 100644 --- a/test/unit/org/apache/cassandra/repair/consistent/LocalSessionTest.java +++ b/test/unit/org/apache/cassandra/repair/consistent/LocalSessionTest.java @@ -19,6 +19,7 @@ package org.apache.cassandra.repair.consistent; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collection; import java.util.HashMap; import java.util.HashSet; @@ -44,11 +45,14 @@ import org.apache.cassandra.SchemaLoader; import org.apache.cassandra.cql3.QueryProcessor; import org.apache.cassandra.cql3.UntypedResultSet; import org.apache.cassandra.cql3.statements.schema.CreateTableStatement; +import org.apache.cassandra.dht.Range; +import org.apache.cassandra.dht.Token; import org.apache.cassandra.locator.RangesAtEndpoint; import org.apache.cassandra.net.Message; import org.apache.cassandra.repair.AbstractRepairTest; import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.repair.KeyspaceRepairManager; +import org.apache.cassandra.schema.TableId; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.SchemaConstants; @@ -206,6 +210,18 @@ public class LocalSessionTest extends AbstractRepairTest { return sessionHasData; } + + @Override + protected TableMetadata getTableMetadata(TableId tableId) + { + return cfm; + } + + @Override + protected List> getLocalRanges(String keyspace) + { + return Arrays.asList(RANGE1, RANGE2, RANGE3); + } } private static TableMetadata cfm;