From 9719f2d06346e4ceba89fcd9c821e03ec9c76fc2 Mon Sep 17 00:00:00 2001 From: Blake Eggleston Date: Tue, 2 Jun 2026 14:23:23 -0700 Subject: [PATCH 01/12] CEP-45: Fix MutationTrackingRepairTest.testRepairFailsOnTopologyChange The test wrote consistent data before installing the offset broadcast filter, which let the write path mark the writes reconciled and the sync coordinator complete before the topology change was observed. Write inconsistent data after the filter is in place instead. Cherry-picked from aci-cassandra 274dc829ea. The @Ignore(NO_TOPOLOGY_CHANGES) it added was removed again by 55af9fd957 once the underlying issue was fixed, so only the ignore reason constant is carried over here. --- .../distributed/test/repair/MutationTrackingRepairTest.java | 5 +++-- .../distributed/test/tracking/MutationTrackingUtils.java | 1 + 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingRepairTest.java b/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingRepairTest.java index f5ad9f7d26ff..87b1eb0457be 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingRepairTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingRepairTest.java @@ -881,11 +881,12 @@ public void testRepairSyncTimeout() @Test public void testRepairFailsOnTopologyChange() throws Exception { - insertData("tbl", 0, 50); - // Block offset broadcasts so the sync coordinator stays alive waiting IMessageFilters.Filter offsetFilter = CLUSTER.filters().verbs(Verb.MT_BROADCAST_LOG_OFFSETS.id).drop(); + // must use inconsistent data to prevent write process from marking the writes reconciled + insertDataWithInconsistency("tbl", 0, 50); + // Use a latch to detect when the sync request has been sent, meaning // the sync coordinator is active and tracking shard references CountDownLatch syncStarted = new CountDownLatch(1); diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingUtils.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingUtils.java index 6bec6394f017..6c37af2f5abc 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingUtils.java +++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingUtils.java @@ -60,6 +60,7 @@ public class MutationTrackingUtils public static class IgnoreReasons { + public static final String NO_TOPOLOGY_CHANGES = "NO_TOPOLOGY_CHANGES"; public static final String NO_RANGE_MOVEMENTS = "NO_RANGE_MOVEMENTS"; } From 87171df8294a096fd235f1988464a008a9f75c07 Mon Sep 17 00:00:00 2001 From: Blake Eggleston Date: Tue, 2 Jun 2026 14:37:01 -0700 Subject: [PATCH 02/12] CEP-45: Fix MutationTrackingSyncCoordinatorTest cancellation test Drop MT_SYNC_RSP so the coordinator never receives sync responses, rather than pausing offset broadcasts, which no longer keeps the coordinator waiting. Cherry-picked from aci-cassandra eff3d6de2b. --- .../MutationTrackingSyncCoordinatorTest.java | 15 +++++---------- 1 file changed, 5 insertions(+), 10 deletions(-) diff --git a/test/distributed/org/apache/cassandra/distributed/test/replication/MutationTrackingSyncCoordinatorTest.java b/test/distributed/org/apache/cassandra/distributed/test/replication/MutationTrackingSyncCoordinatorTest.java index 7a13f606f2b1..c1d82ef104fc 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/replication/MutationTrackingSyncCoordinatorTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/replication/MutationTrackingSyncCoordinatorTest.java @@ -29,6 +29,7 @@ import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.api.ConsistencyLevel; import org.apache.cassandra.distributed.test.TestBaseImpl; +import org.apache.cassandra.net.Verb; import org.apache.cassandra.repair.RepairJobDesc; import org.apache.cassandra.repair.SharedContext; import org.apache.cassandra.replication.MutationTrackingService; @@ -64,12 +65,6 @@ private String tableName(String suffix) return KS_NAME + suffix + '.' + TBL_NAME; } - private void pauseOffsetBroadcasts(Cluster cluster, boolean pause) - { - for (int i = 1; i <= cluster.size(); i++) - cluster.get(i).runOnInstance(() -> MutationTrackingService.instance().pauseOffsetBroadcast(pause)); - } - private static Range fullTokenRange() { return new Range<>( @@ -233,9 +228,6 @@ public void testSyncCoordinatorCancel() throws Throwable { createTrackedKeyspace(cluster, "4"); - // Pause offset broadcasts on all nodes to prevent sync from completing - pauseOffsetBroadcasts(cluster, true); - for (int i = 0; i < 100; i++) { cluster.coordinator(1).execute( @@ -243,7 +235,10 @@ public void testSyncCoordinatorCancel() throws Throwable ConsistencyLevel.ONE, i, i); } - // Start coordinator - it will be stuck waiting for offsets + // Drop sync responses so the coordinator never receives them + cluster.filters().verbs(Verb.MT_SYNC_RSP.id).drop(); + + // Start coordinator - it will be stuck waiting for sync responses Boolean wasCancelled = cluster.get(1).callOnInstance(() -> { Range range = fullTokenRange(); RepairJobDesc desc = new RepairJobDesc(TimeUUID.Generator.nextTimeUUID(), From f73ef376641957404f7ce5d55c50cbffcde25169 Mon Sep 17 00:00:00 2001 From: Blake Eggleston Date: Wed, 3 Jun 2026 10:45:36 -0700 Subject: [PATCH 03/12] ninja - minor mutation tracking test fixes IndexStreamingTest: include isWide in the parameterized test name. TrackedImportFailureTest: disable the background reconciler in importMissedActivation. Cherry-picked from aci-cassandra bde1f1daf7 and c3789023fa. --- .../cassandra/distributed/test/sai/IndexStreamingTest.java | 2 +- .../distributed/test/tracking/TrackedImportFailureTest.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/test/distributed/org/apache/cassandra/distributed/test/sai/IndexStreamingTest.java b/test/distributed/org/apache/cassandra/distributed/test/sai/IndexStreamingTest.java index db3d841bddf2..ebec55383b06 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/sai/IndexStreamingTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/sai/IndexStreamingTest.java @@ -69,7 +69,7 @@ private static int sstableStreamingComponentsCount() @Parameterized.Parameter(2) public boolean isWide; - @Parameterized.Parameters(name = "isLiteral={0}, isZeroCopyStreaming={1}") + @Parameterized.Parameters(name = "isLiteral={0}, isZeroCopyStreaming={1}, isWide={2}") public static List data() { List result = new ArrayList<>(); diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedImportFailureTest.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedImportFailureTest.java index 623013ce10b9..7973e1037d53 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedImportFailureTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedImportFailureTest.java @@ -73,7 +73,7 @@ public void importMissedActivationCommit() throws Throwable public void importMissedActivation(ActivationRequest.Phase phase) throws Throwable { int MISSED_ACTIVATION = 2; - try (Cluster cluster = cluster(TrackedTransferTestBase.ByteBuddyInjections.SkipActivation.install(MISSED_ACTIVATION))) + try (Cluster cluster = disableBackgroundReconciler(cluster(TrackedTransferTestBase.ByteBuddyInjections.SkipActivation.install(MISSED_ACTIVATION)))) { TrackedTransferTestBase.ByteBuddyInjections.SkipActivation.setup(cluster, phase); createSchema(cluster); From 8808541881c77f3c10afafba7ce8c514759bb9d1 Mon Sep 17 00:00:00 2001 From: Blake Eggleston Date: Wed, 7 Oct 2026 12:41:50 -0700 Subject: [PATCH 04/12] CEP-45: Explicitly configure STCS in TrackedUnreconciledPromotionTest --- .../db/compaction/TrackedUnreconciledPromotionTest.java | 1 + 1 file changed, 1 insertion(+) diff --git a/test/unit/org/apache/cassandra/db/compaction/TrackedUnreconciledPromotionTest.java b/test/unit/org/apache/cassandra/db/compaction/TrackedUnreconciledPromotionTest.java index 284e644128f8..7d36ffdc2395 100644 --- a/test/unit/org/apache/cassandra/db/compaction/TrackedUnreconciledPromotionTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/TrackedUnreconciledPromotionTest.java @@ -122,6 +122,7 @@ private static ColumnFamilyStore newTrackedTable() TableMetadata.builder(ks, "tbl") .addPartitionKeyColumn("k", Int32Type.instance) .addRegularColumn("v", Int32Type.instance) + .compaction(CompactionParams.stcs(Collections.emptyMap())) .build(); SchemaLoader.createKeyspace(ks, KeyspaceParams.simple(1, ReplicationType.tracked), tableMetadata); From 4146b4179ff44191d30e68a0ad5b24947af276b7 Mon Sep 17 00:00:00 2001 From: Blake Eggleston Date: Wed, 7 Oct 2026 13:06:41 -0700 Subject: [PATCH 05/12] CEP-45: Handle dropped tables in mutation journal recovery and isolate test journal state --- .../replication/MutationJournal.java | 29 ++++++ .../replication/UnreconciledMutations.java | 18 +++- .../replication/CoordinatorLogTest.java | 14 +++ .../UnreconciledMutationsTest.java | 92 +++++++++++++++++++ 4 files changed, 152 insertions(+), 1 deletion(-) diff --git a/src/java/org/apache/cassandra/replication/MutationJournal.java b/src/java/org/apache/cassandra/replication/MutationJournal.java index 22a66f61c685..2fc2dd9d49c3 100644 --- a/src/java/org/apache/cassandra/replication/MutationJournal.java +++ b/src/java/org/apache/cassandra/replication/MutationJournal.java @@ -53,6 +53,7 @@ import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.db.commitlog.CommitLogPosition; import org.apache.cassandra.db.partitions.PartitionUpdate; +import org.apache.cassandra.exceptions.UnknownTableException; import org.apache.cassandra.io.util.DataInputPlus; import org.apache.cassandra.io.util.DataOutputPlus; import org.apache.cassandra.io.util.File; @@ -427,6 +428,25 @@ public void replayStaticSegments() { replay(new DeserializedRecordConsumer<>(MutationSerializer.INSTANCE) { + @Override + public void accept(long segment, int position, ShortMutationId key, ByteBuffer buffer, int userVersion) + { + try + { + super.accept(segment, position, key, buffer, userVersion); + } + catch (RuntimeException e) + { + if (e.getCause() instanceof UnknownTableException) + { + UnknownTableException ute = (UnknownTableException) e.getCause(); + logger.warn("Skipping replaying mutation {} from unknown (probably removed) table with id {}", key, ute.id); + return; + } + throw e; + } + } + @Override protected void accept(long segmentId, int position, ShortMutationId key, Mutation value) { @@ -943,6 +963,15 @@ public ActiveOffsetRanges rebuild(Descriptor descriptor, KeySupport Date: Wed, 7 Oct 2026 13:10:56 -0700 Subject: [PATCH 06/12] Use skiplist memtable in SplitDomainMemtableTest --- .../cassandra/db/memtable/SplitDomainMemtableTest.java | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/test/unit/org/apache/cassandra/db/memtable/SplitDomainMemtableTest.java b/test/unit/org/apache/cassandra/db/memtable/SplitDomainMemtableTest.java index b8d8ff1c72fc..1ca389da74e1 100644 --- a/test/unit/org/apache/cassandra/db/memtable/SplitDomainMemtableTest.java +++ b/test/unit/org/apache/cassandra/db/memtable/SplitDomainMemtableTest.java @@ -62,8 +62,10 @@ import org.apache.cassandra.replication.MutationJournal; import org.apache.cassandra.replication.MutationTrackingService; import org.apache.cassandra.schema.KeyspaceParams; +import org.apache.cassandra.schema.MemtableParams; import org.apache.cassandra.schema.ReplicationType; import org.apache.cassandra.schema.TableMetadata; +import org.apache.cassandra.schema.TableParams; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.concurrent.OpOrder; @@ -106,6 +108,8 @@ private static ColumnFamilyStore newTrackedTable() TableMetadata metadata = TableMetadata.builder(ks, "tbl") .addPartitionKeyColumn("k", Int32Type.instance) .addRegularColumn("v", Int32Type.instance) + // test relies on Memtable#lastToken, so we explicitly set skiplist + .params(TableParams.builder().memtable(MemtableParams.get("skiplist")).build()) .build(); SchemaLoader.createKeyspace(ks, KeyspaceParams.simple(1, ReplicationType.tracked), metadata); ColumnFamilyStore cfs = Keyspace.open(ks).getColumnFamilyStore("tbl"); @@ -228,6 +232,7 @@ private static void appendToCommitLog() TableMetadata metadata = TableMetadata.builder(ks, "tbl") .addPartitionKeyColumn("k", Int32Type.instance) .addRegularColumn("v", Int32Type.instance) + .params(TableParams.builder().memtable(MemtableParams.get("skiplist")).build()) .build(); SchemaLoader.createKeyspace(ks, KeyspaceParams.simple(1, ReplicationType.untracked), metadata); From 9dcf1ade71ec6847008efe9621dde5dea360628b Mon Sep 17 00:00:00 2001 From: Blake Eggleston Date: Wed, 7 Oct 2026 13:25:09 -0700 Subject: [PATCH 07/12] CEP-45: Enable Feature.NETWORK in MutationTrackingReadRepairTest --- .../distributed/test/repair/MutationTrackingReadRepairTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingReadRepairTest.java b/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingReadRepairTest.java index 7134a9d5cdbd..bf4a437f6d18 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingReadRepairTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingReadRepairTest.java @@ -104,7 +104,7 @@ public static void setupCluster() throws IOException .set("repair.retries.max_attempts", 10) .set("repair.retries.base_sleep_time", "100ms") .set("repair.retries.max_sleep_time", "500ms") - .with(Feature.GOSSIP)) + .with(Feature.GOSSIP, Feature.NETWORK)) .start(); } From a13866831e689634cf28f494d6dc39a596475707 Mon Sep 17 00:00:00 2001 From: Blake Eggleston Date: Wed, 7 Oct 2026 15:02:08 -0700 Subject: [PATCH 08/12] CEP-45: Track journal segment references via coordinator log offset overlap --- .../db/virtual/MutationTrackingTables.java | 2 +- .../org/apache/cassandra/journal/Journal.java | 2 +- .../replication/CoordinatorLogOffsets.java | 9 + .../ImmutableCoordinatorLogOffsets.java | 15 + .../replication/MutationJournal.java | 71 ++- .../MutationJournalSegmentCompactor.java | 2 +- .../apache/cassandra/replication/Offsets.java | 22 + .../replication/SegmentReferenceTracker.java | 212 ++++--- .../MutationJournalSegmentRefcountTest.java | 15 + .../replication/MutationJournalTest.java | 33 + .../cassandra/replication/OffsetsTest.java | 43 ++ .../SegmentReferenceTrackerTest.java | 580 ++++++++++++------ 12 files changed, 727 insertions(+), 279 deletions(-) diff --git a/src/java/org/apache/cassandra/db/virtual/MutationTrackingTables.java b/src/java/org/apache/cassandra/db/virtual/MutationTrackingTables.java index 4c3b80142239..9af071fe65b1 100644 --- a/src/java/org/apache/cassandra/db/virtual/MutationTrackingTables.java +++ b/src/java/org/apache/cassandra/db/virtual/MutationTrackingTables.java @@ -107,7 +107,7 @@ public DataSet data() for (Segment segment : MutationJournal.instance().getAllSegments()) { - List referringSstables = referenceTracker.referrerDescriptors(segment.id()); + List referringSstables = referenceTracker.referrerDescriptors(segment); result.row(segment.id()) .column(IS_ACTIVE, segment instanceof ActiveSegment) .column(BYTES_ON_DISK, segment.segmentSizeOnDisk()) diff --git a/src/java/org/apache/cassandra/journal/Journal.java b/src/java/org/apache/cassandra/journal/Journal.java index f9569747f26d..9f3f66659b1f 100644 --- a/src/java/org/apache/cassandra/journal/Journal.java +++ b/src/java/org/apache/cassandra/journal/Journal.java @@ -1019,7 +1019,7 @@ public ActiveSegment currentActiveSegment() return currentSegment; } - @Nullable protected Segment getSegment(long timestamp) + @Nullable public Segment getSegment(long timestamp) { return segments().get(timestamp); } diff --git a/src/java/org/apache/cassandra/replication/CoordinatorLogOffsets.java b/src/java/org/apache/cassandra/replication/CoordinatorLogOffsets.java index 5a394fc08df9..b6ae38171de4 100644 --- a/src/java/org/apache/cassandra/replication/CoordinatorLogOffsets.java +++ b/src/java/org/apache/cassandra/replication/CoordinatorLogOffsets.java @@ -38,6 +38,10 @@ public interface CoordinatorLogOffsets interface Mutations extends Iterable { O offsets(long logId); + default O getOffsets(long logId) + { + return offsets(logId); + } int size(); default boolean isEmpty() { @@ -47,6 +51,11 @@ default boolean isEmpty() Mutations mutations(); + default boolean isEmpty() + { + return mutations().isEmpty() && transfers().isEmpty(); + } + default ActivatedTransfers transfers() { return ActivatedTransfers.EMPTY; diff --git a/src/java/org/apache/cassandra/replication/ImmutableCoordinatorLogOffsets.java b/src/java/org/apache/cassandra/replication/ImmutableCoordinatorLogOffsets.java index d2de41ca8bb3..15a052794f48 100644 --- a/src/java/org/apache/cassandra/replication/ImmutableCoordinatorLogOffsets.java +++ b/src/java/org/apache/cassandra/replication/ImmutableCoordinatorLogOffsets.java @@ -29,6 +29,7 @@ import javax.annotation.concurrent.NotThreadSafe; +import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.ImmutableSet; import com.google.common.collect.Iterators; @@ -160,6 +161,14 @@ public Builder add(MutationId mutationId) return this; } + @VisibleForTesting + public Builder add(long logId, int start, int end) + { + ids.computeIfAbsent(logId, k -> new Offsets.Immutable.Builder(new CoordinatorLogId(k))) + .add(start, end); + return this; + } + private Builder addAll(CoordinatorLogOffsets.Mutations mutations) { for (long log : mutations) @@ -292,6 +301,12 @@ public Offsets.Immutable offsets(long logId) return offsets; } + @Override + public Offsets.Immutable getOffsets(long logId) + { + return ids.get(logId); + } + @Override public int size() { diff --git a/src/java/org/apache/cassandra/replication/MutationJournal.java b/src/java/org/apache/cassandra/replication/MutationJournal.java index 2fc2dd9d49c3..759cbdc99385 100644 --- a/src/java/org/apache/cassandra/replication/MutationJournal.java +++ b/src/java/org/apache/cassandra/replication/MutationJournal.java @@ -77,6 +77,7 @@ import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.TableId; +import org.apache.cassandra.service.StorageService; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.utils.Crc; import org.apache.cassandra.utils.concurrent.OpOrder; @@ -204,7 +205,9 @@ protected void closeActiveSegmentAndOpenAsStatic(ActiveSegment journal.compactor().triggerNow()); + segmentReferenceTracker = new SegmentReferenceTracker(() -> journal.compactor().triggerNow(), + StorageService.instance::getLocalHostUUID, + journal::getAllSegments); } /** @@ -744,6 +747,25 @@ protected static long range(int minOffset, int maxOffset) abstract Map asMap(); + @FunctionalInterface + interface RangePredicate + { + boolean test(long logId, int minOffset, int maxOffset); + } + + abstract boolean anyMatch(RangePredicate predicate); + + public boolean overlaps(CoordinatorLogOffsets offsets) + { + if (offsets == null || offsets.isEmpty()) + return false; + + return anyMatch((logId, minOffset, maxOffset) -> { + Offsets sstOffsets = offsets.mutations().getOffsets(logId); + return sstOffsets != null && !sstOffsets.isEmpty() && sstOffsets.overlaps(minOffset, maxOffset); + }); + } + @Override public String toString() { @@ -784,6 +806,22 @@ protected Map asMap() return ranges; } + @Override + boolean anyMatch(RangePredicate predicate) + { + for (Map.Entry entry : ranges.entrySet()) + { + Long val = entry.getValue(); + if (val != null) + { + long range = val.longValue(); + if (predicate.test(entry.getKey().longValue(), minOffset(range), maxOffset(range))) + return true; + } + } + return false; + } + @Override public boolean mayContain(ShortMutationId id) { @@ -876,6 +914,37 @@ public boolean mayContain(ShortMutationId id) return range != NO_VALUE && mayContain(range, id); } + @Override + boolean anyMatch(RangePredicate predicate) + { + for (Long2LongHashMap.EntryIterator iter = ranges.entrySet().iterator(); iter.hasNext();) + { + iter.next(); + long logId = iter.getLongKey(); + long range = iter.getLongValue(); + if (predicate.test(logId, minOffset(range), maxOffset(range))) + return true; + } + return false; + } + + @VisibleForTesting + static StaticOffsetRanges of(long logId, int minOffset, int maxOffset) + { + Long2LongHashMap map = new Long2LongHashMap(1, 0.65f, NO_VALUE, false); + map.put(logId, range(minOffset, maxOffset)); + return new StaticOffsetRanges(map); + } + + @VisibleForTesting + static StaticOffsetRanges of(Map rangesMap) + { + Long2LongHashMap map = new Long2LongHashMap(rangesMap.size(), 0.65f, NO_VALUE, false); + for (Map.Entry entry : rangesMap.entrySet()) + map.put(entry.getKey().longValue(), entry.getValue().longValue()); + return new StaticOffsetRanges(map); + } + static StaticOffsetRanges read(DataInputPlus in) throws IOException { CRC32 crc = Crc.crc32(); diff --git a/src/java/org/apache/cassandra/replication/MutationJournalSegmentCompactor.java b/src/java/org/apache/cassandra/replication/MutationJournalSegmentCompactor.java index bc16343571ab..d25e2261c480 100644 --- a/src/java/org/apache/cassandra/replication/MutationJournalSegmentCompactor.java +++ b/src/java/org/apache/cassandra/replication/MutationJournalSegmentCompactor.java @@ -84,7 +84,7 @@ public Collection> select(Collection segment : candidates) { if (!segment.metadata().needsReplay() - && !journal.segmentReferenceTracker().isReferenced(segment.id()) + && !journal.segmentReferenceTracker().isReferenced(segment) && ((MutationJournal.StaticOffsetRanges) segment.keyStats()).isFullyCovered(durablyReconciled)) selected.add(segment); } diff --git a/src/java/org/apache/cassandra/replication/Offsets.java b/src/java/org/apache/cassandra/replication/Offsets.java index 4430a95556f5..64659538047b 100644 --- a/src/java/org/apache/cassandra/replication/Offsets.java +++ b/src/java/org/apache/cassandra/replication/Offsets.java @@ -209,6 +209,28 @@ public boolean containsRange(int minOffset, int maxOffset) return maxOffset <= bounds[end]; } + /** + * @return whether any offset in the inclusive range [minOffset, maxOffset] is present in these Offsets + */ + public boolean overlaps(int minOffset, int maxOffset) + { + if (size == 0 || minOffset > maxOffset) + return false; + + if (maxOffset < bounds[0] || minOffset > bounds[size - 1]) + return false; + + int pos = Arrays.binarySearch(bounds, 0, size, minOffset); + if (pos >= 0) + return true; + + int insertionPoint = -pos - 1; + if (insertionPoint % 2 != 0) + return true; + + return bounds[insertionPoint] <= maxOffset; + } + public void digest(Digest digest) { digest.updateWithLong(logId.asLong()); diff --git a/src/java/org/apache/cassandra/replication/SegmentReferenceTracker.java b/src/java/org/apache/cassandra/replication/SegmentReferenceTracker.java index 45c9f9551773..b636c24f1b90 100644 --- a/src/java/org/apache/cassandra/replication/SegmentReferenceTracker.java +++ b/src/java/org/apache/cassandra/replication/SegmentReferenceTracker.java @@ -20,22 +20,25 @@ import java.util.ArrayList; import java.util.Collection; import java.util.Collections; +import java.util.HashMap; import java.util.HashSet; -import java.util.Iterator; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.UUID; import java.util.concurrent.locks.ReentrantLock; -import java.util.function.LongConsumer; import java.util.function.Supplier; import com.google.common.annotations.VisibleForTesting; import org.agrona.collections.Long2ObjectHashMap; +import org.agrona.collections.LongHashSet; +import org.agrona.collections.LongHashSet.LongIterator; -import org.apache.cassandra.db.commitlog.CommitLogPosition; -import org.apache.cassandra.db.commitlog.IntervalSet; +import org.apache.cassandra.db.Mutation; import org.apache.cassandra.io.sstable.format.SSTableReader; +import org.apache.cassandra.journal.KeyStats; +import org.apache.cassandra.journal.Segment; import org.apache.cassandra.notifications.INotification; import org.apache.cassandra.notifications.INotificationConsumer; import org.apache.cassandra.notifications.InitialSSTableAddedNotification; @@ -43,59 +46,47 @@ import org.apache.cassandra.notifications.SSTableListChangedNotification; import org.apache.cassandra.notifications.SSTableRepairStatusChanged; import org.apache.cassandra.schema.ReplicationType; -import org.apache.cassandra.service.StorageService; /** - * Tracks, for each mutation journal segment, the set of local unrepaired sstables of tracked tables that - * reference it. + * Tracks, for each mutation journal segment, whether any unreconciled sstables could reference mutation ids it contains * - *

A sstable references a segment iff the sstable's {@code StatsMetadata.commitLogIntervals} — - * which for tracked tables stores mutation journal positions, not commit log positions — covers - * any position within that segment. The granularity is coarse: the referenced range is the union - * of the memtable lower/upper bounds for the sstable's lineage. This matches existing commit log - * retention semantics; it can include a segment with no actual data from this sstable, which only - * defers (never incorrectly enables) segment dropping. + *

An sstable references a segment if its {@code StatsMetadata.coordinatorLogOffsets} overlap with the segment's + * recorded offset ranges for any coordinator log. * - *

Used by {@link MutationJournal} to decide when a static segment may be dropped: a segment may be - * dropped only once it has no unrepaired references and {@code !needsReplay}. This guarantees the - * journal can rebuild any unrepaired sstable from the journal if minority writes need to be filtered - * out (CASSANDRA-21407). + *

Used by {@link MutationJournal} to decide when a static segment can be dropped: a segment can be + * dropped only when it has no unrepaired references and {@code !needsReplay}. This guarantees the journal can rebuild + * any unrepaired sstable from the journal if minority writes need to be filtered out (CASSANDRA-21407). * *

A per-segment set of referrers (rather than a bare refcount) is kept so that individual * referrers can be located and dropped from a segment's set — needed to reason about keyspaces migrating * to and from tracked replication (CASSANDRA-21406) and to surface which sstables hold a segment. + * */ public class SegmentReferenceTracker implements INotificationConsumer { - // Guards both referrersBySegment and trackedSstables to keep transitions atomic across notifications. private final ReentrantLock lock = new ReentrantLock(); - // segment id -> set of local unrepaired sstables referencing it + // Index mapping each segment id to the set of unrepaired SSTables that overlap it. + // Used by the compactor to quickly determine if a segment can be dropped. private final Long2ObjectHashMap> referrersBySegment = new Long2ObjectHashMap<>(); - // Sstables we currently hold refs for. - // Required so SSTableRepairStatusChanged can transition an sstable in/out without the notification - // having to carry the previous repair state. - private final Set trackedSSTables = new HashSet<>(); + // Reverse index mapping each tracked SSTable to the journal segment ids it references. + // Used to remove references across all impacted segments when an SSTable is deleted or repaired. + private final Map segmentsBySSTable = new HashMap<>(); - // Invoked whenever a notification drives at least one segment's reference count to zero, - // so the journal can attempt to drop the now-unreferenced segment(s). private final Runnable onSegmentsUnreferenced; - // Resolves the local host id. May return null very early in startup, in which case no sstable is - // treated as locally originated. private final Supplier localHostIdSupplier; - public SegmentReferenceTracker(Runnable onSegmentsUnreferenced) - { - this(onSegmentsUnreferenced, StorageService.instance::getLocalHostUUID); - } + private final Supplier>> allSegmentsSupplier; - @VisibleForTesting - SegmentReferenceTracker(Runnable onSegmentsUnreferenced, Supplier localHostIdSupplier) + public SegmentReferenceTracker(Runnable onSegmentsUnreferenced, + Supplier localHostIdSupplier, + Supplier>> allSegmentsSupplier) { this.onSegmentsUnreferenced = onSegmentsUnreferenced; this.localHostIdSupplier = localHostIdSupplier; + this.allSegmentsSupplier = allSegmentsSupplier; } @Override @@ -111,6 +102,15 @@ else if (notification instanceof SSTableRepairStatusChanged) onRepairStatusChanged(((SSTableRepairStatusChanged) notification).sstables); } + /** + * @param segment the journal segment + * @return whether any unrepaired local sstable references the given segment + */ + public boolean isReferenced(Segment segment) + { + return segment != null && isReferenced(segment.id()); + } + /** * @param segmentId the identifier for the segment * @return whether any unrepaired local sstable references the given segment id @@ -120,7 +120,15 @@ boolean isReferenced(long segmentId) lock.lock(); try { - return referrersBySegment.containsKey(segmentId); + Set referrers = referrersBySegment.get(segmentId); + if (referrers == null) + return false; + if (referrers.isEmpty()) + { + referrersBySegment.remove(segmentId); + return false; + } + return true; } finally { @@ -148,8 +156,6 @@ private void onListChanged(SSTableListChangedNotification notification) lock.lock(); try { - // Process additions before removals so refcounts are never observed briefly empty - // between a compaction's input drop and output add when both span the same segment. for (SSTableReader sstable : notification.added) acquireIfTracked(sstable); for (SSTableReader sstable : notification.removed) @@ -186,8 +192,7 @@ private void onRepairStatusChanged(Collection changed) } /** - * Release every reference currently held for the given sstables, dropping each from every segment's referrer - * set. + * Release every reference currently held for the given sstables. */ public void evict(Iterable sstables) { @@ -214,9 +219,12 @@ public Set trackedSSTables() lock.lock(); try { - if (trackedSSTables.isEmpty()) + if (segmentsBySSTable.isEmpty()) + return Set.of(); + segmentsBySSTable.entrySet().removeIf(entry -> entry.getValue() == null || entry.getValue().isEmpty()); + if (segmentsBySSTable.isEmpty()) return Set.of(); - return Set.copyOf(trackedSSTables); + return Set.copyOf(segmentsBySSTable.keySet()); } finally { @@ -224,10 +232,6 @@ public Set trackedSSTables() } } - /** - * @return true when all these conditions are true, the sstable is locally originated, unrepaired, carries - * tracked mutations, and it belongs to a tracked table - */ boolean shouldTrack(SSTableReader sstable) { return isLocallyOriginated(sstable) @@ -251,58 +255,77 @@ private boolean isLocallyOriginated(SSTableReader sstable) private void acquireIfTracked(SSTableReader sstable) { - if (shouldTrack(sstable) && trackedSSTables.add(sstable)) - forEachSegment(sstable, segmentId -> - referrersBySegment.computeIfAbsent(segmentId, k -> new HashSet<>()).add(sstable)); + if (!shouldTrack(sstable) || segmentsBySSTable.containsKey(sstable)) + return; + + LongHashSet segmentIds = new LongHashSet(); + if (allSegmentsSupplier != null) + { + Iterable> allSegments = allSegmentsSupplier.get(); + if (allSegments != null) + { + for (Segment segment : allSegments) + { + KeyStats keyStats = segment.keyStats(); + if (keyStats instanceof MutationJournal.OffsetRanges) + { + MutationJournal.OffsetRanges segRanges = (MutationJournal.OffsetRanges) keyStats; + if (segRanges.overlaps(sstable.getCoordinatorLogOffsets())) + { + segmentIds.add(segment.id()); + referrersBySegment.computeIfAbsent(segment.id(), k -> new HashSet<>()).add(sstable); + } + } + } + } + } + if (!segmentIds.isEmpty()) + segmentsBySSTable.put(sstable, segmentIds); } - /** - * @return true if releasing this sstable emptied at least one segment's referrer set - */ private boolean releaseIfTracked(SSTableReader sstable) { - if (!trackedSSTables.remove(sstable)) + LongHashSet segmentIds = segmentsBySSTable.remove(sstable); + if (segmentIds == null || segmentIds.isEmpty()) return false; - boolean[] anyEmptied = { false }; - forEachSegment(sstable, segmentId -> { + + boolean anyEmptied = false; + LongIterator idIterator = segmentIds.iterator(); + while (idIterator.hasNext()) + { + long segmentId = idIterator.nextValue(); Set referrers = referrersBySegment.get(segmentId); if (referrers != null && referrers.remove(sstable) && referrers.isEmpty()) { referrersBySegment.remove(segmentId); - anyEmptied[0] = true; + anyEmptied = true; } - }); - return anyEmptied[0]; - } - - private static void forEachSegment(SSTableReader sstable, LongConsumer consumer) - { - IntervalSet intervals = sstable.getSSTableMetadata().commitLogIntervals; - if (intervals.isEmpty()) - return; - - // IntervalSet guarantees starts and ends are returned in matching order. - Iterator startIt = intervals.starts().iterator(); - Iterator endIt = intervals.ends().iterator(); - while (startIt.hasNext()) - { - CommitLogPosition start = startIt.next(); - CommitLogPosition end = endIt.next(); - for (long s = start.segmentId; s <= end.segmentId; s++) - consumer.accept(s); } + return anyEmptied; } /** * Number of unrepaired local sstables currently holding the given segment (for diagnostics / the vtable). */ - int referenceCount(long segmentId) + public int referenceCount(Segment segment) + { + return segment != null ? referenceCount(segment.id()) : 0; + } + + public int referenceCount(long segmentId) { lock.lock(); try { Set referrers = referrersBySegment.get(segmentId); - return referrers == null ? 0 : referrers.size(); + if (referrers == null) + return 0; + if (referrers.isEmpty()) + { + referrersBySegment.remove(segmentId); + return 0; + } + return referrers.size(); } finally { @@ -313,6 +336,11 @@ int referenceCount(long segmentId) /** * Sorted base filenames of the sstables currently holding the given segment (for diagnostics / the vtable). */ + public List referrerDescriptors(Segment segment) + { + return segment != null ? referrerDescriptors(segment.id()) : Collections.emptyList(); + } + public List referrerDescriptors(long segmentId) { lock.lock(); @@ -320,10 +348,14 @@ public List referrerDescriptors(long segmentId) { Set referrers = referrersBySegment.get(segmentId); if (referrers == null || referrers.isEmpty()) + { + if (referrers != null) + referrersBySegment.remove(segmentId); return Collections.emptyList(); + } List names = new ArrayList<>(referrers.size()); for (SSTableReader sstable : referrers) - names.add(sstable.descriptor.baseFile().name()); + names.add(sstable.descriptor != null ? sstable.descriptor.baseFile().name() : sstable.getFilename()); Collections.sort(names); return names; } @@ -333,6 +365,12 @@ public List referrerDescriptors(long segmentId) } } + @VisibleForTesting + long referenceCountForTesting(Segment segment) + { + return referenceCount(segment); + } + @VisibleForTesting long referenceCountForTesting(long segmentId) { @@ -345,7 +383,8 @@ int trackedSstableCountForTesting() lock.lock(); try { - return trackedSSTables.size(); + segmentsBySSTable.entrySet().removeIf(entry -> entry.getValue() == null || entry.getValue().isEmpty()); + return segmentsBySSTable.size(); } finally { @@ -360,6 +399,7 @@ void addReferenceForTesting(long segmentId, SSTableReader referrer) try { referrersBySegment.computeIfAbsent(segmentId, k -> new HashSet<>()).add(referrer); + segmentsBySSTable.computeIfAbsent(referrer, k -> new LongHashSet()).add(segmentId); } finally { @@ -370,16 +410,30 @@ void addReferenceForTesting(long segmentId, SSTableReader referrer) @VisibleForTesting void removeReferenceForTesting(long segmentId, SSTableReader referrer) { + boolean removed = false; lock.lock(); try { Set referrers = referrersBySegment.get(segmentId); - if (referrers != null && referrers.remove(referrer) && referrers.isEmpty()) - referrersBySegment.remove(segmentId); + if (referrers != null && referrers.remove(referrer)) + { + if (referrers.isEmpty()) + referrersBySegment.remove(segmentId); + removed = true; + } + LongHashSet segmentIds = segmentsBySSTable.get(referrer); + if (segmentIds != null) + { + segmentIds.remove(segmentId); + if (segmentIds.isEmpty()) + segmentsBySSTable.remove(referrer); + } } finally { lock.unlock(); } + if (removed) + onSegmentsUnreferenced.run(); } } diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationJournalSegmentRefcountTest.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationJournalSegmentRefcountTest.java index b21f3bddcd5a..b5bcd808b41d 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationJournalSegmentRefcountTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationJournalSegmentRefcountTest.java @@ -84,6 +84,21 @@ public void testSegmentRetainedUntilSSTableRepaired() throws Throwable before, after); })); + int preBounceCount = cluster.get(1).callOnInstance(() -> MutationJournal.instance().countStaticSegmentsForTesting()); + assertTrue(preBounceCount > 0); + + // Restart a replica to confirm segment references survive restart and keep static segments pinned. + cluster.get(1).shutdown().get(); + cluster.get(1).startup(); + cluster.get(1).runOnInstance(() -> { + int postBounceCount = MutationJournal.instance().countStaticSegmentsForTesting(); + assertEquals(preBounceCount, postBounceCount); + + MutationTrackingService.instance().persistLogStateForTesting(); + int postDropCount = MutationJournal.instance().countStaticSegmentsForTesting(); + assertEquals(preBounceCount, postDropCount); + }); + // Restore broadcast, exchange witnesses, and persist so isDurablyReconciled is now true everywhere. cluster.filters().reset(); cluster.forEach(i -> i.runOnInstance(() -> MutationTrackingService.instance().broadcastOffsetsForTesting())); diff --git a/test/unit/org/apache/cassandra/replication/MutationJournalTest.java b/test/unit/org/apache/cassandra/replication/MutationJournalTest.java index b9805b01df6e..4d75a1e0dd55 100644 --- a/test/unit/org/apache/cassandra/replication/MutationJournalTest.java +++ b/test/unit/org/apache/cassandra/replication/MutationJournalTest.java @@ -146,6 +146,39 @@ public void testActiveOffsetRanges() } } + @Test + public void testActiveOffsetRangesOverlaps() + { + ActiveOffsetRanges ranges = new ActiveOffsetRanges(); + ranges.update(id(100L, 10)); + ranges.update(id(100L, 50)); + ranges.update(id(200L, 30)); + ranges.update(id(200L, 60)); + + assertFalse(ranges.overlaps(null)); + assertFalse(ranges.overlaps(ImmutableCoordinatorLogOffsets.NONE)); + + ImmutableCoordinatorLogOffsets noMatchLog = new ImmutableCoordinatorLogOffsets.Builder() + .add(300L, 10, 50) + .build(); + assertFalse(ranges.overlaps(noMatchLog)); + + ImmutableCoordinatorLogOffsets noMatchOffset = new ImmutableCoordinatorLogOffsets.Builder() + .add(100L, 60, 80) + .build(); + assertFalse(ranges.overlaps(noMatchOffset)); + + ImmutableCoordinatorLogOffsets match100 = new ImmutableCoordinatorLogOffsets.Builder() + .add(100L, 5, 15) + .build(); + assertTrue(ranges.overlaps(match100)); + + ImmutableCoordinatorLogOffsets match200 = new ImmutableCoordinatorLogOffsets.Builder() + .add(200L, 40, 45) + .build(); + assertTrue(ranges.overlaps(match200)); + } + @Test public void testStaticOffsetRanges() { diff --git a/test/unit/org/apache/cassandra/replication/OffsetsTest.java b/test/unit/org/apache/cassandra/replication/OffsetsTest.java index aff871337f06..d6348b7c1115 100644 --- a/test/unit/org/apache/cassandra/replication/OffsetsTest.java +++ b/test/unit/org/apache/cassandra/replication/OffsetsTest.java @@ -908,4 +908,47 @@ public void testContainsRange() assertFalse(offsets.containsRange(9, 9)); } } + + @Test + public void testOverlaps() + { + { + Offsets.Mutable offsets = offsets(); + assertFalse(offsets.overlaps(0, 1)); + assertFalse(offsets.overlaps(2, 4)); + } + + { + Offsets.Mutable offsets = offsets(2, 4); + + assertTrue(offsets.overlaps(2, 4)); + assertTrue(offsets.overlaps(3, 4)); + assertTrue(offsets.overlaps(2, 3)); + assertTrue(offsets.overlaps(1, 2)); + assertTrue(offsets.overlaps(1, 3)); + assertTrue(offsets.overlaps(1, 5)); + assertTrue(offsets.overlaps(4, 5)); + + assertFalse(offsets.overlaps(0, 1)); + assertFalse(offsets.overlaps(5, 6)); + assertFalse(offsets.overlaps(4, 2)); + } + + { + Offsets.Mutable offsets = offsets(2, 4, 6, 8); + + assertTrue(offsets.overlaps(2, 4)); + assertTrue(offsets.overlaps(6, 8)); + assertTrue(offsets.overlaps(0, 2)); + assertTrue(offsets.overlaps(3, 5)); + assertTrue(offsets.overlaps(4, 6)); + assertTrue(offsets.overlaps(5, 7)); + assertTrue(offsets.overlaps(7, 9)); + assertTrue(offsets.overlaps(0, 10)); + + assertFalse(offsets.overlaps(0, 1)); + assertFalse(offsets.overlaps(5, 5)); + assertFalse(offsets.overlaps(9, 10)); + } + } } diff --git a/test/unit/org/apache/cassandra/replication/SegmentReferenceTrackerTest.java b/test/unit/org/apache/cassandra/replication/SegmentReferenceTrackerTest.java index c1504ae205d4..4d68bd99f9c5 100644 --- a/test/unit/org/apache/cassandra/replication/SegmentReferenceTrackerTest.java +++ b/test/unit/org/apache/cassandra/replication/SegmentReferenceTrackerTest.java @@ -17,7 +17,9 @@ */ package org.apache.cassandra.replication; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.UUID; import java.util.concurrent.atomic.AtomicReference; import java.util.function.BooleanSupplier; @@ -25,13 +27,14 @@ import org.junit.Test; import org.mockito.Mockito; +import org.apache.cassandra.db.Mutation; import org.apache.cassandra.db.Slice; -import org.apache.cassandra.db.commitlog.CommitLogPosition; import org.apache.cassandra.db.commitlog.IntervalSet; import org.apache.cassandra.db.compaction.OperationType; import org.apache.cassandra.db.rows.Cell; import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.io.sstable.metadata.StatsMetadata; +import org.apache.cassandra.journal.Segment; import org.apache.cassandra.notifications.InitialSSTableAddedNotification; import org.apache.cassandra.notifications.SSTableAddedNotification; import org.apache.cassandra.notifications.SSTableListChangedNotification; @@ -51,19 +54,24 @@ public class SegmentReferenceTrackerTest { private static final UUID LOCAL_HOST = UUID.randomUUID(); private static final UUID REMOTE_HOST = UUID.randomUUID(); + private static final long LOG_ID = 100L; @Test public void testInitialAddRefsEverySegmentInTheInterval() { - int startSegment = 5; - int endSegment = 7; - SegmentReferenceTracker tracker = newTracker(); - SSTableReader sstable = unrepaired(intervals(startSegment, 0, endSegment, 100)); + Map> segments = new HashMap<>(); + segments.put(4L, mockSegment(4L, LOG_ID, 0, 5)); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 20)); + segments.put(6L, mockSegment(6L, LOG_ID, 25, 40)); + segments.put(7L, mockSegment(7L, LOG_ID, 45, 60)); + segments.put(8L, mockSegment(8L, LOG_ID, 65, 80)); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader sstable = unrepaired(offsets(LOG_ID, 10, 60)); tracker.handleNotification(new InitialSSTableAddedNotification(List.of(sstable)), null); - // Coarse range covers segments 5..7 inclusive. - for (long segment = startSegment; segment <= endSegment; segment++) + for (long segment = 5; segment <= 7; segment++) assertEquals("segment " + segment, 1L, tracker.referenceCountForTesting(segment)); assertFalse(tracker.isReferenced(4)); assertFalse(tracker.isReferenced(8)); @@ -73,14 +81,17 @@ public void testInitialAddRefsEverySegmentInTheInterval() @Test public void testAddedRepairedSSTableHoldsNoRefs() { - int startSegment = 5; - int endSegment = 7; - SegmentReferenceTracker tracker = newTracker(); - SSTableReader sstable = repaired(intervals(startSegment, 0, endSegment, 100)); + Map> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 20)); + segments.put(6L, mockSegment(6L, LOG_ID, 25, 40)); + segments.put(7L, mockSegment(7L, LOG_ID, 45, 60)); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader sstable = repaired(offsets(LOG_ID, 10, 60)); tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); - for (long segment = startSegment; segment <= endSegment; segment++) + for (long segment = 5; segment <= 7; segment++) assertFalse("segment " + segment, tracker.isReferenced(segment)); assertEquals(0, tracker.trackedSstableCountForTesting()); } @@ -88,48 +99,45 @@ public void testAddedRepairedSSTableHoldsNoRefs() @Test public void testUnrepairedSSTableWithoutCoordinatorLogOffsetsHoldsNoRefs() { - SegmentReferenceTracker tracker = newTracker(); - // Unrepaired, but carries no tracked mutations (empty coordinatorLogOffsets) -> not tracked. This is the - // untracked / pre-migration case: such an sstable's commitLogIntervals reference the commit log, not the - // mutation journal, so it must never hold a segment reference. - SSTableReader sstable = sstable(intervals(5, 0, 7, 100), () -> false, coordinatorLogOffsets(false)); + Map> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 20)); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader sstable = sstable(ImmutableCoordinatorLogOffsets.NONE, () -> false, LOCAL_HOST, ReplicationType.tracked); tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); - for (long segment = 5; segment <= 7; segment++) - assertFalse("segment " + segment, tracker.isReferenced(segment)); + assertFalse(tracker.isReferenced(5)); assertEquals(0, tracker.trackedSstableCountForTesting()); } @Test public void testStreamedSSTableHoldsNoRefs() { - SegmentReferenceTracker tracker = newTracker(); - // Unrepaired with tracked mutations, but streamed from another host: its commitLogIntervals reference that - // host's journal segments, not ours, so it must never hold a local segment reference (CASSANDRA-21406). - SSTableReader sstable = sstable(intervals(5, 0, 7, 100), () -> false, coordinatorLogOffsets(true), REMOTE_HOST); + Map> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 20)); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader sstable = sstable(offsets(LOG_ID, 10, 20), () -> false, REMOTE_HOST, ReplicationType.tracked); tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); - for (long segment = 5; segment <= 7; segment++) - assertFalse("segment " + segment, tracker.isReferenced(segment)); + assertFalse(tracker.isReferenced(5)); assertEquals(0, tracker.trackedSstableCountForTesting()); } @Test public void testUntrackedTableSSTableHoldsNoRefs() { - SegmentReferenceTracker tracker = newTracker(); - // Local + unrepaired + non-empty coordinatorLogOffsets, but its table has migrated away from tracked. Such - // an sstable (or a compaction output that inherits its offsets) is never promoted to repaired, so it must - // not be counted or its segments would be pinned forever (CASSANDRA-21406). - SSTableReader sstable = sstable(intervals(5, 0, 7, 100), () -> false, coordinatorLogOffsets(true), - LOCAL_HOST, ReplicationType.untracked); + Map> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 20)); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader sstable = sstable(offsets(LOG_ID, 10, 20), () -> false, LOCAL_HOST, ReplicationType.untracked); tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); - for (long segment = 5; segment <= 7; segment++) - assertFalse("segment " + segment, tracker.isReferenced(segment)); + assertFalse(tracker.isReferenced(5)); assertEquals(0, tracker.trackedSstableCountForTesting()); } @@ -137,14 +145,17 @@ public void testUntrackedTableSSTableHoldsNoRefs() public void testEvictReleasesReferencesAndFiresCallback() { int[] calls = { 0 }; - SegmentReferenceTracker tracker = newTracker(() -> calls[0]++); - SSTableReader sstable = unrepaired(intervals(5, 0, 7, 0)); + Map> segments = new HashMap<>(); + for (long s = 5; s <= 7; s++) + segments.put(s, mockSegment(s, LOG_ID, (int) s * 10, (int) s * 10 + 9)); + + SegmentReferenceTracker tracker = newTracker(segments, () -> calls[0]++); + SSTableReader sstable = unrepaired(offsets(LOG_ID, 50, 79)); tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); for (long segment = 5; segment <= 7; segment++) assertEquals(1L, tracker.referenceCountForTesting(segment)); - // Simulates a keyspace migrating away from tracked: evict its sstables' references directly. tracker.evict(List.of(sstable)); for (long segment = 5; segment <= 7; segment++) @@ -156,14 +167,16 @@ public void testEvictReleasesReferencesAndFiresCallback() @Test public void testEvictOnlyReleasesGivenSstables() { - SegmentReferenceTracker tracker = newTracker(); - SSTableReader a = unrepaired(intervals(5, 0, 5, 100)); - SSTableReader b = unrepaired(intervals(5, 0, 5, 200)); + Map> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 20)); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader a = unrepaired(offsets(LOG_ID, 10, 15), "a"); + SSTableReader b = unrepaired(offsets(LOG_ID, 16, 20), "b"); tracker.handleNotification(new SSTableAddedNotification(List.of(a, b), null), null); assertEquals(2L, tracker.referenceCountForTesting(5)); - // Evicting only 'a' leaves segment 5 referenced by 'b'; evicting an untracked sstable is a no-op. tracker.evict(List.of(a)); assertTrue(tracker.isReferenced(5)); assertEquals(1L, tracker.referenceCountForTesting(5)); @@ -176,10 +189,11 @@ public void testEvictOnlyReleasesGivenSstables() @Test public void testAddIsIdempotent() { - int startSegment = 5; - int endSegment = 7; - SegmentReferenceTracker tracker = newTracker(); - SSTableReader sstable = unrepaired(intervals(startSegment, 0, endSegment, 100)); + Map> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 20)); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader sstable = unrepaired(offsets(LOG_ID, 10, 20)); tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); @@ -191,79 +205,64 @@ public void testAddIsIdempotent() @Test public void testMultipleDisjointIntervalsRefEachContainedSegment() { - SegmentReferenceTracker tracker = newTracker(); - // Two disjoint intervals: [3:0..3:100] and [9:0..10:50]. - IntervalSet.Builder builder = new IntervalSet.Builder<>(); - builder.add(new CommitLogPosition(3, 0), new CommitLogPosition(3, 100)); - builder.add(new CommitLogPosition(9, 0), new CommitLogPosition(10, 50)); - SSTableReader sstable = unrepaired(builder.build()); + Map> segments = new HashMap<>(); + segments.put(3L, mockSegment(3L, LOG_ID, 0, 100)); + segments.put(5L, mockSegment(5L, LOG_ID, 200, 300)); + segments.put(9L, mockSegment(9L, LOG_ID, 500, 550)); + segments.put(10L, mockSegment(10L, LOG_ID, 551, 600)); + + SegmentReferenceTracker tracker = newTracker(segments); + ImmutableCoordinatorLogOffsets offsets = new ImmutableCoordinatorLogOffsets.Builder() + .add(LOG_ID, 0, 100) + .add(LOG_ID, 500, 600) + .build(); + SSTableReader sstable = unrepaired(offsets); tracker.handleNotification(new InitialSSTableAddedNotification(List.of(sstable)), null); assertEquals(1L, tracker.referenceCountForTesting(3)); assertEquals(1L, tracker.referenceCountForTesting(9)); assertEquals(1L, tracker.referenceCountForTesting(10)); - // Gap between disjoint intervals is not referenced. - for (long gap = 4; gap <= 8; gap++) - assertFalse("segment " + gap, tracker.isReferenced(gap)); + assertFalse(tracker.isReferenced(5)); } @Test public void testMultipleIntervalsInSameSegmentCountedOnce() { - SegmentReferenceTracker tracker = newTracker(); - // Two disjoint intervals within the SAME segment (different position ranges): forEachSegment emits - // segment 5 twice for this sstable, but the referrer set de-dupes so it is counted exactly once. - IntervalSet.Builder builder = new IntervalSet.Builder<>(); - builder.add(new CommitLogPosition(5, 0), new CommitLogPosition(5, 100)); - builder.add(new CommitLogPosition(5, 200), new CommitLogPosition(5, 300)); - SSTableReader sstable = unrepaired(builder.build()); - - tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); - - assertEquals("a segment spanned by multiple intervals of one sstable is referenced once, not n times", - 1L, tracker.referenceCountForTesting(5)); - assertEquals(1, tracker.trackedSstableCountForTesting()); - - // Releasing clears the single reference (no residual double-count). - tracker.handleNotification( - new SSTableListChangedNotification(List.of(), List.of(sstable), OperationType.COMPACTION), - null); - assertFalse(tracker.isReferenced(5)); - assertEquals(0, tracker.trackedSstableCountForTesting()); - } + Map> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 0, 300)); - @Test - public void testEmptyIntervalsHoldNoRefs() - { - SegmentReferenceTracker tracker = newTracker(); - SSTableReader sstable = unrepaired(IntervalSet.empty()); + SegmentReferenceTracker tracker = newTracker(segments); + ImmutableCoordinatorLogOffsets offsets = new ImmutableCoordinatorLogOffsets.Builder() + .add(LOG_ID, 0, 100) + .add(LOG_ID, 200, 300) + .build(); + SSTableReader sstable = unrepaired(offsets); tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); - // Tracked sstable but no segments to ref. - assertEquals(1, tracker.trackedSstableCountForTesting()); - assertFalse(tracker.isReferenced(0)); + assertEquals(1L, tracker.referenceCountForTesting(5)); + assertEquals(List.of(sstable.getFilename()), tracker.referrerDescriptors(5)); } @Test public void testCompactionPreservesRefsWhenInputAndOutputOverlapSameSegment() { - SegmentReferenceTracker tracker = newTracker(); - SSTableReader input = unrepaired(intervals(5, 0, 5, 100)); - SSTableReader output = unrepaired(intervals(5, 0, 5, 200)); + Map> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 50)); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader input = unrepaired(offsets(LOG_ID, 10, 30), "input"); + SSTableReader output = unrepaired(offsets(LOG_ID, 10, 50), "output"); tracker.handleNotification(new SSTableAddedNotification(List.of(input), null), null); assertEquals(1L, tracker.referenceCountForTesting(5)); - // Compaction emits the SSTableListChangedNotification with added + removed atomically. - tracker.handleNotification( - new SSTableListChangedNotification(List.of(output), - List.of(input), - OperationType.COMPACTION), - null); + tracker.handleNotification(new SSTableListChangedNotification(List.of(output), + List.of(input), + OperationType.COMPACTION), + null); - // Net: still one unrepaired sstable referencing segment 5. assertEquals(1L, tracker.referenceCountForTesting(5)); assertEquals(1, tracker.trackedSstableCountForTesting()); } @@ -271,18 +270,20 @@ public void testCompactionPreservesRefsWhenInputAndOutputOverlapSameSegment() @Test public void testCompactionToRepairedOutputReleasesRefs() { - SegmentReferenceTracker tracker = newTracker(); - SSTableReader input = unrepaired(intervals(5, 0, 5, 100)); - SSTableReader output = repaired(intervals(5, 0, 5, 200)); + Map> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 50)); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader input = unrepaired(offsets(LOG_ID, 10, 30)); + SSTableReader output = repaired(offsets(LOG_ID, 10, 50)); tracker.handleNotification(new SSTableAddedNotification(List.of(input), null), null); assertEquals(1L, tracker.referenceCountForTesting(5)); - tracker.handleNotification( - new SSTableListChangedNotification(List.of(output), - List.of(input), - OperationType.COMPACTION), - null); + tracker.handleNotification(new SSTableListChangedNotification(List.of(output), + List.of(input), + OperationType.COMPACTION), + null); assertFalse(tracker.isReferenced(5)); assertEquals(0, tracker.trackedSstableCountForTesting()); @@ -291,16 +292,53 @@ public void testCompactionToRepairedOutputReleasesRefs() @Test public void testRepairPromotionReleasesRefs() { - SegmentReferenceTracker tracker = newTracker(); + Map> segments = new HashMap<>(); + for (long s = 5; s <= 7; s++) + segments.put(s, mockSegment(s, LOG_ID, (int) s * 10, (int) s * 10 + 9)); + + SegmentReferenceTracker tracker = newTracker(segments); + AtomicReference repaired = new AtomicReference<>(false); + SSTableReader sstable = sstableWithRepairSupplier(offsets(LOG_ID, 50, 79), repaired::get); + + tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); + for (long segment = 5; segment <= 7; segment++) + assertEquals(1L, tracker.referenceCountForTesting(segment)); + + repaired.set(true); + tracker.handleNotification(new SSTableRepairStatusChanged(List.of(sstable)), null); + + for (long segment = 5; segment <= 7; segment++) + assertFalse("segment " + segment, tracker.isReferenced(segment)); + assertEquals(0, tracker.trackedSstableCountForTesting()); + } + + @Test + public void testRepairPromotionWithClearedOffsetsReleasesRefs() + { + Map> segments = new HashMap<>(); + for (long s = 5; s <= 7; s++) + segments.put(s, mockSegment(s, LOG_ID, (int) s * 10, (int) s * 10 + 9)); + + SegmentReferenceTracker tracker = newTracker(segments); AtomicReference repaired = new AtomicReference<>(false); - SSTableReader sstable = sstableWithRepairSupplier(intervals(5, 0, 7, 0), repaired::get); + AtomicReference offsetsRef = new AtomicReference<>(offsets(LOG_ID, 50, 79)); + + SSTableReader sstable = Mockito.mock(SSTableReader.class); + Mockito.when(sstable.isRepaired()).thenAnswer(inv -> repaired.get()); + Mockito.when(sstable.getSSTableMetadata()).thenReturn(stats(LOCAL_HOST)); + Mockito.when(sstable.getCoordinatorLogOffsets()).thenAnswer(inv -> offsetsRef.get()); + TableMetadata tableMetadata = Mockito.mock(TableMetadata.class); + Mockito.when(tableMetadata.replicationType()).thenReturn(ReplicationType.tracked); + Mockito.when(sstable.metadata()).thenReturn(tableMetadata); + Mockito.when(sstable.getFilename()).thenReturn("sstable"); tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); for (long segment = 5; segment <= 7; segment++) assertEquals(1L, tracker.referenceCountForTesting(segment)); - // Promote to repaired; repair-status-changed delivers the transition. + // When promoted, SSTableReader.mutatePromotedToRepairedAndReload marks repaired and clears coordinatorLogOffsets. repaired.set(true); + offsetsRef.set(ImmutableCoordinatorLogOffsets.NONE); tracker.handleNotification(new SSTableRepairStatusChanged(List.of(sstable)), null); for (long segment = 5; segment <= 7; segment++) @@ -311,15 +349,16 @@ public void testRepairPromotionReleasesRefs() @Test public void testRepairStatusFlippingBackToUnrepairedReAcquires() { - SegmentReferenceTracker tracker = newTracker(); + Map> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 20)); + + SegmentReferenceTracker tracker = newTracker(segments); AtomicReference repaired = new AtomicReference<>(true); - SSTableReader sstable = sstableWithRepairSupplier(intervals(5, 0, 5, 100), repaired::get); + SSTableReader sstable = sstableWithRepairSupplier(offsets(LOG_ID, 10, 20), repaired::get); - // Starts repaired -> add is a no-op. tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); assertEquals(0, tracker.trackedSstableCountForTesting()); - // Flip back to unrepaired (e.g. failed repair session) and deliver repair-status-changed. repaired.set(false); tracker.handleNotification(new SSTableRepairStatusChanged(List.of(sstable)), null); @@ -330,9 +369,14 @@ public void testRepairStatusFlippingBackToUnrepairedReAcquires() @Test public void testMultipleSstablesAccumulateRefsOnSharedSegments() { - SegmentReferenceTracker tracker = newTracker(); - SSTableReader a = unrepaired(intervals(5, 0, 6, 0)); - SSTableReader b = unrepaired(intervals(6, 0, 7, 0)); + Map> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 50, 59)); + segments.put(6L, mockSegment(6L, LOG_ID, 60, 69)); + segments.put(7L, mockSegment(7L, LOG_ID, 70, 79)); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader a = unrepaired(offsets(LOG_ID, 50, 65), "a"); + SSTableReader b = unrepaired(offsets(LOG_ID, 65, 79), "b"); tracker.handleNotification(new SSTableAddedNotification(List.of(a, b), null), null); @@ -340,48 +384,53 @@ public void testMultipleSstablesAccumulateRefsOnSharedSegments() assertEquals(2L, tracker.referenceCountForTesting(6)); assertEquals(1L, tracker.referenceCountForTesting(7)); - tracker.handleNotification( - new SSTableListChangedNotification(List.of(), - List.of(a), - OperationType.COMPACTION), - null); + tracker.handleNotification(new SSTableListChangedNotification(List.of(), + List.of(a), + OperationType.COMPACTION), + null); assertFalse(tracker.isReferenced(5)); - assertEquals("there's still a reference from the shared segment from b", 1L, tracker.referenceCountForTesting(6)); + assertEquals("reference from b remains", 1L, tracker.referenceCountForTesting(6)); assertEquals(1L, tracker.referenceCountForTesting(7)); } @Test public void testReleaseOfUntrackedSstableIsNoOp() { - SegmentReferenceTracker tracker = newTracker(); - SSTableReader sstable = unrepaired(intervals(5, 0, 5, 100)); + Map> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 20)); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader sstable = unrepaired(offsets(LOG_ID, 10, 20)); assertFalse(tracker.isReferenced(5)); - tracker.handleNotification( - new SSTableListChangedNotification(List.of(), - List.of(sstable), - OperationType.COMPACTION), - null); + tracker.handleNotification(new SSTableListChangedNotification(List.of(), + List.of(sstable), + OperationType.COMPACTION), + null); - assertFalse("ensure a never added sstable does not underflow during removal", tracker.isReferenced(5)); + assertFalse(tracker.isReferenced(5)); } @Test public void testCallbackFiredWhenLastReferenceReleased() { int[] calls = { 0 }; - SegmentReferenceTracker tracker = newTracker(() -> calls[0]++); - SSTableReader sstable = unrepaired(intervals(5, 0, 7, 0)); + Map> segments = new HashMap<>(); + for (long s = 5; s <= 7; s++) + segments.put(s, mockSegment(s, LOG_ID, (int) s * 10, (int) s * 10 + 9)); + + SegmentReferenceTracker tracker = newTracker(segments, () -> calls[0]++); + SSTableReader sstable = unrepaired(offsets(LOG_ID, 50, 79)); tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); - assertEquals("adding never fires the unreferenced callback", 0, calls[0]); + assertEquals("adding does not fire the unreferenced callback", 0, calls[0]); - // Releasing the only referrer drives segments 5..7 to zero -> callback fires (once per notification). - tracker.handleNotification( - new SSTableListChangedNotification(List.of(), List.of(sstable), OperationType.COMPACTION), - null); + tracker.handleNotification(new SSTableListChangedNotification(List.of(), + List.of(sstable), + OperationType.COMPACTION), + null); assertFalse(tracker.isReferenced(5)); assertEquals("releasing the last reference fires the unreferenced callback", 1, calls[0]); @@ -391,125 +440,264 @@ public void testCallbackFiredWhenLastReferenceReleased() public void testCallbackNotFiredWhileSegmentStillReferenced() { int[] calls = { 0 }; - SegmentReferenceTracker tracker = newTracker(() -> calls[0]++); - SSTableReader a = unrepaired(intervals(5, 0, 5, 100)); - SSTableReader b = unrepaired(intervals(5, 0, 5, 200)); + Map> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 20)); + + SegmentReferenceTracker tracker = newTracker(segments, () -> calls[0]++); + SSTableReader a = unrepaired(offsets(LOG_ID, 10, 15), "a"); + SSTableReader b = unrepaired(offsets(LOG_ID, 16, 20), "b"); tracker.handleNotification(new SSTableAddedNotification(List.of(a, b), null), null); assertEquals(2L, tracker.referenceCountForTesting(5)); - // Releasing one of two referrers leaves segment 5 still referenced -> no callback. - tracker.handleNotification( - new SSTableListChangedNotification(List.of(), List.of(a), OperationType.COMPACTION), - null); + tracker.handleNotification(new SSTableListChangedNotification(List.of(), + List.of(a), + OperationType.COMPACTION), + null); + + assertTrue(tracker.isReferenced(5)); + assertEquals(0, calls[0]); + } + + @Test + public void testSSTableWithNoOverlappingSegmentsIsNotTracked() + { + Map> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 20)); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader sstable = unrepaired(offsets(LOG_ID, 100, 200)); + + tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); + + assertEquals(0, tracker.trackedSstableCountForTesting()); + assertTrue(tracker.trackedSSTables().isEmpty()); + assertFalse(tracker.isReferenced(5)); + + tracker.handleNotification(new SSTableListChangedNotification(List.of(), List.of(sstable), OperationType.COMPACTION), + null); + assertEquals(0, tracker.trackedSstableCountForTesting()); + } + + @Test + public void testSegmentOverloadsAndNullHandling() + { + Map> segments = new HashMap<>(); + Segment segment5 = mockSegment(5L, LOG_ID, 10, 20); + segments.put(5L, segment5); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader sstable = unrepaired(offsets(LOG_ID, 10, 20), "sstable5"); + + assertFalse(tracker.isReferenced(segment5)); + assertFalse(tracker.isReferenced((Segment) null)); + assertEquals(0, tracker.referenceCount(segment5)); + assertEquals(0, tracker.referenceCount((Segment) null)); + assertEquals(List.of(), tracker.referrerDescriptors(segment5)); + assertEquals(List.of(), tracker.referrerDescriptors((Segment) null)); + + tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); + assertTrue(tracker.isReferenced(segment5)); + assertEquals(1, tracker.referenceCount(segment5)); + assertEquals(1L, tracker.referenceCountForTesting(segment5)); + assertEquals(List.of("sstable5"), tracker.referrerDescriptors(segment5)); + } + + @Test + public void testDefensivePruneOfEmptyReferrers() + { + Map> segments = new HashMap<>(); + segments.put(5L, mockSegment(5L, LOG_ID, 10, 20)); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader sstable = unrepaired(offsets(LOG_ID, 10, 20), "sstable5"); + + tracker.addReferenceForTesting(5L, sstable); assertTrue(tracker.isReferenced(5)); - assertEquals("callback must not fire while the segment is still referenced", 0, calls[0]); + assertEquals(1, tracker.referenceCount(5)); + + tracker.evict(List.of(sstable)); + assertFalse(tracker.isReferenced(5)); + assertEquals(0, tracker.referenceCount(5)); + assertTrue(tracker.referrerDescriptors(5).isEmpty()); + } + + @Test + public void testTrackingActiveSegment() + { + MutationJournal.ActiveOffsetRanges activeRanges = new MutationJournal.ActiveOffsetRanges(); + activeRanges.update(new ShortMutationId(LOG_ID, 10)); + activeRanges.update(new ShortMutationId(LOG_ID, 50)); + + @SuppressWarnings("unchecked") + Segment activeSegment = Mockito.mock(Segment.class); + Mockito.when(activeSegment.id()).thenReturn(5L); + Mockito.when(activeSegment.keyStats()).thenReturn(activeRanges); + + Map> segments = new HashMap<>(); + segments.put(5L, activeSegment); + + SegmentReferenceTracker tracker = newTracker(segments); + SSTableReader sstable = unrepaired(offsets(LOG_ID, 20, 30)); + + tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); + + assertTrue(tracker.isReferenced(activeSegment)); + assertEquals(1, tracker.referenceCount(activeSegment)); + assertEquals(List.of(sstable.getFilename()), tracker.referrerDescriptors(activeSegment)); + + tracker.handleNotification(new SSTableListChangedNotification(List.of(), List.of(sstable), OperationType.COMPACTION), + null); + + assertFalse(tracker.isReferenced(activeSegment)); + assertEquals(0, tracker.referenceCount(activeSegment)); + } + + @Test + public void testSSTableWithMultipleLogsRefsAllOverlappingSegments() + { + long log1 = 100L; + long log2 = 200L; + + Map> segments = new HashMap<>(); + segments.put(1L, mockSegment(1L, log1, 10, 20)); + segments.put(2L, mockSegment(2L, log2, 30, 40)); + + SegmentReferenceTracker tracker = newTracker(segments); + ImmutableCoordinatorLogOffsets offsets = new ImmutableCoordinatorLogOffsets.Builder() + .add(log1, 15, 25) + .add(log2, 35, 45) + .build(); + SSTableReader sstable = unrepaired(offsets, "sstable-multi-log"); + + tracker.handleNotification(new SSTableAddedNotification(List.of(sstable), null), null); + + assertTrue(tracker.isReferenced(1L)); + assertTrue(tracker.isReferenced(2L)); + assertEquals(1, tracker.referenceCount(1L)); + assertEquals(1, tracker.referenceCount(2L)); + assertEquals(1, tracker.trackedSstableCountForTesting()); + + tracker.handleNotification(new SSTableListChangedNotification(List.of(), List.of(sstable), OperationType.COMPACTION), + null); + + assertFalse(tracker.isReferenced(1L)); + assertFalse(tracker.isReferenced(2L)); + assertEquals(0, tracker.referenceCount(1L)); + assertEquals(0, tracker.referenceCount(2L)); + assertEquals(0, tracker.trackedSstableCountForTesting()); } // -- helpers --------------------------------------------------------- - private static SegmentReferenceTracker newTracker() + private static SegmentReferenceTracker newTracker(Map> segments) + { + return newTracker(segments, () -> {}); + } + + private static SegmentReferenceTracker newTracker(Map> segments, Runnable onSegmentsUnreferenced) + { + return new SegmentReferenceTracker(onSegmentsUnreferenced, () -> LOCAL_HOST, segments::values); + } + + @SuppressWarnings("unchecked") + private static Segment mockSegment(long segmentId, long logId, int minOffset, int maxOffset) { - return newTracker(() -> {}); + Segment segment = Mockito.mock(Segment.class); + Mockito.when(segment.id()).thenReturn(segmentId); + Mockito.when(segment.keyStats()).thenReturn(MutationJournal.StaticOffsetRanges.of(logId, minOffset, maxOffset)); + return segment; } - private static SegmentReferenceTracker newTracker(Runnable onSegmentsUnreferenced) + private static ImmutableCoordinatorLogOffsets offsets(long logId, int start, int end) { - // Fixed local host id so the mock sstables (originating from LOCAL_HOST) are treated as locally originated. - return new SegmentReferenceTracker(onSegmentsUnreferenced, () -> LOCAL_HOST); + return new ImmutableCoordinatorLogOffsets.Builder().add(logId, start, end).build(); } - private static IntervalSet intervals(long startSegment, int startPosition, long endSegment, int endPosition) + private static SSTableReader unrepaired(ImmutableCoordinatorLogOffsets offsets) { - assertTrue("startSegment " + startSegment + " is less than or equal to endSegment " + endSegment, startSegment <= endSegment); - assertTrue("startPosition " + startPosition + " is less than or equal to endPosition " + endPosition, startPosition <= endPosition); - return new IntervalSet<>(new CommitLogPosition(startSegment, startPosition), - new CommitLogPosition(endSegment, endPosition)); + return stub(offsets, false, "sstable"); } - private static SSTableReader unrepaired(IntervalSet intervals) + private static SSTableReader unrepaired(ImmutableCoordinatorLogOffsets offsets, String name) { - return stub(intervals, false); + return stub(offsets, false, name); } - private static SSTableReader repaired(IntervalSet intervals) + private static SSTableReader repaired(ImmutableCoordinatorLogOffsets offsets) { - return stub(intervals, true); + return stub(offsets, true, "sstable"); } - private static SSTableReader stub(IntervalSet intervals, boolean isRepaired) + private static SSTableReader stub(ImmutableCoordinatorLogOffsets offsets, boolean isRepaired, String name) { - return sstableWithRepairSupplier(intervals, () -> isRepaired); + return sstableWithRepairSupplier(offsets, () -> isRepaired, name); } - private static SSTableReader sstableWithRepairSupplier(IntervalSet intervals, + private static SSTableReader sstableWithRepairSupplier(ImmutableCoordinatorLogOffsets offsets, BooleanSupplier isRepairedSupplier) { - return sstable(intervals, isRepairedSupplier, coordinatorLogOffsets(true)); + return sstableWithRepairSupplier(offsets, isRepairedSupplier, "sstable"); } - private static SSTableReader sstable(IntervalSet intervals, - BooleanSupplier isRepairedSupplier, - ImmutableCoordinatorLogOffsets coordinatorLogOffsets) + private static SSTableReader sstableWithRepairSupplier(ImmutableCoordinatorLogOffsets offsets, + BooleanSupplier isRepairedSupplier, + String name) { - return sstable(intervals, isRepairedSupplier, coordinatorLogOffsets, LOCAL_HOST); + return sstable(offsets, isRepairedSupplier, LOCAL_HOST, ReplicationType.tracked, name); } - private static SSTableReader sstable(IntervalSet intervals, + private static SSTableReader sstable(ImmutableCoordinatorLogOffsets offsets, BooleanSupplier isRepairedSupplier, - ImmutableCoordinatorLogOffsets coordinatorLogOffsets, - UUID originatingHostId) + UUID originatingHostId, + ReplicationType replicationType) { - return sstable(intervals, isRepairedSupplier, coordinatorLogOffsets, originatingHostId, ReplicationType.tracked); + return sstable(offsets, isRepairedSupplier, originatingHostId, replicationType, "sstable"); } - private static SSTableReader sstable(IntervalSet intervals, + private static SSTableReader sstable(ImmutableCoordinatorLogOffsets offsets, BooleanSupplier isRepairedSupplier, - ImmutableCoordinatorLogOffsets coordinatorLogOffsets, UUID originatingHostId, - ReplicationType replicationType) + ReplicationType replicationType, + String name) { SSTableReader reader = Mockito.mock(SSTableReader.class); - Mockito.when(reader.isRepaired()).thenAnswer(ref -> isRepairedSupplier.getAsBoolean()); - Mockito.when(reader.getSSTableMetadata()).thenReturn(stats(intervals, originatingHostId)); - Mockito.when(reader.getCoordinatorLogOffsets()).thenReturn(coordinatorLogOffsets); + Mockito.when(reader.isRepaired()).thenAnswer(inv -> isRepairedSupplier.getAsBoolean()); + Mockito.when(reader.getSSTableMetadata()).thenReturn(stats(originatingHostId)); + Mockito.when(reader.getCoordinatorLogOffsets()).thenReturn(offsets); TableMetadata tableMetadata = Mockito.mock(TableMetadata.class); Mockito.when(tableMetadata.replicationType()).thenReturn(replicationType); Mockito.when(reader.metadata()).thenReturn(tableMetadata); - return reader; - } + Mockito.when(reader.getFilename()).thenReturn(name); - private static ImmutableCoordinatorLogOffsets coordinatorLogOffsets(boolean nonEmpty) - { - // A bare mock reports isEmpty()==false (Mockito's default boolean), which is all the tracker inspects. - return nonEmpty ? Mockito.mock(ImmutableCoordinatorLogOffsets.class) : ImmutableCoordinatorLogOffsets.NONE; + return reader; } - private static StatsMetadata stats(IntervalSet intervals, UUID originatingHostId) + private static StatsMetadata stats(UUID originatingHostId) { - return new StatsMetadata(new EstimatedHistogram(155), // estimatedPartitionSize - new EstimatedHistogram(118), // estimatedCellPerPartitionCount - intervals, // commitLogIntervals - 0L, // minTimestamp - 0L, // maxTimestamp - Cell.NO_DELETION_TIME, // minLocalDeletionTime - Cell.NO_DELETION_TIME, // maxLocalDeletionTime - Cell.NO_TTL, // minTTL - Cell.NO_TTL, // maxTTL - -1.0, // compressionRatio + return new StatsMetadata(new EstimatedHistogram(155), + new EstimatedHistogram(118), + IntervalSet.empty(), + 0L, + 0L, + Cell.NO_DELETION_TIME, + Cell.NO_DELETION_TIME, + Cell.NO_TTL, + Cell.NO_TTL, + -1.0, TombstoneHistogram.createDefault(), - 0, // sstableLevel - List.of(), // clusteringTypes - Slice.ALL, // coveredClustering - false, // hasLegacyCounterShards + 0, + List.of(), + Slice.ALL, + false, ActiveRepairService.UNREPAIRED_SSTABLE, - 0L, // totalColumnsSet - 0L, // totalRows - Double.NaN, // tokenSpaceCoverage + 0L, + 0L, + Double.NaN, originatingHostId, ActiveRepairService.NO_PENDING_REPAIR, - false, // hasPartitionLevelDeletions + false, ImmutableCoordinatorLogOffsets.NONE, ByteBufferUtil.EMPTY_BYTE_BUFFER, ByteBufferUtil.EMPTY_BYTE_BUFFER); From 2f2aa11ba8fa1d06b22f251d27ea5f8c17530fe8 Mon Sep 17 00:00:00 2001 From: Blake Eggleston Date: Wed, 7 Oct 2026 20:50:34 -0700 Subject: [PATCH 09/12] CEP-45: Fix forward target detection in PaxosMutationTrackingForwardingV1Test --- ...PaxosMutationTrackingForwardingV1Test.java | 50 +++++++++---------- 1 file changed, 23 insertions(+), 27 deletions(-) diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/PaxosMutationTrackingForwardingV1Test.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/PaxosMutationTrackingForwardingV1Test.java index f88f07447659..d059aaa9416e 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/tracking/PaxosMutationTrackingForwardingV1Test.java +++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/PaxosMutationTrackingForwardingV1Test.java @@ -215,32 +215,29 @@ public void testV1CommitForwardingRetryAfterCoordinatorBehind() cluster.coordinator(1).execute("INSERT INTO " + ks + ".tbl (k, v) VALUES (" + KEY + ", 1)", ConsistencyLevel.ALL); - // Strategy: Hold PAXOS_COMMIT_REQ at nodes 2 and 3 (the non-forwarding replicas). - // While held, ALTER the keyspace to untracked. When released, nodes 2,3 see the - // commit request from the stale forward handler (node 1) and reject with - // COORDINATOR_BEHIND. This bubbles back to node 4's retry loop. - AssertingLatch commitArrived = new AssertingLatch("PAXOS_COMMIT_REQ at nodes 2,3"); + AtomicInteger forwardTarget = new AtomicInteger(-1); + AssertingLatch commitArrived = new AssertingLatch("PAXOS_COMMIT_REQ from forward target"); AssertingLatch alterDone = new AssertingLatch("V1 commit forwarding retry - alter to untracked"); - - // Count ALL inbound PAXOS_COMMIT_REQ at nodes 2,3 regardless of source. After the retry, - // node 4 (now on untracked) calls commitPaxosUntracked directly and sends to replicas - // {1,2,3}, yielding 2 more messages at {2,3}. Total = initial 2 + retry 2 = 4. - // If the retry didn't happen (regression), total would be 2 and the CAS would fail. AtomicInteger commitsAtReplicas = new AtomicInteger(); - - // Hold PAXOS_COMMIT_REQ at nodes 2 and 3 until ALTER completes. - // Also check retry messages (from node 4) for mutation ID absence. One filter handles - // both roles because the from-node distinguishes initial (node 1) from retry (node 4) - // and two overlapping MessageSpy instances on the same verb+destination would not - // compose cleanly — the cluster filter system applies ALL matching filters. AtomicInteger retryCommitsWithId = new AtomicInteger(); + + // Learn which replica the proximity measures chose for the forward — any of {1,2,3} is possible. + cluster.filters() + .verbs(Verb.PAXOS_COMMIT_FORWARD_REQ.id) + .messagesMatching((from, to, msg) -> { + forwardTarget.set(to); + return false; + }).drop(); + + // Hold PAXOS_COMMIT_REQ from the forward target until ALTER completes. Count ALL inbound + // PAXOS_COMMIT_REQ at replicas to verify the retry fires, and check retry messages (from + // node 4) for mutation ID absence. Conditional behaviour inside one filter — stays manual. cluster.filters() .verbs(Verb.PAXOS_COMMIT_REQ.id) - .to(2, 3) + .to(1, 2, 3) .messagesMatching((from, to, msg) -> { commitsAtReplicas.incrementAndGet(); - // Only hold commits from node 1 (the forward handler acting as coordinator) - if (from == 1) + if (from == forwardTarget.get()) { commitArrived.countDown(); alterDone.await(); @@ -269,8 +266,8 @@ else if (from == 4) } finally { - // Release the held commits — nodes 2,3 now see migration mismatch -> COORDINATOR_BEHIND - alterDone.countDown(); + alterDone.release(); + commitArrived.release(); } // With the fix: node 4's retry loop catches CoordinatorBehindException, retries @@ -291,12 +288,11 @@ else if (from == 4) // (node 1 gets it via the tracked local write in the initial forward, nodes 2,3 via retry). assertReplicasHaveValue(cluster, ks, KEY, 100, 1, 2, 3); - // Authenticate that the retry actually happened by counting inbound PAXOS_COMMIT_REQ at - // the two non-forwarding replicas. Initial attempt (from node 1, held then released) = 2, - // retry (from node 4 directly on untracked path) = 2. Total = 4. A result of 2 would mean - // the CAS succeeded without retry (bug). - assertEquals("Expected initial (2) + retry (2) = 4 PAXOS_COMMIT_REQ at nodes 2,3", - 4, commitsAtReplicas.get()); + // Retry verification: forward target sends to 2 other replicas (initial attempt), node 4 + // retries directly on the untracked path sending to all 3 replicas. Total = 5. + assertTrue("Forward target should have been identified", forwardTarget.get() > 0); + assertEquals("Expected initial (2) + retry (3) = 5 PAXOS_COMMIT_REQ at replicas", + 5, commitsAtReplicas.get()); // Retry messages from node 4 (untracked path) must not carry mutation IDs assertEquals("Retry PAXOS_COMMIT_REQ from node 4 should NOT carry mutation IDs (untracked path)", From f87548e03cc26bb76b35c25b4c8c6a6bdf272627 Mon Sep 17 00:00:00 2001 From: Blake Eggleston Date: Wed, 7 Oct 2026 22:29:31 -0700 Subject: [PATCH 10/12] Use incremental repair for migration in TrackedRepairFailureTest --- .../distributed/test/tracking/TrackedRepairFailureTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedRepairFailureTest.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedRepairFailureTest.java index 8a49fcb4dedc..dca7695e61fc 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedRepairFailureTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedRepairFailureTest.java @@ -340,7 +340,7 @@ public void testUntrackedTransferRejectionReportsAMigrationThatCompletedDuringTh IInvokableInstance coordinator = cluster.get(1); IInvokableInstance receiver = cluster.get(2); - coordinator.nodetoolResult("repair", "--full", keyspace).asserts().success(); + coordinator.nodetoolResult("repair", keyspace).asserts().success(); coordinator.executeInternal("INSERT INTO " + tableWithKeyspace(keyspace) + " (pk, v) VALUES (?, 1)", KEY_100); coordinator.flush(keyspace); From 344ffaca15723582456a383faae6f1b1825e9522 Mon Sep 17 00:00:00 2001 From: Blake Eggleston Date: Thu, 8 Oct 2026 16:28:03 -0700 Subject: [PATCH 11/12] review feedback --- .../apache/cassandra/replication/CoordinatorLogOffsets.java | 5 +---- .../replication/ImmutableCoordinatorLogOffsets.java | 2 +- .../org/apache/cassandra/replication/MutationJournal.java | 2 +- .../distributed/test/tracking/MutationTrackingUtils.java | 1 - 4 files changed, 3 insertions(+), 7 deletions(-) diff --git a/src/java/org/apache/cassandra/replication/CoordinatorLogOffsets.java b/src/java/org/apache/cassandra/replication/CoordinatorLogOffsets.java index b6ae38171de4..cb78181abbaa 100644 --- a/src/java/org/apache/cassandra/replication/CoordinatorLogOffsets.java +++ b/src/java/org/apache/cassandra/replication/CoordinatorLogOffsets.java @@ -51,10 +51,7 @@ default boolean isEmpty() Mutations mutations(); - default boolean isEmpty() - { - return mutations().isEmpty() && transfers().isEmpty(); - } + boolean isEmpty(); default ActivatedTransfers transfers() { diff --git a/src/java/org/apache/cassandra/replication/ImmutableCoordinatorLogOffsets.java b/src/java/org/apache/cassandra/replication/ImmutableCoordinatorLogOffsets.java index 15a052794f48..df851339b03e 100644 --- a/src/java/org/apache/cassandra/replication/ImmutableCoordinatorLogOffsets.java +++ b/src/java/org/apache/cassandra/replication/ImmutableCoordinatorLogOffsets.java @@ -162,7 +162,7 @@ public Builder add(MutationId mutationId) } @VisibleForTesting - public Builder add(long logId, int start, int end) + Builder add(long logId, int start, int end) { ids.computeIfAbsent(logId, k -> new Offsets.Immutable.Builder(new CoordinatorLogId(k))) .add(start, end); diff --git a/src/java/org/apache/cassandra/replication/MutationJournal.java b/src/java/org/apache/cassandra/replication/MutationJournal.java index 759cbdc99385..045296d76048 100644 --- a/src/java/org/apache/cassandra/replication/MutationJournal.java +++ b/src/java/org/apache/cassandra/replication/MutationJournal.java @@ -1033,7 +1033,7 @@ public ActiveOffsetRanges rebuild(Descriptor descriptor, KeySupport Date: Thu, 8 Oct 2026 16:57:03 -0700 Subject: [PATCH 12/12] review feedback - CEP-45: Track journal segment references via coordinator log offset overlap --- .../replication/MutationJournal.java | 54 ++++++++----------- 1 file changed, 21 insertions(+), 33 deletions(-) diff --git a/src/java/org/apache/cassandra/replication/MutationJournal.java b/src/java/org/apache/cassandra/replication/MutationJournal.java index 045296d76048..ca64cdcfc149 100644 --- a/src/java/org/apache/cassandra/replication/MutationJournal.java +++ b/src/java/org/apache/cassandra/replication/MutationJournal.java @@ -747,23 +747,21 @@ protected static long range(int minOffset, int maxOffset) abstract Map asMap(); - @FunctionalInterface - interface RangePredicate - { - boolean test(long logId, int minOffset, int maxOffset); - } - - abstract boolean anyMatch(RangePredicate predicate); + abstract boolean overlapsLog(long logId, Offsets offsets); public boolean overlaps(CoordinatorLogOffsets offsets) { if (offsets == null || offsets.isEmpty()) return false; - return anyMatch((logId, minOffset, maxOffset) -> { - Offsets sstOffsets = offsets.mutations().getOffsets(logId); - return sstOffsets != null && !sstOffsets.isEmpty() && sstOffsets.overlaps(minOffset, maxOffset); - }); + CoordinatorLogOffsets.Mutations mutations = offsets.mutations(); + for (long logId : mutations) + { + Offsets sstOffsets = mutations.getOffsets(logId); + if (sstOffsets != null && !sstOffsets.isEmpty() && overlapsLog(logId, sstOffsets)) + return true; + } + return false; } @Override @@ -807,19 +805,14 @@ protected Map asMap() } @Override - boolean anyMatch(RangePredicate predicate) + boolean overlapsLog(long logId, Offsets offsets) { - for (Map.Entry entry : ranges.entrySet()) - { - Long val = entry.getValue(); - if (val != null) - { - long range = val.longValue(); - if (predicate.test(entry.getKey().longValue(), minOffset(range), maxOffset(range))) - return true; - } - } - return false; + Long val = ranges.get(logId); + if (val == null) + return false; + + long range = val; + return offsets.overlaps(minOffset(range), maxOffset(range)); } @Override @@ -915,17 +908,12 @@ public boolean mayContain(ShortMutationId id) } @Override - boolean anyMatch(RangePredicate predicate) + boolean overlapsLog(long logId, Offsets offsets) { - for (Long2LongHashMap.EntryIterator iter = ranges.entrySet().iterator(); iter.hasNext();) - { - iter.next(); - long logId = iter.getLongKey(); - long range = iter.getLongValue(); - if (predicate.test(logId, minOffset(range), maxOffset(range))) - return true; - } - return false; + long range = ranges.get(logId); + if (range == NO_VALUE) + return false; + return offsets.overlaps(minOffset(range), maxOffset(range)); } @VisibleForTesting