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..cb78181abbaa 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,8 @@ default boolean isEmpty() Mutations mutations(); + boolean 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..df851339b03e 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 + 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 22a66f61c685..ca64cdcfc149 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; @@ -76,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; @@ -203,7 +205,9 @@ protected void closeActiveSegmentAndOpenAsStatic(ActiveSegment journal.compactor().triggerNow()); + segmentReferenceTracker = new SegmentReferenceTracker(() -> journal.compactor().triggerNow(), + StorageService.instance::getLocalHostUUID, + journal::getAllSegments); } /** @@ -427,6 +431,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) { @@ -724,6 +747,23 @@ protected static long range(int minOffset, int maxOffset) abstract Map asMap(); + abstract boolean overlapsLog(long logId, Offsets offsets); + + public boolean overlaps(CoordinatorLogOffsets offsets) + { + if (offsets == null || offsets.isEmpty()) + return false; + + 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 public String toString() { @@ -764,6 +804,17 @@ protected Map asMap() return ranges; } + @Override + boolean overlapsLog(long logId, Offsets offsets) + { + Long val = ranges.get(logId); + if (val == null) + return false; + + long range = val; + return offsets.overlaps(minOffset(range), maxOffset(range)); + } + @Override public boolean mayContain(ShortMutationId id) { @@ -856,6 +907,32 @@ public boolean mayContain(ShortMutationId id) return range != NO_VALUE && mayContain(range, id); } + @Override + boolean overlapsLog(long logId, Offsets offsets) + { + long range = ranges.get(logId); + if (range == NO_VALUE) + return false; + return offsets.overlaps(minOffset(range), maxOffset(range)); + } + + @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(); @@ -943,6 +1020,15 @@ public ActiveOffsetRanges rebuild(Descriptor descriptor, KeySupport> 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/src/java/org/apache/cassandra/replication/UnreconciledMutations.java b/src/java/org/apache/cassandra/replication/UnreconciledMutations.java index b78e84d25624..81778518bdb1 100644 --- a/src/java/org/apache/cassandra/replication/UnreconciledMutations.java +++ b/src/java/org/apache/cassandra/replication/UnreconciledMutations.java @@ -36,6 +36,7 @@ import org.apache.cassandra.dht.AbstractBounds; import org.apache.cassandra.dht.Bounds; import org.apache.cassandra.dht.Token; +import org.apache.cassandra.exceptions.UnknownTableException; import org.apache.cassandra.schema.TableId; /** @@ -268,7 +269,22 @@ static UnreconciledMutations loadFromJournal(Node2OffsetsMap witnessedOffsets, i for (int offset = iter.start(), end = iter.end(); offset <= end; offset++) { ShortMutationId id = new ShortMutationId(witnessed.logId, offset); - Mutation mutation = MutationJournal.instance().read(id); + Mutation mutation; + try + { + mutation = MutationJournal.instance().read(id); + } + catch (RuntimeException e) + { + if (e.getCause() instanceof UnknownTableException) + { + UnknownTableException ute = (UnknownTableException) e.getCause(); + logger.warn("Skipping loading mutation {} from unknown (probably removed) table with id {}", id, ute.id); + continue; + } + throw e; + } + if (mutation != null) { result.addDirectly(mutation); 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(); } 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/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(), 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/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/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)", 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); 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); 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); 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); diff --git a/test/unit/org/apache/cassandra/replication/CoordinatorLogTest.java b/test/unit/org/apache/cassandra/replication/CoordinatorLogTest.java index 374afab28977..c54e51efdb7a 100644 --- a/test/unit/org/apache/cassandra/replication/CoordinatorLogTest.java +++ b/test/unit/org/apache/cassandra/replication/CoordinatorLogTest.java @@ -21,6 +21,8 @@ import java.util.ArrayList; import java.util.List; +import org.junit.AfterClass; +import org.junit.Before; import org.junit.BeforeClass; import org.junit.Test; @@ -71,6 +73,18 @@ public static void setUp() throws IOException MutationJournal.start(); } + @Before + public void beforeTest() + { + MutationJournal.instance().truncateForTesting(); + } + + @AfterClass + public static void tearDown() + { + MutationJournal.instance().truncateForTesting(); + } + private static Token tk(String key) { return new ByteOrderedPartitioner.BytesToken(ByteBufferUtil.bytes(key)); 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); diff --git a/test/unit/org/apache/cassandra/replication/UnreconciledMutationsTest.java b/test/unit/org/apache/cassandra/replication/UnreconciledMutationsTest.java index d283eadede01..97155161b23f 100644 --- a/test/unit/org/apache/cassandra/replication/UnreconciledMutationsTest.java +++ b/test/unit/org/apache/cassandra/replication/UnreconciledMutationsTest.java @@ -22,6 +22,7 @@ import java.util.List; import org.junit.Assert; +import org.junit.AfterClass; import org.junit.BeforeClass; import org.junit.Test; @@ -41,9 +42,11 @@ import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.TableId; import org.apache.cassandra.schema.TableMetadata; +import org.apache.cassandra.schema.SchemaTestUtil; import org.apache.cassandra.utils.ByteBufferUtil; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; public class UnreconciledMutationsTest { @@ -63,6 +66,13 @@ public static void setUp() throws IOException TABLE_ID = Schema.instance.getTableMetadata(KEYSPACE, TABLE).id; } + @AfterClass + public static void tearDown() + { + if (MutationJournal.instance() != null) + MutationJournal.instance().truncateForTesting(); + } + private static Token tokenFor(int key) { return new ByteOrderedPartitioner.BytesToken(ByteBufferUtil.bytes(key)); @@ -447,4 +457,86 @@ public void testLoadFromJournal() unreconciled.collect(mutation7.key().getToken(), TABLE_ID, false, loadedOffsets); assertEquals(List.of(6, 7), loadedOffsets.asList()); } + + @Test + public void testLoadFromJournalWithDroppedTable() + { + MutationJournal.start(); + MutationJournal.instance().truncateForTesting(); + + String dropKs = "drop_ks"; + String dropTbl = "drop_tbl"; + SchemaLoader.createKeyspace(dropKs, KeyspaceParams.simple(1), + TableMetadata.builder(dropKs, dropTbl) + .addPartitionKeyColumn("k", Int32Type.instance) + .addRegularColumn("v", Int32Type.instance) + .build()); + TableMetadata dropTableMetadata = Schema.instance.getTableMetadata(dropKs, dropTbl); + + CoordinatorLogId logId = new CoordinatorLogId(10, 1); + + Offsets.Mutable offsets1 = new Offsets.Mutable(logId); + offsets1.add(1, 2); + + Offsets.Mutable offsets2 = new Offsets.Mutable(logId); + + Node2OffsetsMap witnessed = new Node2OffsetsMap(); + witnessed.set(1, offsets1); + witnessed.set(2, offsets2); + + MutationId id1 = new MutationId(logId.asLong(), MutationId.sequenceId(1, 0)); + Mutation mutation1 = new RowUpdateBuilder(dropTableMetadata, 0, 1) + .add("v", 1) + .build() + .withMutationId(id1); + + TableMetadata tableMetadata = Schema.instance.getTableMetadata(KEYSPACE, TABLE); + MutationId id2 = new MutationId(logId.asLong(), MutationId.sequenceId(2, 0)); + Mutation mutation2 = new RowUpdateBuilder(tableMetadata, 0, 2) + .add("v", 2) + .build() + .withMutationId(id2); + + MutationJournal.instance().write(mutation1.id(), mutation1); + MutationJournal.instance().write(mutation2.id(), mutation2); + + SchemaTestUtil.announceTableDrop(dropKs, dropTbl); + + UnreconciledMutations unreconciled = UnreconciledMutations.loadFromJournal(witnessed, 1); + Offsets.Mutable loadedOffsets = new Offsets.Mutable(logId); + unreconciled.collect(mutation2.key().getToken(), TABLE_ID, false, loadedOffsets); + assertEquals(List.of(2, 2), loadedOffsets.asList()); + assertTrue(loadedOffsets.contains(2)); + assertEquals(1, loadedOffsets.offsetCount()); + } + + @Test + public void testReplayStaticSegmentsWithDroppedTable() + { + MutationJournal.start(); + MutationJournal.instance().truncateForTesting(); + + String dropKs = "replay_drop_ks"; + String dropTbl = "replay_drop_tbl"; + SchemaLoader.createKeyspace(dropKs, KeyspaceParams.simple(1), + TableMetadata.builder(dropKs, dropTbl) + .addPartitionKeyColumn("k", Int32Type.instance) + .addRegularColumn("v", Int32Type.instance) + .build()); + TableMetadata dropTableMetadata = Schema.instance.getTableMetadata(dropKs, dropTbl); + + CoordinatorLogId logId = new CoordinatorLogId(20, 1); + MutationId id1 = new MutationId(logId.asLong(), MutationId.sequenceId(1, 0)); + Mutation mutation1 = new RowUpdateBuilder(dropTableMetadata, 0, 1) + .add("v", 1) + .build() + .withMutationId(id1); + + MutationJournal.instance().write(mutation1.id(), mutation1); + MutationJournal.instance().closeCurrentSegmentForTestingIfNonEmpty(); + + SchemaTestUtil.announceTableDrop(dropKs, dropTbl); + + MutationJournal.instance().replayStaticSegments(); + } } \ No newline at end of file