This is an automated email from the ASF dual-hosted git repository.
frankgh pushed a commit to branch cep-45-mutation-tracking
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/cep-45-mutation-tracking by
this push:
new 6459184151 CEP-45: Prevent dropping journal segments referenced by
partially reconciled sstables
6459184151 is described below
commit 6459184151a515e43a42a91830591c2112b699f3
Author: Francisco Guerrero <[email protected]>
AuthorDate: Thu Jun 25 14:23:39 2026 -0500
CEP-45: Prevent dropping journal segments referenced by partially
reconciled sstables
Prevent dropping of any segment we might still need in
the future to rebuild a partially reconciled sstable,
even if every mutation in that segment has been fully
reconciled.
patch by Francisco Guerrero; reviewed by Aleksey Yeshchenko for
CASSANDRA-21406
---
CHANGES.txt | 1 +
.../cassandra/config/MutationTrackingSpec.java | 20 +
.../org/apache/cassandra/db/ColumnFamilyStore.java | 3 +
src/java/org/apache/cassandra/db/Keyspace.java | 68 +--
.../db/tracked/TrackedKeyspaceWriteHandler.java | 5 +-
.../db/virtual/MutationTrackingTables.java | 16 +-
.../org/apache/cassandra/journal/Compactor.java | 149 ++++--
src/java/org/apache/cassandra/journal/Journal.java | 64 ++-
.../apache/cassandra/journal/SegmentCompactor.java | 31 ++
.../org/apache/cassandra/journal/Segments.java | 24 +-
.../cassandra/replication/CoordinatorLog.java | 10 +-
.../cassandra/replication/MutationJournal.java | 135 ++++--
.../MutationJournalSegmentCompactor.java | 233 ++++++++++
.../replication/MutationTrackingService.java | 91 ++--
.../replication/SegmentReferenceTracker.java | 385 +++++++++++++++
.../org/apache/cassandra/replication/Shard.java | 3 +-
.../cassandra/service/accord/AccordJournal.java | 3 +-
.../MutationJournalSegmentRefcountTest.java | 318 +++++++++++++
.../tracking/MutationTrackingLogPersisterTest.java | 15 +-
.../db/virtual/MutationJournalTableTest.java | 6 +-
.../cassandra/replication/MutationJournalTest.java | 135 ++++--
.../replication/SegmentReferenceTrackerTest.java | 517 +++++++++++++++++++++
22 files changed, 2031 insertions(+), 201 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index c79abe7f88..56b59b9a23 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
cep-45-mutation-tracking
+ * CEP-45: Prevent dropping journal segments referenced by partially
reconciled sstables (CASSANDRA-21406)
* CEP-45: Fixes and improvements to mutation tracking migration
* Add support for counters to mutation tracking (CASSANDRA-20959)
* CEP-45: Query forwarding (CASSANDRA-20309)
diff --git a/src/java/org/apache/cassandra/config/MutationTrackingSpec.java
b/src/java/org/apache/cassandra/config/MutationTrackingSpec.java
index 30f8acaef5..67def080d8 100644
--- a/src/java/org/apache/cassandra/config/MutationTrackingSpec.java
+++ b/src/java/org/apache/cassandra/config/MutationTrackingSpec.java
@@ -30,4 +30,24 @@ public class MutationTrackingSpec
* The interval in which the backgroun reconciliation process runs
*/
public volatile DurationSpec.LongMillisecondsBound
background_reconciliation_interval = new
DurationSpec.LongMillisecondsBound("1s");
+ /**
+ * Whether the mutation journal's segment-dropping/promotion compactor is
enabled
+ */
+ public volatile boolean journal_compaction_enabled = true;
+ /**
+ * When the on-disk mutation journal grows beyond this size, out-of-band
promotion of already
+ * durably-reconciled but still-unrepaired sstables to repaired is
triggered so the journal segments they hold
+ * can be released and reclaimed (CASSANDRA-21406). {@code null} disables
the mechanism. The default is 256MiB.
+ */
+ public volatile DataStorageSpec.LongBytesBound journal_promotion_threshold
= new DataStorageSpec.LongBytesBound("256MiB");
+
+ /**
+ * @return the on-disk mutation journal size (in bytes) beyond which
out-of-band promotion of durably-reconciled
+ * unrepaired sstables is triggered, or 0 if the mechanism is disabled.
+ */
+ public long getJournalPromotionThresholdBytes()
+ {
+ DataStorageSpec.LongBytesBound threshold = journal_promotion_threshold;
+ return threshold == null ? 0 : threshold.toBytes();
+ }
}
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index 1626867245..288b419cd7 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -538,6 +538,9 @@ public class ColumnFamilyStore implements
ColumnFamilyStoreMBean, Memtable.Owner
data.subscribe(StorageService.instance.sstablesTracker);
data.subscribe(SnapshotManager.instance);
+ if (DatabaseDescriptor.getMutationTrackingEnabled())
+
data.subscribe(MutationJournal.instance().segmentReferenceTracker());
+
Collection<SSTableReader> sstables = null;
// scan for sstables corresponding to this cf and load them
if (data.loadsstables)
diff --git a/src/java/org/apache/cassandra/db/Keyspace.java
b/src/java/org/apache/cassandra/db/Keyspace.java
index 27072f60a6..0020306d79 100644
--- a/src/java/org/apache/cassandra/db/Keyspace.java
+++ b/src/java/org/apache/cassandra/db/Keyspace.java
@@ -57,7 +57,6 @@ import org.apache.cassandra.index.SecondaryIndexManager;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.locator.AbstractReplicationStrategy;
-import org.apache.cassandra.locator.RangesAtEndpoint;
import org.apache.cassandra.metrics.KeyspaceMetrics;
import org.apache.cassandra.repair.KeyspaceRepairManager;
import org.apache.cassandra.replication.MutationTrackingService;
@@ -659,33 +658,19 @@ public class Keyspace
continue;
}
- // If this range is only witnessed then don't apply the
update to the underlying column family store
- // We still want the mutation tracking log to see the
update so that it can witness it
- // and participate in reconciliation of the mutation
- AbstractReplicationStrategy replicationStrategy =
cfs.keyspace.getReplicationStrategy();
- if (replicationStrategy.hasTransientReplicas())
+ // If this range is only witnessed then don't apply the
update to the underlying column family
+ // store. We still want the mutation tracking log to
witness it (via startWriting above) so it can
+ // participate in reconciliation of the mutation. (The
journal write in beginWrite already skipped
+ // marking the segment dirty for witnessed-only mutations;
see MutationJournal.write.)
+ if
(!isFullReplicaFor(cfs.keyspace.getReplicationStrategy(),
upd.partitionKey().getToken(), cm))
{
- RangesAtEndpoint localRanges =
replicationStrategy.getLocalRanges(cm);
- Token token = upd.partitionKey().getToken();
- boolean foundMatch = false;
- for (Range<Token> r : localRanges.onlyFull().ranges())
- {
- if (r.contains(token))
- {
- foundMatch = true;
- break;
- }
- }
- if (!foundMatch)
- {
- // TODO
checkState(!mutation.allowsPotentialTxnConflicts)
- // Basically if a transaction system thinks it is
writing to a non-witness but is writing to a
- // witness then we are probably going to have
issues.
- // This is problematic/racy in general because
schema changes aren't really synced with
- // transaction systems yet when in reality they
really should be completed by a transaction or some
- // other similar solution.
- continue;
- }
+ // TODO
checkState(!mutation.allowsPotentialTxnConflicts)
+ // Basically if a transaction system thinks it is
writing to a non-witness but is writing to a
+ // witness then we are probably going to have issues.
+ // This is problematic/racy in general because schema
changes aren't really synced with
+ // transaction systems yet when in reality they really
should be completed by a transaction or
+ // some other similar solution.
+ continue;
}
cfs.getWriteHandler().write(mutation.id(), upd, ctx, true);
@@ -710,6 +695,35 @@ public class Keyspace
return getMetadata().replicationStrategy;
}
+ /**
+ * @param token the token to check for the given keyspace
+ * @param cm the cluster metadata instance
+ * @return whether this node is a full replica for the given token in this
keyspace. Keyspaces without
+ * transient replicas are always full replicas for tokens they own.
+ */
+ public boolean isFullReplicaFor(Token token, ClusterMetadata cm)
+ {
+ return isFullReplicaFor(getReplicationStrategy(), token, cm);
+ }
+
+ /**
+ * @param replicationStrategy the replication strategy for the keyspace
+ * @param token the token to check for the given keyspace
+ * @param cm the cluster metadata instance
+ * @return whether this node is a full replica for the given token.
Keyspaces without
+ * transient replicas are always full replicas for tokens they own.
+ */
+ static boolean isFullReplicaFor(AbstractReplicationStrategy
replicationStrategy, Token token, ClusterMetadata cm)
+ {
+ if (!replicationStrategy.hasTransientReplicas())
+ return true;
+
+ for (Range<Token> range :
replicationStrategy.getLocalRanges(cm).onlyFull().ranges())
+ if (range.contains(token))
+ return true;
+ return false;
+ }
+
public List<Future<?>> flush(ColumnFamilyStore.FlushReason reason)
{
List<Future<?>> futures = new ArrayList<>(columnFamilyStores.size());
diff --git
a/src/java/org/apache/cassandra/db/tracked/TrackedKeyspaceWriteHandler.java
b/src/java/org/apache/cassandra/db/tracked/TrackedKeyspaceWriteHandler.java
index 2be356538c..783b12d3fd 100644
--- a/src/java/org/apache/cassandra/db/tracked/TrackedKeyspaceWriteHandler.java
+++ b/src/java/org/apache/cassandra/db/tracked/TrackedKeyspaceWriteHandler.java
@@ -26,6 +26,7 @@ import org.apache.cassandra.db.commitlog.CommitLogPosition;
import org.apache.cassandra.exceptions.RequestExecutionException;
import org.apache.cassandra.replication.MutationJournal;
import org.apache.cassandra.service.replication.migration.MigrationRouter;
+import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.concurrent.OpOrder;
@@ -45,7 +46,9 @@ public class TrackedKeyspaceWriteHandler implements
KeyspaceWriteHandler
if (makeDurable)
{
Tracing.trace("Appending to mutation journal");
- pointer = MutationJournal.instance().write(mutation.id(),
mutation);
+ Keyspace keyspace = Keyspace.open(mutation.getKeyspaceName());
+ boolean isFullReplica =
keyspace.isFullReplicaFor(mutation.key().getToken(), ClusterMetadata.current());
+ pointer = MutationJournal.instance().write(mutation.id(),
mutation, isFullReplica);
}
return new CassandraWriteContext(group, pointer);
diff --git
a/src/java/org/apache/cassandra/db/virtual/MutationTrackingTables.java
b/src/java/org/apache/cassandra/db/virtual/MutationTrackingTables.java
index e0498028ee..4c3b801422 100644
--- a/src/java/org/apache/cassandra/db/virtual/MutationTrackingTables.java
+++ b/src/java/org/apache/cassandra/db/virtual/MutationTrackingTables.java
@@ -43,6 +43,7 @@ import org.apache.cassandra.replication.CoordinatorLog;
import org.apache.cassandra.replication.CoordinatorLogId;
import org.apache.cassandra.replication.MutationJournal;
import org.apache.cassandra.replication.MutationTrackingService;
+import org.apache.cassandra.replication.SegmentReferenceTracker;
import org.apache.cassandra.replication.Shard;
import org.apache.cassandra.replication.ShortMutationId;
import org.apache.cassandra.schema.TableId;
@@ -78,7 +79,8 @@ public class MutationTrackingTables
private static final String FSYNCED_TO = "fsynced_to";
private static final String NEEDS_REPLAY = "needs_replay";
private static final String FILE_PATH = "file_path";
-
+ private static final String REFERRING_SSTABLES = "referring_sstables";
+
MutationJournalTable(String keyspace)
{
super(TableMetadata.builder(keyspace, MUTATION_JOURNAL)
@@ -93,16 +95,19 @@ public class MutationTrackingTables
.addRegularColumn(FSYNCED_TO,
Int32Type.instance)
.addRegularColumn(NEEDS_REPLAY,
BooleanType.instance)
.addRegularColumn(FILE_PATH, UTF8Type.instance)
+ .addRegularColumn(REFERRING_SSTABLES,
ListType.getInstance(UTF8Type.instance, false))
.build());
}
-
+
@Override
public DataSet data()
{
SimpleDataSet result = new SimpleDataSet(metadata());
-
+ SegmentReferenceTracker referenceTracker =
MutationJournal.instance().segmentReferenceTracker();
+
for (Segment<ShortMutationId, Mutation> segment :
MutationJournal.instance().getAllSegments())
{
+ List<String> referringSstables =
referenceTracker.referrerDescriptors(segment.id());
result.row(segment.id())
.column(IS_ACTIVE, segment instanceof ActiveSegment)
.column(BYTES_ON_DISK, segment.segmentSizeOnDisk())
@@ -110,9 +115,10 @@ public class MutationTrackingTables
.column(WRITTEN_TO, segment.writtenTo())
.column(FSYNCED_TO, segment.fsyncedTo())
.column(NEEDS_REPLAY, segment.metadata().needsReplay())
- .column(FILE_PATH, segment.filePath());
+ .column(FILE_PATH, segment.filePath())
+ .column(REFERRING_SSTABLES, referringSstables);
}
-
+
return result;
}
}
diff --git a/src/java/org/apache/cassandra/journal/Compactor.java
b/src/java/org/apache/cassandra/journal/Compactor.java
index b6749bc390..80e5cd98da 100644
--- a/src/java/org/apache/cassandra/journal/Compactor.java
+++ b/src/java/org/apache/cassandra/journal/Compactor.java
@@ -21,14 +21,20 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.google.common.base.Throwables;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.ScheduledExecutorPlus;
import org.apache.cassandra.concurrent.Shutdownable;
+import org.apache.cassandra.utils.NoSpamLogger;
import org.apache.cassandra.utils.concurrent.WaitQueue;
import static
org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
@@ -36,11 +42,13 @@ import static
org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFac
public final class Compactor<K, V> implements Runnable, Shutdownable
{
private static final Logger logger =
LoggerFactory.getLogger(Compactor.class);
+ private static final NoSpamLogger noSpamLogger =
NoSpamLogger.getLogger(logger, 5L, TimeUnit.MINUTES);
private final Journal<K, V> journal;
private final SegmentCompactor<K, V> segmentCompactor;
private final ScheduledExecutorPlus executor;
private Future<?> scheduled;
+ private final AtomicBoolean triggerPending = new AtomicBoolean();
public final WaitQueue compacted = WaitQueue.newWaitQueue();
Compactor(Journal<K, V> journal, SegmentCompactor<K, V> segmentCompactor)
@@ -61,51 +69,131 @@ public final class Compactor<K, V> implements Runnable,
Shutdownable
scheduled = executor.scheduleWithFixedDelay(this, period, period,
units);
}
- public synchronized void updateCompactionPeriod(int period, TimeUnit units)
+ public synchronized void updateCompactionPeriod(long period, TimeUnit
units)
{
- if (!journal.params.enableCompaction())
- return;
-
- if (scheduled != null)
- scheduled.cancel(false);
+ cancelPeriodic();
- schedule(period, units);
+ if (journal.params.enableCompaction() && !executor.isShutdown())
+ schedule(period, units);
}
@Override
public void run()
{
- List<StaticSegment<K, V>> toCompact = new ArrayList<>();
- journal.segments().selectStatic(toCompact);
- if (toCompact.isEmpty())
- return;
+ runInternal(true);
+ }
- int limit = journal.params.compactMaxSegments();
- if (toCompact.size() > limit)
+ private void runInternal(boolean runPostCompaction)
+ {
+ triggerPending.set(false);
+ try
{
- toCompact.sort(StaticSegment::compareTo);
- toCompact.subList(limit, toCompact.size()).clear();
+ List<StaticSegment<K, V>> candidates = new ArrayList<>();
+ journal.segments().selectStatic(candidates);
+ if (candidates.isEmpty())
+ return;
+
+ List<StaticSegment<K, V>> toCompact =
maybeWrapArrayList(segmentCompactor.select(candidates));
+ if (toCompact.isEmpty())
+ return;
+
+ int limit = journal.params.compactMaxSegments();
+ if (limit < 0)
+ {
+ noSpamLogger.warn("Misconfigured Journal's compaction max
segments (\"{}\") for journal {}. " +
+ "Compacting all segments ({})",
+ limit, journal.name, toCompact.size());
+ }
+ else if (toCompact.size() > limit)
+ {
+ toCompact.sort(StaticSegment::compareTo);
+ toCompact.subList(limit, toCompact.size()).clear();
+ }
+
+ try
+ {
+ Collection<StaticSegment<K, V>> newSegments =
segmentCompactor.compact(toCompact);
+
+ for (StaticSegment<K, V> segment : newSegments)
+ toCompact.remove(segment);
+
+ journal.replaceCompactedSegments(toCompact, newSegments);
+ for (StaticSegment<K, V> segment : toCompact)
+ segment.discard(journal);
+ }
+ catch (IOException e)
+ {
+ throw new RuntimeException("Could not compact segments: " +
toCompact, e);
+ }
}
-
- try
+ finally
{
- Collection<StaticSegment<K, V>> newSegments =
segmentCompactor.compact(toCompact);
+ compacted.signalAll();
+ if (runPostCompaction)
+ segmentCompactor.onCompacted();
+ }
+ }
- for (StaticSegment<K, V> segment : newSegments)
- toCompact.remove(segment);
+ /**
+ * Runs a compaction pass ahead of the next scheduled tick, without
introducing any additional concurrency.
+ */
+ public void triggerNow()
+ {
+ if (journal.params.enableCompaction() && !executor.isShutdown() &&
triggerPending.compareAndSet(false, true))
+ executor.execute(this);
+ }
- journal.replaceCompactedSegments(toCompact, newSegments);
- for (StaticSegment<K, V> segment : toCompact)
- segment.discard(journal);
+ /**
+ * Like {@link #triggerNow()}, but runs on the same dedicated executor and
blocks until the pass completes.
+ * Skip when the executor is already shutdown
+ */
+ public void runNowBlocking()
+ {
+ submitBlocking(this);
+ }
- compacted.signalAll();
+ public void drainBlocking()
+ {
+ cancelPeriodic();
+ submitBlocking(() -> runInternal(false));
+ }
+
+ private void submitBlocking(Runnable compactorTask)
+ {
+ // Must not be called from the compactor executor thread:
submit(...).get() would self-deadlock
+ if (executor.isShutdown())
+ return;
+
+ try
+ {
+ executor.submit(compactorTask).get();
+ }
+ catch (InterruptedException e)
+ {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ catch (ExecutionException e)
+ {
+ Throwable cause = e.getCause();
+ Throwables.throwIfUnchecked(cause);
+ throw new RuntimeException(cause);
}
- catch (IOException e)
+ catch (CancellationException ignored)
{
- throw new RuntimeException("Could not compact segments: " +
toCompact);
+ // ignore when it's too late to schedule
}
}
+ private synchronized void cancelPeriodic()
+ {
+ if (scheduled == null)
+ return;
+
+ scheduled.cancel(false);
+ scheduled = null;
+ }
+
@Override
public boolean isTerminated()
{
@@ -115,7 +203,7 @@ public final class Compactor<K, V> implements Runnable,
Shutdownable
@Override
public void shutdown()
{
- logger.debug("Shutting down " + executor);
+ logger.debug("Shutting down {}", executor);
executor.shutdown();
}
@@ -130,4 +218,11 @@ public final class Compactor<K, V> implements Runnable,
Shutdownable
{
return executor.awaitTermination(timeout, units);
}
+
+ private List<StaticSegment<K, V>>
maybeWrapArrayList(Collection<StaticSegment<K, V>> collection)
+ {
+ if (collection instanceof ArrayList<?>)
+ return (List<StaticSegment<K, V>>) collection;
+ return new ArrayList<>(collection);
+ }
}
diff --git a/src/java/org/apache/cassandra/journal/Journal.java
b/src/java/org/apache/cassandra/journal/Journal.java
index 2519b58b26..f9569747f2 100644
--- a/src/java/org/apache/cassandra/journal/Journal.java
+++ b/src/java/org/apache/cassandra/journal/Journal.java
@@ -28,6 +28,7 @@ import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
@@ -304,6 +305,8 @@ public class Journal<K, V> implements Shutdownable
"Unexpected journal state during initialization:
%s", state);
metrics.register(flusher);
+ ensureJournalDirectoryExists();
+
deleteTmpFiles();
List<Descriptor> descriptors = Descriptor.list(directory);
@@ -323,14 +326,19 @@ public class Journal<K, V> implements Shutdownable
"Unexpected journal state after initialization",
state);
flusher.start();
compactor.start();
+ }
+
+ public void waitForCompaction(int maxSegments)
+ {
+ if (!params.enableCompaction())
+ return;
- final int maxSegments = 100;
- if (segments.get().count(Segment::isStatic) > maxSegments)
+ if (countStaticSegments() > maxSegments)
{
while (true)
{
WaitQueue.Signal signal = compactor.compacted.register();
- int count = segments.get().count(Segment::isStatic);
+ int count = countStaticSegments();
if (count <= maxSegments)
{
signal.cancel();
@@ -351,6 +359,11 @@ public class Journal<K, V> implements Shutdownable
return compactor;
}
+ private void ensureJournalDirectoryExists()
+ {
+ directory.createDirectoriesIfNotExists();
+ }
+
/**
* Cleans up unfinished component files from previous run (metadata and
index)
*/
@@ -961,11 +974,6 @@ public class Journal<K, V> implements Shutdownable
swapSegments(current -> current.withoutEmptySegment(activeSegment));
}
- private void removeStaticSegments(Collection<StaticSegment<K, V>>
staticSegments)
- {
- swapSegments(current -> current.withoutStaticSegments(staticSegments));
- }
-
private void replaceCompletedSegment(ActiveSegment<K, V> activeSegment,
StaticSegment<K, V> staticSegment)
{
swapSegments(current -> current.withCompletedSegment(activeSegment,
staticSegment));
@@ -1095,18 +1103,6 @@ public class Journal<K, V> implements Shutdownable
closer.execute(new CloseActiveSegmentRunnable(activeSegment, onDone));
}
- public int dropStaticSegments(Predicate<StaticSegment<K, V>> dropIf)
- {
- Set<StaticSegment<K, V>> toDrop = new HashSet<>();
- segments().selectStatic(dropIf, toDrop);
- if (toDrop.isEmpty())
- return 0;
- removeStaticSegments(toDrop);
- for (StaticSegment<K, V> segment : toDrop)
- segment.discard(this);
- return toDrop.size();
- }
-
/*
* Error handling
*/
@@ -1417,7 +1413,7 @@ public class Journal<K, V> implements Shutdownable
@VisibleForTesting
public void runCompactorForTesting()
{
- compactor.run();
+ compactor.runNowBlocking();
}
@VisibleForTesting
@@ -1431,6 +1427,19 @@ public class Journal<K, V> implements Shutdownable
{
LockSupport.parkNanos(1000);
}
+ try
+ {
+ closer.submit(() -> {}).get();
+ }
+ catch (InterruptedException e)
+ {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ catch (ExecutionException e)
+ {
+ throw new RuntimeException(e);
+ }
}
@VisibleForTesting
@@ -1441,9 +1450,20 @@ public class Journal<K, V> implements Shutdownable
toReset.forEach(s -> s.metadata().clearNeedsReplay());
}
+ public int countStaticSegments()
+ {
+ return segments.get().count(Segment::isStatic);
+ }
+
@VisibleForTesting
public int countStaticSegmentsForTesting()
{
- return segments.get().count(Segment::isStatic);
+ return countStaticSegments();
+ }
+
+ @VisibleForTesting
+ public int countStaticSegmentsForTesting(Predicate<StaticSegment<K, V>>
matches)
+ {
+ return segments().countStatic(matches);
}
}
diff --git a/src/java/org/apache/cassandra/journal/SegmentCompactor.java
b/src/java/org/apache/cassandra/journal/SegmentCompactor.java
index 7b84ea82e1..aa15313944 100644
--- a/src/java/org/apache/cassandra/journal/SegmentCompactor.java
+++ b/src/java/org/apache/cassandra/journal/SegmentCompactor.java
@@ -20,8 +20,13 @@ package org.apache.cassandra.journal;
import java.io.IOException;
import java.util.Collection;
+/**
+ * Decides which static segments of a {@link Journal} are compacted, and how.
Used by {@link Compactor}
+ * to run periodic or on-demand compaction passes.
+ */
public interface SegmentCompactor<K, V>
{
+ /** A no-op compactor */
SegmentCompactor<?, ?> NOOP = (SegmentCompactor<Object, Object>)
(segments) -> segments;
static <K, V> SegmentCompactor<K, V> noop()
@@ -30,5 +35,31 @@ public interface SegmentCompactor<K, V>
return (SegmentCompactor<K, V>) NOOP;
}
+ /**
+ * Picks which of the given candidate segments should be compacted this
pass. Defaults to selecting
+ * all of them.
+ *
+ * @param candidates all static segments currently eligible for compaction
+ * @return the subset of {@code candidates} to pass to {@link
#compact(Collection)}
+ */
+ default Collection<StaticSegment<K, V>> select(Collection<StaticSegment<K,
V>> candidates)
+ {
+ return candidates;
+ }
+
+ /**
+ * Compacts the given {@code segments}, previously chosen by {@link
#select(Collection)}.
+ *
+ * @param segments the segments to compact
+ * @return the new segments that replace them; empty when the segments are
dropped
+ */
Collection<StaticSegment<K, V>> compact(Collection<StaticSegment<K, V>>
segments) throws IOException;
+
+ /**
+ * Invoked at the end of every {@link Compactor} run, whether or not any
segments were selected or
+ * compacted this pass, and whether or not the pass failed.
+ */
+ default void onCompacted()
+ {
+ }
}
diff --git a/src/java/org/apache/cassandra/journal/Segments.java
b/src/java/org/apache/cassandra/journal/Segments.java
index 8c7db1f3cd..b7314b7faf 100644
--- a/src/java/org/apache/cassandra/journal/Segments.java
+++ b/src/java/org/apache/cassandra/journal/Segments.java
@@ -23,6 +23,8 @@ import java.util.function.Predicate;
import javax.annotation.Nullable;
+import com.google.common.annotations.VisibleForTesting;
+
import org.agrona.collections.Long2ObjectHashMap;
import accord.utils.Invariants;
@@ -75,21 +77,6 @@ public class Segments<K, V>
return new Segments<>(newSegments);
}
- Segments<K, V> withoutStaticSegments(Collection<StaticSegment<K, V>>
removeSegments)
- {
- if (removeSegments.isEmpty())
- return this;
-
- Long2ObjectHashMap<Segment<K, V>> newSegments = new
Long2ObjectHashMap<>(segments);
- for (StaticSegment<K, V> segment : removeSegments)
- {
- Segment<K, V> oldValue =
newSegments.remove(segment.descriptor.timestamp);
- if (oldValue != null)
- Invariants.require(oldValue.isStatic());
- }
- return new Segments<>(newSegments);
- }
-
Segments<K, V> withCompletedSegment(ActiveSegment<K, V> activeSegment,
StaticSegment<K, V> staticSegment)
{
Invariants.requireArgument(activeSegment.descriptor.equals(staticSegment.descriptor));
@@ -214,11 +201,14 @@ public class Segments<K, V>
into.add(segment.asStatic());
}
- void selectStatic(Predicate<StaticSegment<K, V>> filter,
Collection<StaticSegment<K, V>> into)
+ @VisibleForTesting
+ int countStatic(Predicate<StaticSegment<K, V>> filter)
{
+ int count = 0;
for (Segment<K, V> segment : segments.values())
if (segment.isStatic() && filter.test(segment.asStatic()))
- into.add(segment.asStatic());
+ count++;
+ return count;
}
/**
diff --git a/src/java/org/apache/cassandra/replication/CoordinatorLog.java
b/src/java/org/apache/cassandra/replication/CoordinatorLog.java
index 317a3e6eea..61835e40ad 100644
--- a/src/java/org/apache/cassandra/replication/CoordinatorLog.java
+++ b/src/java/org/apache/cassandra/replication/CoordinatorLog.java
@@ -595,7 +595,15 @@ public abstract class CoordinatorLog
void collectDurablyReconciledOffsets(Log2OffsetsMap.Mutable into)
{
- into.add(reconciledPersistedOffsets);
+ lock.readLock().lock();
+ try
+ {
+ into.add(reconciledPersistedOffsets);
+ }
+ finally
+ {
+ lock.readLock().unlock();
+ }
}
boolean isDurablyReconciled(ShortMutationId id)
diff --git a/src/java/org/apache/cassandra/replication/MutationJournal.java
b/src/java/org/apache/cassandra/replication/MutationJournal.java
index 1fa4909804..fa93fc7a54 100644
--- a/src/java/org/apache/cassandra/replication/MutationJournal.java
+++ b/src/java/org/apache/cassandra/replication/MutationJournal.java
@@ -27,6 +27,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiConsumer;
+import java.util.function.Supplier;
import java.util.zip.CRC32;
import javax.annotation.Nullable;
@@ -46,6 +47,7 @@ import accord.utils.Invariants;
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.DurationSpec;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.TypeSizes;
@@ -69,12 +71,12 @@ import org.apache.cassandra.journal.Params;
import org.apache.cassandra.journal.RecordConsumer;
import org.apache.cassandra.journal.RecordPointer;
import org.apache.cassandra.journal.Segment;
-import org.apache.cassandra.journal.SegmentCompactor;
import org.apache.cassandra.journal.StaticSegment;
import org.apache.cassandra.journal.ValueSerializer;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.utils.Crc;
import org.apache.cassandra.utils.concurrent.OpOrder;
import org.apache.cassandra.utils.concurrent.Semaphore;
@@ -101,7 +103,10 @@ public class MutationJournal
private static final MutationJournal instance =
DatabaseDescriptor.getMutationTrackingEnabled() ? new MutationJournal() : null;
private final Journal<ShortMutationId, Mutation> journal;
+ @VisibleForTesting
+ final MutationJournalSegmentCompactor segmentCompactor;
private final Map<Long, SegmentStateTracker> segmentStateTrackers;
+ private final SegmentReferenceTracker segmentReferenceTracker;
// Static segments awaiting durable cleanup of their needsReplay=false
metadata.
private final Set<Long> pendingClearReplay = ConcurrentHashMap.newKeySet();
@@ -169,14 +174,22 @@ public class MutationJournal
@VisibleForTesting
MutationJournal(File directory, Params params)
{
- journal =
+ this(directory, params, null);
+ }
+
+ @VisibleForTesting
+ MutationJournal(File directory, Params params, Supplier<Log2OffsetsMap<?>>
durablyReconciledOffsetsSupplier)
+ {
+ segmentStateTrackers = new NonBlockingHashMapLong<>();
+ segmentCompactor = new MutationJournalSegmentCompactor(this,
durablyReconciledOffsetsSupplier);
+ Journal<ShortMutationId, Mutation> journal =
new Journal<>("MutationJournal",
directory,
params,
MutationIdSupport.INSTANCE,
MutationSerializer.INSTANCE,
OffsetRangesFactory.INSTANCE,
- SegmentCompactor.noop(),
+ segmentCompactor,
new OpOrder())
{
// TODO (expected): a cleaner way to override
it; pass a Callbacks object with sanctioned callbacks?
@@ -189,7 +202,17 @@ public class MutationJournal
});
}
};
- segmentStateTrackers = new NonBlockingHashMapLong<>();
+ this.journal = journal;
+ segmentReferenceTracker = new SegmentReferenceTracker(() ->
journal.compactor().triggerNow());
+ }
+
+ /**
+ * Triggers the journal's compactor to initiate a segment-dropping +
promotion pass
+ * (see {@link MutationJournalSegmentCompactor}
+ */
+ private void triggerSegmentCompaction()
+ {
+ journal.compactor().triggerNow();
}
public CommitLogPosition getCurrentPosition()
@@ -207,7 +230,7 @@ public class MutationJournal
{
Invariants.require(segment.isStatic());
SegmentStateTracker tracker = segmentStateTrackers.get(segment.id());
- if (tracker != null && tracker.removeCleanFromDirty())
+ if (tracker == null || tracker.removeCleanFromDirty())
pendingClearReplay.add(segment.id());
}
@@ -226,6 +249,7 @@ public class MutationJournal
*/
public void drainCleanup(PendingClearReplay toDrain)
{
+ boolean anyCleared = false;
for (long segId : toDrain.segments)
{
List<Segment<ShortMutationId, Mutation>> found =
journal.getSegments(segId, segId);
@@ -241,6 +265,7 @@ public class MutationJournal
segment.metadata().clearNeedsReplay();
segment.persistMetadata();
pendingClearReplay.remove(segId);
+ anyCleared = true;
}
catch (Throwable t)
{
@@ -248,6 +273,9 @@ public class MutationJournal
// leave in live queue
}
}
+
+ if (anyCleared)
+ triggerSegmentCompaction();
}
@VisibleForTesting
@@ -273,23 +301,40 @@ public class MutationJournal
journal.shutdown();
}
+ @VisibleForTesting
public RecordPointer write(ShortMutationId id, Mutation mutation)
+ {
+ return write(id, mutation, true);
+ }
+
+ /**
+ * Append a mutation to the journal.
+ *
+ * @param id the short mutation id
+ * @param mutation the mutation to be applied to the journal
+ * @param isFullReplica whether this node is a full replica for the
mutation's token
+ * @return the record pointer to the journal
+ */
+ public RecordPointer write(ShortMutationId id, Mutation mutation, boolean
isFullReplica)
{
// TODO (required): why are we using blocking write here? We
can/should wait for completion on `close` of WriteContext.
RecordPointer ptr = journal.blockingWrite(id, mutation);
// IMPORTANT: there should be no way for mutation to be applied to
memtable before we mark it as dirty here,
// since this will introduce a race between marking as dirty and
marking as clean.
- for (TableId tableId : mutation.getTableIds())
+ if (isFullReplica)
{
- SegmentStateTracker tracker = lastSegmentTracker;
- if (tracker == null || tracker.segmentId() != ptr.segmentId)
+ for (TableId tableId : mutation.getTableIds())
{
- tracker = segmentStateTrackers.computeIfAbsent(ptr.segmentId,
SegmentStateTracker::new);
- lastSegmentTracker = tracker;
- }
+ SegmentStateTracker tracker = lastSegmentTracker;
+ if (tracker == null || tracker.segmentId() != ptr.segmentId)
+ {
+ tracker =
segmentStateTrackers.computeIfAbsent(ptr.segmentId, SegmentStateTracker::new);
+ lastSegmentTracker = tracker;
+ }
- tracker.markDirty(tableId, ptr);
+ tracker.markDirty(tableId, ptr);
+ }
}
return ptr;
@@ -378,6 +423,7 @@ public class MutationJournal
return;
// TODO: if (commitLogReplayer.pointInTimeExceeded(mutation))
final Keyspace keyspace =
Keyspace.open(value.getKeyspaceName());
+ final boolean isFullReplica =
keyspace.isFullReplicaFor(value.key().getToken(), ClusterMetadata.current());
Mutation.PartitionUpdateCollector newPUCollector = null;
// TODO (required): replayFilter
@@ -389,9 +435,9 @@ public class MutationJournal
continue; // dropped
TableId tableId = e.getKey();
- // Start segment state tracking
- segmentStateTrackers.computeIfAbsent(segmentId,
SegmentStateTracker::new)
- .markDirty(tableId, segmentId,
position);
+ if (isFullReplica)
+ segmentStateTrackers.computeIfAbsent(segmentId,
SegmentStateTracker::new)
+ .markDirty(tableId, segmentId,
position);
// TODO (required): shouldReplay
if (newPUCollector == null)
newPUCollector = new
Mutation.PartitionUpdateCollector(value.id(), value.getKeyspaceName(),
value.key());
@@ -440,13 +486,25 @@ public class MutationJournal
}
}
- @VisibleForTesting
- public int dropReconciledSegments(Log2OffsetsMap<?> reconciledOffsets)
+ void runCompactionBlocking()
+ {
+ journal.compactor().runNowBlocking();
+ }
+
+ /**
+ * Reclaim droppable segments without sstable promotion
+ */
+ void drainBlocking()
+ {
+ journal.compactor().drainBlocking();
+ }
+
+ /**
+ * Listener tracking how many unrepaired sstables of tracked tables
reference each static segment.
+ */
+ public SegmentReferenceTracker segmentReferenceTracker()
{
- return journal.dropStaticSegments((segment) -> {
- StaticOffsetRanges ranges = (StaticOffsetRanges)
segment.keyStats();
- return ranges.isFullyCovered(reconciledOffsets) &&
!segment.metadata().needsReplay();
- });
+ return segmentReferenceTracker;
}
public void readAll(RecordConsumer<ShortMutationId> consumer)
@@ -456,6 +514,9 @@ public class MutationJournal
static class JournalParams implements Params
{
+ /** for now match {@link
MutationTrackingService.LogStatePersister#PERSIST_INTERVAL_MILLIS} **/
+ public DurationSpec.IntMillisecondsBound compactionPeriod = new
DurationSpec.IntMillisecondsBound("1s");
+
@Override
public int segmentSize()
{
@@ -488,7 +549,7 @@ public class MutationJournal
@Override
public int compactMaxSegments()
{
- return 0;
+ return Integer.MAX_VALUE;
}
@Override
@@ -500,13 +561,13 @@ public class MutationJournal
@Override
public boolean enableCompaction()
{
- return false;
+ return
DatabaseDescriptor.getMutationTrackingConfig().journal_compaction_enabled;
}
@Override
- public long compactionPeriod(TimeUnit units)
+ public long compactionPeriod(TimeUnit unit)
{
- return 0;
+ return compactionPeriod.to(unit);
}
@Override
@@ -804,8 +865,11 @@ public class MutationJournal
}
/**
- * @return whether all keys in the segment are fully covered by the
specified (durably reconciled) offsets map
+ * @return whether every key range in this segment is fully covered by
the given (durably reconciled)
+ * offsets — i.e. every mutation id the segment holds has been durably
reconciled across peers. Used to
+ * decide when a segment holding witnessed-only data may be dropped.
*/
+ @SuppressWarnings("unchecked")
boolean isFullyCovered(Log2OffsetsMap<?> durablyReconciled)
{
Long2ObjectHashMap<Offsets> reconciledMap =
((Log2OffsetsMap<Offsets>) durablyReconciled).asMap();
@@ -819,9 +883,8 @@ public class MutationJournal
int max = maxOffset(range);
Offsets offsets = reconciledMap.get(logId);
- if (offsets == null)
- return false;
- if (!offsets.containsRange(min, max))
+ if (offsets == null
+ || !offsets.containsRange(min, max))
return false;
}
return true;
@@ -884,7 +947,19 @@ public class MutationJournal
@VisibleForTesting
public int countStaticSegmentsForTesting()
{
- return journal.countStaticSegmentsForTesting();
+ return journal.countStaticSegments();
+ }
+
+ /**
+ * Lets tests wait for reconciliation to converge without having to first
release the sstable references
+ * that gate (R).
+ * @return the number of static segments not yet fully covered by the
given durably-reconciled offsets.
+ */
+ @VisibleForTesting
+ public int
countStaticSegmentsPendingReconciliationForTesting(Log2OffsetsMap<?>
durablyReconciled)
+ {
+ return journal.countStaticSegmentsForTesting(
+ segment -> !((StaticOffsetRanges)
segment.keyStats()).isFullyCovered(durablyReconciled));
}
public long getDiskSpaceUsed()
diff --git
a/src/java/org/apache/cassandra/replication/MutationJournalSegmentCompactor.java
b/src/java/org/apache/cassandra/replication/MutationJournalSegmentCompactor.java
new file mode 100644
index 0000000000..bc16343571
--- /dev/null
+++
b/src/java/org/apache/cassandra/replication/MutationJournalSegmentCompactor.java
@@ -0,0 +1,233 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.replication;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.Mutation;
+import org.apache.cassandra.io.sstable.format.SSTableReader;
+import org.apache.cassandra.io.sstable.metadata.StatsMetadata;
+import org.apache.cassandra.journal.SegmentCompactor;
+import org.apache.cassandra.journal.StaticSegment;
+import org.apache.cassandra.schema.ReplicationType;
+import org.apache.cassandra.service.ActiveRepairService;
+import
org.apache.cassandra.service.replication.migration.KeyspaceMigrationInfo;
+import
org.apache.cassandra.service.replication.migration.MutationTrackingMigrationState;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.utils.Clock;
+import org.apache.cassandra.utils.JVMStabilityInspector;
+import org.apache.cassandra.utils.NoSpamLogger;
+
+/**
+ * Segment compactor: takes static segments, selects the ones that are safe to
drop, and drops them
+ */
+class MutationJournalSegmentCompactor implements
SegmentCompactor<ShortMutationId, Mutation>
+{
+ private static final Logger logger =
LoggerFactory.getLogger(MutationJournalSegmentCompactor.class);
+
+ private static final NoSpamLogger noSpamLogger =
NoSpamLogger.getLogger(logger, 1L, TimeUnit.MINUTES);
+
+ private final MutationJournal journal;
+
+ // for testing only
+ private final Supplier<Log2OffsetsMap<?>> durablyReconciledOffsetsSupplier;
+
+ @VisibleForTesting
+ MutationJournalSegmentCompactor(MutationJournal journal,
Supplier<Log2OffsetsMap<?>> durablyReconciledOffsetsSupplier)
+ {
+ this.journal = journal;
+ this.durablyReconciledOffsetsSupplier =
durablyReconciledOffsetsSupplier;
+ }
+
+ /**
+ * Returns the list of segments that do not need to be replayed, and that
are not referenced by a tracked
+ * SSTable, and that are durably reconciled over the full {@link
#durablyReconciledOffsets durably reconciled offsets}.
+ *
+ * @param candidates the list of potential candidate segments
+ * @return a subset of {@code candidates} that are safe to drop
+ */
+ @Override
+ public Collection<StaticSegment<ShortMutationId, Mutation>>
select(Collection<StaticSegment<ShortMutationId, Mutation>> candidates)
+ {
+ Log2OffsetsMap<?> durablyReconciled = durablyReconciledOffsets();
+ List<StaticSegment<ShortMutationId, Mutation>> selected = new
ArrayList<>();
+ for (StaticSegment<ShortMutationId, Mutation> segment : candidates)
+ {
+ if (!segment.metadata().needsReplay()
+ &&
!journal.segmentReferenceTracker().isReferenced(segment.id())
+ && ((MutationJournal.StaticOffsetRanges)
segment.keyStats()).isFullyCovered(durablyReconciled))
+ selected.add(segment);
+ }
+ return selected;
+ }
+
+ /**
+ * We drop every segment, since already know that the selected segments
are safe to be discarded
+ */
+ @Override
+ public Collection<StaticSegment<ShortMutationId, Mutation>>
compact(Collection<StaticSegment<ShortMutationId, Mutation>> segments)
+ {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public void onCompacted()
+ {
+ try
+ {
+ maybePromoteReconciledSSTables();
+ }
+ catch (Throwable t)
+ {
+ JVMStabilityInspector.inspectThrowable(t);
+ noSpamLogger.warn("Failed to promote reconciled sstables to
repaired; will retry on a later pass", t);
+ }
+ }
+
+ Log2OffsetsMap<?> durablyReconciledOffsets()
+ {
+ if (durablyReconciledOffsetsSupplier == null) //
durablyReconciledOffsetsSupplier is only used for testing
+ {
+ Log2OffsetsMap.Mutable durablyReconciled = new
Log2OffsetsMap.Mutable();
+ if (MutationTrackingService.isEnabled())
+
MutationTrackingService.instance().collectDurablyReconciledOffsets(durablyReconciled);
+ return durablyReconciled;
+ }
+ return durablyReconciledOffsetsSupplier.get();
+ }
+
+ /**
+ * Out-of-band promotion of already durably-reconciled but
still-unrepaired sstables to repaired, attempted at
+ * the end of every compaction pass while the on-disk mutation journal is
over the
+ * {@code mutation_tracking.journal_promotion_threshold}.
+ *
+ * <p>Best-effort: an sstable we cannot make a decision about is skipped,
and a failure to flip
+ * one table's sstables is logged and retried on a later pass.
+ */
+ @VisibleForTesting
+ void maybePromoteReconciledSSTables()
+ {
+ long threshold =
DatabaseDescriptor.getMutationTrackingConfig().getJournalPromotionThresholdBytes();
+ if (threshold <= 0)
+ return;
+
+ ClusterMetadata metadata = ClusterMetadata.currentNullable();
+ if (metadata == null)
+ return;
+
+ Set<SSTableReader> trackedSSTables =
journal.segmentReferenceTracker().trackedSSTables();
+ if (trackedSSTables.isEmpty() || journal.getDiskSpaceUsed() <=
threshold)
+ return;
+
+ MutationTrackingMigrationState mutationTrackingMigrationState =
metadata.mutationTrackingMigrationState;
+ long repairedAt = Clock.Global.currentTimeMillis();
+ Map<ColumnFamilyStore, List<SSTableReader>> toPromoteByTable = new
HashMap<>();
+ for (SSTableReader sstable : trackedSSTables)
+ {
+ ColumnFamilyStore cfs;
+ try
+ {
+ if
(!isReconciliationPromotable(mutationTrackingMigrationState, sstable))
+ continue;
+
+ cfs = ColumnFamilyStore.getIfExists(sstable.metadata().id);
+ }
+ catch (Throwable t)
+ {
+ JVMStabilityInspector.inspectThrowable(t);
+ logger.debug("Skipping sstable ({}) for reconciliation
promotion; could not determine eligibility",
+ sstable.getId(), t);
+ continue;
+ }
+
+ if (cfs == null)
+ continue;
+
+ toPromoteByTable.computeIfAbsent(cfs, ignore -> new
ArrayList<>()).add(sstable);
+ }
+
+ if (toPromoteByTable.isEmpty())
+ return;
+
+ for (Map.Entry<ColumnFamilyStore, List<SSTableReader>> entry :
toPromoteByTable.entrySet())
+ {
+ ColumnFamilyStore cfs = entry.getKey();
+ List<SSTableReader> toPromote = entry.getValue();
+ try
+ {
+ cfs.getCompactionStrategyManager().mutateRepaired(toPromote,
repairedAt, ActiveRepairService.NO_PENDING_REPAIR);
+ logger.debug("Promoted {} reconciled sstables of {}.{} to
repaired to release journal segments",
+ toPromote.size(), cfs.getKeyspaceName(),
cfs.getTableName());
+ }
+ catch (Throwable t)
+ {
+ JVMStabilityInspector.inspectThrowable(t);
+ noSpamLogger.warn("Failed to promote reconciled sstables of
{}.{} to repaired; will retry",
+ cfs.getKeyspaceName(), cfs.getTableName(),
t);
+ }
+ }
+ }
+
+ /**
+ * Whether the {@code sstable} is eligible for reconciliation-based
promotion to repaired.
+ *
+ * <p>We replicate the guard in {@link
org.apache.cassandra.io.sstable.format.SSTableWriter#finalizeMetadata()}
+ * to exclude sstables that have a pending migration.
+ *
+ * <p>During {@code MIGRATE_TO} writes are tracked for all tokens while
reads are not. So we exclude
+ * sstables already owned by a pending repair session.
+ *
+ * @param state the mutation tracking migration state
+ * @param sstable the sstable to test
+ * @return true if the sstable should be considered for promotion to
repaired
+ */
+ private boolean isReconciliationPromotable(MutationTrackingMigrationState
state, SSTableReader sstable)
+ {
+ StatsMetadata stats = sstable.getSSTableMetadata();
+ if (stats.repairedAt != ActiveRepairService.UNREPAIRED_SSTABLE ||
stats.pendingRepair != ActiveRepairService.NO_PENDING_REPAIR)
+ return false;
+
+ ReplicationType replicationType = sstable.metadata().replicationType();
+ if (replicationType == null || !replicationType.isTracked())
+ return false;
+
+ KeyspaceMigrationInfo migrationInfo =
state.getKeyspaceInfo(sstable.metadata().keyspace);
+ boolean inMigrationPendingRange = migrationInfo != null &&
migrationInfo.isRangeInPendingMigration(sstable.metadata().id,
+
sstable.getFirst().getToken(),
+
sstable.getLast().getToken());
+ if (inMigrationPendingRange)
+ return false;
+
+ return
MutationTrackingService.instance().isDurablyReconciled(stats.coordinatorLogOffsets);
+ }
+}
diff --git
a/src/java/org/apache/cassandra/replication/MutationTrackingService.java
b/src/java/org/apache/cassandra/replication/MutationTrackingService.java
index db87a88f01..9f6d4e395a 100644
--- a/src/java/org/apache/cassandra/replication/MutationTrackingService.java
+++ b/src/java/org/apache/cassandra/replication/MutationTrackingService.java
@@ -53,6 +53,7 @@ import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.DurationSpec;
import org.apache.cassandra.config.MutationTrackingSpec;
import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.Keyspace;
@@ -252,8 +253,14 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
onNewClusterMetadata(null, metadata);
- if (!keyspaceShards.isEmpty() &&
!config.background_reconciliation_enabled)
-
logBackgroundReconciliationDisabledWarning(keyspaceShards.keySet());
+ if (!keyspaceShards.isEmpty())
+ {
+ if (!config.background_reconciliation_enabled)
+
logBackgroundReconciliationDisabledWarning(keyspaceShards.keySet());
+
+ if (!config.journal_compaction_enabled)
+ logJournalCompactionDisabledWarning(keyspaceShards.keySet());
+ }
offsetsBroadcaster.start();
offsetsPersister.start();
@@ -339,10 +346,14 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
if (!executor.awaitTermination(1, TimeUnit.MINUTES))
logger.warn("Mutation tracking executor did not terminate within 1
minute; forcing shutdown");
- // attempt to persist offsets and mark segments as
- // not needing replay one last time before shutdown
+ // attempt to persist offsets and mark segments as not needing replay
one last time before shutdown, then
+ // reclaim any now-droppable segments synchronously (the executor is
stopped, so scheduled attempts from
+ // drainCleanup would never run).
if (isStarted())
- offsetsPersister.run(true);
+ {
+ offsetsPersister.persistAndDrain();
+ MutationJournal.instance().drainBlocking();
+ }
ExpiredStatePurger.instance.shutdownBlocking();
}
@@ -1005,6 +1016,7 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
{
case MIGRATE_FROM:
// TODO (expected): Implement shard deletion for tracked →
untracked migration completion (CASSANDRA-20955)
+ evictSegmentReferences(keyspace);
case NONE:
if (current != null)
updated.put(keyspace, current);
@@ -1041,6 +1053,20 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
return updated;
}
+ private static void evictSegmentReferences(String keyspace)
+ {
+ if (!DatabaseDescriptor.getMutationTrackingEnabled())
+ return;
+
+ Keyspace ks = Schema.instance.getKeyspaceInstance(keyspace);
+ if (ks == null)
+ return;
+
+ SegmentReferenceTracker tracker =
MutationJournal.instance().segmentReferenceTracker();
+ for (ColumnFamilyStore cfs : ks.getColumnFamilyStores())
+ tracker.evict(cfs.getLiveSSTables());
+ }
+
// TODO (expected): when topology and state truncation is implemented,
implement cleanup of this map as well
private void onNewLog(Shard shard, CoordinatorLog log)
{
@@ -1055,21 +1081,26 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
}
}
- private void truncateMutationJournal()
+ void collectDurablyReconciledOffsets(Log2OffsetsMap.Mutable into)
{
- Log2OffsetsMap.Mutable reconciledOffsets = new
Log2OffsetsMap.Mutable();
- collectDurablyReconciledOffsets(reconciledOffsets);
- MutationJournal.instance().dropReconciledSegments(reconciledOffsets);
+ forEachKeyspace(keyspace ->
keyspace.collectDurablyReconciledOffsets(into));
}
/**
- * Collect every log's durably reconciled offsets. Every mutation covered
- * by these offsets can be compacted away by the journal, assuming that all
- * relevant memtables had been flushed to disk.
+ * @return true once every static journal segment is fully covered by the
durably-reconciled offsets
*/
- private void collectDurablyReconciledOffsets(Log2OffsetsMap.Mutable into)
+ @VisibleForTesting
+ public boolean allStaticSegmentsDurablyReconciledForTesting()
{
- forEachKeyspace(keyspace ->
keyspace.collectDurablyReconciledOffsets(into));
+ Log2OffsetsMap.Mutable durablyReconciled = new
Log2OffsetsMap.Mutable();
+ collectDurablyReconciledOffsets(durablyReconciled);
+ return
MutationJournal.instance().countStaticSegmentsPendingReconciliationForTesting(durablyReconciled)
== 0;
+ }
+
+ @VisibleForTesting
+ public void maybePromoteReconciledSSTablesForTesting()
+ {
+
MutationJournal.instance().segmentCompactor.maybePromoteReconciledSSTables();
}
public SyncTasks alignToShardBoundaries(Keyspace keyspace, List<SyncTask>
tasks)
@@ -1138,6 +1169,14 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
"Unreconciled mutations will not be automatically repaired
in the background.", keyspaces);
}
+ private void logJournalCompactionDisabledWarning(Set<String> keyspaces)
+ {
+ logger.warn("Mutation journal compaction is disabled but mutation
tracking keyspaces exist: {}. " +
+ "Journal segments will not be reclaimed and
durably-reconciled sstables will not be promoted " +
+ "to repaired
(mutation_tracking.journal_promotion_threshold has no effect while compaction
is " +
+ "disabled); the on-disk journal will grow without bound.",
keyspaces);
+ }
+
public static class KeyspaceShards
{
private enum UpdateDecision
@@ -1380,17 +1419,17 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
});
}
- void collectDurablyReconciledOffsets(Log2OffsetsMap.Mutable into)
- {
- forEachShard(shard -> shard.collectDurablyReconciledOffsets(into));
- }
-
void forEachShard(Consumer<Shard> consumer)
{
for (Shard shard : shards.values())
consumer.accept(shard);
}
+ void collectDurablyReconciledOffsets(Log2OffsetsMap.Mutable into)
+ {
+ forEachShard(shard -> shard.collectDurablyReconciledOffsets(into));
+ }
+
Shard lookUp(Mutation mutation)
{
return lookUp(mutation.key());
@@ -1657,12 +1696,11 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
{
if (isPaused)
return;
- run(true);
+ persistAndDrain();
}
- private void run(boolean dropSegments)
+ void persistAndDrain()
{
-
MutationJournal.PendingClearReplay toDrain =
MutationJournal.instance().snapshotPendingClearReplay();
boolean writesOk;
@@ -1679,9 +1717,6 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
if (writesOk)
MutationJournal.instance().drainCleanup(toDrain);
-
- if (dropSegments)
- MutationTrackingService.instance().truncateMutationJournal();
}
private void run(KeyspaceShards shards)
@@ -1698,13 +1733,15 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
@VisibleForTesting
public void persistLogStateForTesting()
{
- offsetsPersister.run();
+ persistLogStateForTesting(true);
}
@VisibleForTesting
public void persistLogStateForTesting(boolean dropSegments)
{
- offsetsPersister.run(dropSegments);
+ offsetsPersister.persistAndDrain();
+ if (dropSegments)
+ MutationJournal.instance().runCompactionBlocking();
}
@VisibleForTesting
diff --git
a/src/java/org/apache/cassandra/replication/SegmentReferenceTracker.java
b/src/java/org/apache/cassandra/replication/SegmentReferenceTracker.java
new file mode 100644
index 0000000000..45c9f95517
--- /dev/null
+++ b/src/java/org/apache/cassandra/replication/SegmentReferenceTracker.java
@@ -0,0 +1,385 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.replication;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+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.apache.cassandra.db.commitlog.CommitLogPosition;
+import org.apache.cassandra.db.commitlog.IntervalSet;
+import org.apache.cassandra.io.sstable.format.SSTableReader;
+import org.apache.cassandra.notifications.INotification;
+import org.apache.cassandra.notifications.INotificationConsumer;
+import org.apache.cassandra.notifications.InitialSSTableAddedNotification;
+import org.apache.cassandra.notifications.SSTableAddedNotification;
+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.
+ *
+ * <p>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.
+ *
+ * <p>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).
+ *
+ * <p>A per-segment <em>set</em> 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
+ private final Long2ObjectHashMap<Set<SSTableReader>> 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<SSTableReader> trackedSSTables = new HashSet<>();
+
+ // 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<UUID> localHostIdSupplier;
+
+ public SegmentReferenceTracker(Runnable onSegmentsUnreferenced)
+ {
+ this(onSegmentsUnreferenced,
StorageService.instance::getLocalHostUUID);
+ }
+
+ @VisibleForTesting
+ SegmentReferenceTracker(Runnable onSegmentsUnreferenced, Supplier<UUID>
localHostIdSupplier)
+ {
+ this.onSegmentsUnreferenced = onSegmentsUnreferenced;
+ this.localHostIdSupplier = localHostIdSupplier;
+ }
+
+ @Override
+ public void handleNotification(INotification notification, Object sender)
+ {
+ if (notification instanceof SSTableAddedNotification)
+ onAdded(((SSTableAddedNotification) notification).added);
+ else if (notification instanceof InitialSSTableAddedNotification)
+ onAdded(((InitialSSTableAddedNotification) notification).added);
+ else if (notification instanceof SSTableListChangedNotification)
+ onListChanged((SSTableListChangedNotification) notification);
+ else if (notification instanceof SSTableRepairStatusChanged)
+ onRepairStatusChanged(((SSTableRepairStatusChanged)
notification).sstables);
+ }
+
+ /**
+ * @param segmentId the identifier for the segment
+ * @return whether any unrepaired local sstable references the given
segment id
+ */
+ boolean isReferenced(long segmentId)
+ {
+ lock.lock();
+ try
+ {
+ return referrersBySegment.containsKey(segmentId);
+ }
+ finally
+ {
+ lock.unlock();
+ }
+ }
+
+ private void onAdded(Iterable<SSTableReader> added)
+ {
+ lock.lock();
+ try
+ {
+ for (SSTableReader sstable : added)
+ acquireIfTracked(sstable);
+ }
+ finally
+ {
+ lock.unlock();
+ }
+ }
+
+ private void onListChanged(SSTableListChangedNotification notification)
+ {
+ boolean anyReleased = false;
+ 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)
+ anyReleased |= releaseIfTracked(sstable);
+ }
+ finally
+ {
+ lock.unlock();
+ }
+ if (anyReleased)
+ onSegmentsUnreferenced.run();
+ }
+
+ private void onRepairStatusChanged(Collection<SSTableReader> changed)
+ {
+ boolean anyReleased = false;
+ lock.lock();
+ try
+ {
+ for (SSTableReader sstable : changed)
+ {
+ if (shouldTrack(sstable))
+ acquireIfTracked(sstable);
+ else
+ anyReleased |= releaseIfTracked(sstable);
+ }
+ }
+ finally
+ {
+ lock.unlock();
+ }
+ if (anyReleased)
+ onSegmentsUnreferenced.run();
+ }
+
+ /**
+ * Release every reference currently held for the given sstables, dropping
each from every segment's referrer
+ * set.
+ */
+ public void evict(Iterable<SSTableReader> sstables)
+ {
+ boolean anyReleased = false;
+ lock.lock();
+ try
+ {
+ for (SSTableReader sstable : sstables)
+ anyReleased |= releaseIfTracked(sstable);
+ }
+ finally
+ {
+ lock.unlock();
+ }
+ if (anyReleased)
+ onSegmentsUnreferenced.run();
+ }
+
+ /**
+ * @return an immutable snapshot of the currently tracked SSTables
+ */
+ public Set<SSTableReader> trackedSSTables()
+ {
+ lock.lock();
+ try
+ {
+ if (trackedSSTables.isEmpty())
+ return Set.of();
+ return Set.copyOf(trackedSSTables);
+ }
+ finally
+ {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * @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)
+ && !sstable.isRepaired()
+ && !sstable.getCoordinatorLogOffsets().isEmpty()
+ && isTrackedTable(sstable);
+ }
+
+ private static boolean isTrackedTable(SSTableReader sstable)
+ {
+ ReplicationType replicationType = sstable.metadata().replicationType();
+ return replicationType != null && replicationType.isTracked();
+ }
+
+ private boolean isLocallyOriginated(SSTableReader sstable)
+ {
+ UUID originatingHostId =
sstable.getSSTableMetadata().originatingHostId;
+ UUID localHostId = localHostIdSupplier.get();
+ return originatingHostId != null &&
originatingHostId.equals(localHostId);
+ }
+
+ private void acquireIfTracked(SSTableReader sstable)
+ {
+ if (shouldTrack(sstable) && trackedSSTables.add(sstable))
+ forEachSegment(sstable, segmentId ->
+
referrersBySegment.computeIfAbsent(segmentId, k -> new
HashSet<>()).add(sstable));
+ }
+
+ /**
+ * @return true if releasing this sstable emptied at least one segment's
referrer set
+ */
+ private boolean releaseIfTracked(SSTableReader sstable)
+ {
+ if (!trackedSSTables.remove(sstable))
+ return false;
+ boolean[] anyEmptied = { false };
+ forEachSegment(sstable, segmentId -> {
+ Set<SSTableReader> referrers = referrersBySegment.get(segmentId);
+ if (referrers != null && referrers.remove(sstable) &&
referrers.isEmpty())
+ {
+ referrersBySegment.remove(segmentId);
+ anyEmptied[0] = true;
+ }
+ });
+ return anyEmptied[0];
+ }
+
+ private static void forEachSegment(SSTableReader sstable, LongConsumer
consumer)
+ {
+ IntervalSet<CommitLogPosition> intervals =
sstable.getSSTableMetadata().commitLogIntervals;
+ if (intervals.isEmpty())
+ return;
+
+ // IntervalSet guarantees starts and ends are returned in matching
order.
+ Iterator<CommitLogPosition> startIt = intervals.starts().iterator();
+ Iterator<CommitLogPosition> 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);
+ }
+ }
+
+ /**
+ * Number of unrepaired local sstables currently holding the given segment
(for diagnostics / the vtable).
+ */
+ int referenceCount(long segmentId)
+ {
+ lock.lock();
+ try
+ {
+ Set<SSTableReader> referrers = referrersBySegment.get(segmentId);
+ return referrers == null ? 0 : referrers.size();
+ }
+ finally
+ {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Sorted base filenames of the sstables currently holding the given
segment (for diagnostics / the vtable).
+ */
+ public List<String> referrerDescriptors(long segmentId)
+ {
+ lock.lock();
+ try
+ {
+ Set<SSTableReader> referrers = referrersBySegment.get(segmentId);
+ if (referrers == null || referrers.isEmpty())
+ return Collections.emptyList();
+ List<String> names = new ArrayList<>(referrers.size());
+ for (SSTableReader sstable : referrers)
+ names.add(sstable.descriptor.baseFile().name());
+ Collections.sort(names);
+ return names;
+ }
+ finally
+ {
+ lock.unlock();
+ }
+ }
+
+ @VisibleForTesting
+ long referenceCountForTesting(long segmentId)
+ {
+ return referenceCount(segmentId);
+ }
+
+ @VisibleForTesting
+ int trackedSstableCountForTesting()
+ {
+ lock.lock();
+ try
+ {
+ return trackedSSTables.size();
+ }
+ finally
+ {
+ lock.unlock();
+ }
+ }
+
+ @VisibleForTesting
+ void addReferenceForTesting(long segmentId, SSTableReader referrer)
+ {
+ lock.lock();
+ try
+ {
+ referrersBySegment.computeIfAbsent(segmentId, k -> new
HashSet<>()).add(referrer);
+ }
+ finally
+ {
+ lock.unlock();
+ }
+ }
+
+ @VisibleForTesting
+ void removeReferenceForTesting(long segmentId, SSTableReader referrer)
+ {
+ lock.lock();
+ try
+ {
+ Set<SSTableReader> referrers = referrersBySegment.get(segmentId);
+ if (referrers != null && referrers.remove(referrer) &&
referrers.isEmpty())
+ referrersBySegment.remove(segmentId);
+ }
+ finally
+ {
+ lock.unlock();
+ }
+ }
+}
diff --git a/src/java/org/apache/cassandra/replication/Shard.java
b/src/java/org/apache/cassandra/replication/Shard.java
index dd4e349023..e8bd73c2e9 100644
--- a/src/java/org/apache/cassandra/replication/Shard.java
+++ b/src/java/org/apache/cassandra/replication/Shard.java
@@ -317,7 +317,8 @@ public class Shard
void collectDurablyReconciledOffsets(Log2OffsetsMap.Mutable into)
{
- logs.values().forEach(log ->
log.collectDurablyReconciledOffsets(into));
+ for (CoordinatorLog log : logs.values())
+ log.collectDurablyReconciledOffsets(into);
}
private CoordinatorLog getOrCreate(Mutation mutation)
diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java
b/src/java/org/apache/cassandra/service/accord/AccordJournal.java
index a255a890fc..5f13593aed 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java
@@ -202,6 +202,7 @@ public class AccordJournal implements accord.api.Journal,
RangeSearcher.Supplier
// start table first to scrub directories before compactor starts
journalTable.start();
journal.start();
+ journal.waitForCompaction(100);
}
public boolean started()
@@ -1253,4 +1254,4 @@ public class AccordJournal implements accord.api.Journal,
RangeSearcher.Supplier
}
}
}
-}
\ No newline at end of file
+}
diff --git
a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationJournalSegmentRefcountTest.java
b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationJournalSegmentRefcountTest.java
new file mode 100644
index 0000000000..b21f3bddcd
--- /dev/null
+++
b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationJournalSegmentRefcountTest.java
@@ -0,0 +1,318 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test.tracking;
+
+import org.junit.Test;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.net.Verb;
+import org.apache.cassandra.replication.MutationJournal;
+import org.apache.cassandra.replication.MutationTrackingService;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * End-to-end coverage of CASSANDRA-21406: a static journal segment must be
retained while any
+ * unrepaired sstable references it, and must be droppable once every
referencing sstable has
+ * been promoted to repaired (e.g. by compaction once mutations are durably
reconciled).
+ */
+public class MutationJournalSegmentRefcountTest extends TestBaseImpl
+{
+ private static final String CREATE_KEYSPACE =
+ "CREATE KEYSPACE %s WITH replication = {'class': 'SimpleStrategy',
'replication_factor': 3} " +
+ "AND replication_type = 'tracked'";
+
+ private static final String CREATE_TABLE = "CREATE TABLE %s.tbl (pk int
PRIMARY KEY, val text)";
+
+ @Test(timeout = 120_000)
+ public void testSegmentRetainedUntilSSTableRepaired() throws Throwable
+ {
+ try (Cluster cluster = Cluster.build(3)
+ .withConfig(cfg ->
cfg.with(Feature.NETWORK).with(Feature.GOSSIP))
+ .start())
+ {
+ cluster.schemaChange(withKeyspace(CREATE_KEYSPACE));
+ cluster.schemaChange(String.format(CREATE_TABLE, KEYSPACE));
+
+ // Disable autocompaction so flushed sstables stay until we
explicitly compact.
+ cluster.forEach(i -> i.nodetoolResult("disableautocompaction",
KEYSPACE, "tbl").asserts().success());
+
+ // Block offset broadcasts: each node only sees its own witnesses,
so isDurablyReconciled is
+ // false everywhere and SSTableWriter cannot auto-mark the flushed
sstables as repaired.
+ cluster.filters().verbs(Verb.MT_BROADCAST_LOG_OFFSETS.id).drop();
+
+ for (int i = 0; i < 50; i++)
+ {
+ cluster.coordinator(1)
+ .execute(withKeyspace("INSERT INTO %s.tbl (pk, val)
VALUES (?, ?)"),
+ ConsistencyLevel.QUORUM, i, "v" + i);
+ }
+
+ // Flush and force the active journal segment to roll so we have a
static segment to inspect.
+ cluster.forEach(i -> i.nodetoolResult("flush",
KEYSPACE).asserts().success());
+ cluster.forEach(i -> i.runOnInstance(() ->
MutationJournal.instance().closeCurrentSegmentForTestingIfNonEmpty()));
+
+ // Confirm there is a static segment that the new refcount is
keeping alive, then try to drop:
+ // the dropping pass must be a no-op because every flushed sstable
is unrepaired.
+ cluster.forEach(i -> i.runOnInstance(() -> {
+ int before =
MutationJournal.instance().countStaticSegmentsForTesting();
+ assertTrue("Expected at least one static segment after
flush+segment close, got " + before, before > 0);
+ MutationTrackingService.instance().persistLogStateForTesting();
+ int after =
MutationJournal.instance().countStaticSegmentsForTesting();
+ assertEquals("Static segments must not be dropped while
unrepaired sstables reference them",
+ before, after);
+ }));
+
+ // Restore broadcast, exchange witnesses, and persist so
isDurablyReconciled is now true everywhere.
+ cluster.filters().reset();
+ cluster.forEach(i -> i.runOnInstance(() ->
MutationTrackingService.instance().broadcastOffsetsForTesting()));
+ cluster.forEach(i -> i.runOnInstance(() ->
MutationTrackingService.instance().persistLogStateForTesting()));
+
+ // Major-compact the table. Compaction rewrites the sstable
through SSTableWriter.finalizeMetadata,
+ // which detects that all mutations are durably reconciled and
stamps repairedAt on the output.
+ // SSTableListChangedNotification then releases refs from the
(unrepaired) inputs without acquiring
+ // any from the (repaired) output -> refcount drops to zero.
+ cluster.forEach(i -> i.nodetoolResult("compact", KEYSPACE,
"tbl").asserts().success());
+
+ // Now the persister can drop the static segments.
+ cluster.forEach(i -> i.runOnInstance(() ->
MutationTrackingService.instance().persistLogStateForTesting()));
+ cluster.forEach(i -> i.runOnInstance(() -> {
+ int remaining =
MutationJournal.instance().countStaticSegmentsForTesting();
+ assertEquals("Static segments must be dropped once their
sstables are promoted to repaired",
+ 0, remaining);
+ }));
+ }
+ }
+
+ /**
+ * CASSANDRA-21406 (item 1, tracked -> untracked migration): a keyspace's
unrepaired sstables hold journal
+ * segments while it is tracked, but once it migrates away from tracked
those sstables are never promoted to
+ * repaired (that path is gated on the table being tracked). Their
references must therefore be evicted on
+ * MIGRATE_FROM so the segments can be reclaimed rather than pinned
forever.
+ */
+ @Test(timeout = 120_000)
+ public void testSegmentsReleasedWhenKeyspaceMigratesFromTracked() throws
Throwable
+ {
+ try (Cluster cluster = Cluster.build(3)
+ .withConfig(cfg ->
cfg.with(Feature.NETWORK).with(Feature.GOSSIP))
+ .start())
+ {
+ cluster.schemaChange(withKeyspace(CREATE_KEYSPACE));
+ cluster.schemaChange(String.format(CREATE_TABLE, KEYSPACE));
+
+ cluster.forEach(i -> i.nodetoolResult("disableautocompaction",
KEYSPACE, "tbl").asserts().success());
+
+ for (int i = 0; i < 50; i++)
+ {
+ cluster.coordinator(1)
+ .execute(withKeyspace("INSERT INTO %s.tbl (pk, val)
VALUES (?, ?)"),
+ ConsistencyLevel.QUORUM, i, "v" + i);
+ }
+
+ cluster.forEach(i -> i.nodetoolResult("flush",
KEYSPACE).asserts().success());
+ cluster.forEach(i -> i.runOnInstance(() ->
MutationJournal.instance().closeCurrentSegmentForTestingIfNonEmpty()));
+
+ // Durably reconcile every offset (but keep the flushed sstables
unrepaired, since we never compact),
+ // driving the cross-node offset exchange to convergence. This
satisfies the reconciliation gate (W) so
+ // the *only* thing still pinning the segments is the unrepaired
sstable references (R) -- making the
+ // eviction below the sole reason they can finally drop.
+ boolean reconciled = false;
+ for (int round = 0; round < 30 && !reconciled; round++)
+ {
+ for (int n = 1; n <= cluster.size(); n++)
+ cluster.get(n).runOnInstance(() ->
MutationTrackingService.instance().persistLogStateForTesting());
+ for (int n = 1; n <= cluster.size(); n++)
+ cluster.get(n).runOnInstance(() ->
MutationTrackingService.instance().broadcastOffsetsForTesting());
+ for (int n = 1; n <= cluster.size(); n++)
+ cluster.get(n).runOnInstance(() ->
MutationTrackingService.instance().persistLogStateForTesting());
+
+ reconciled = true;
+ for (int n = 1; n <= cluster.size(); n++)
+ reconciled &= cluster.get(n).callOnInstance(() ->
+
MutationTrackingService.instance().allStaticSegmentsDurablyReconciledForTesting());
+ }
+ assertTrue("Reconciliation must converge so only the sstable
references still pin the segments", reconciled);
+
+ // While tracked, the (fully reconciled but) unrepaired sstables
still hold their segments: a drop pass
+ // is a no-op.
+ cluster.forEach(i -> i.runOnInstance(() -> {
+ int before =
MutationJournal.instance().countStaticSegmentsForTesting();
+ assertTrue("Expected a static segment held by the unrepaired
sstables, got " + before, before > 0);
+ MutationTrackingService.instance().persistLogStateForTesting();
+ assertEquals("Segments must be retained while unrepaired
tracked sstables reference them",
+ before,
MutationJournal.instance().countStaticSegmentsForTesting());
+ }));
+
+ // Migrate the keyspace away from tracked. tracked->untracked is
instant; MIGRATE_FROM evicts the
+ // keyspace's sstable references from the segment tracker on every
node.
+ cluster.schemaChange(withKeyspace("ALTER KEYSPACE %s WITH
replication_type = 'untracked'"));
+
+ // The eviction released the references; a drop pass now reclaims
the segments.
+ cluster.forEach(i -> i.runOnInstance(() ->
MutationTrackingService.instance().persistLogStateForTesting()));
+ cluster.forEach(i -> i.runOnInstance(() -> {
+ int remaining =
MutationJournal.instance().countStaticSegmentsForTesting();
+ assertEquals("Static segments must be dropped after the
keyspace migrates away from tracked",
+ 0, remaining);
+ }));
+ }
+ }
+
+ /**
+ * CASSANDRA-21406 (item 4): under normal operation an unrepaired sstable
can hold its journal segments for an
+ * unbounded time (compaction, where reconciled sstables are born
repaired, may not run for cold tables). When
+ * the journal grows past mutation_tracking.journal_promotion_threshold,
already durably-reconciled unrepaired
+ * sstables are promoted to repaired out of band, releasing their segments
even without a compaction.
+ */
+ @Test(timeout = 120_000)
+ public void testSegmentsReleasedBySizeTriggeredPromotion() throws Throwable
+ {
+ try (Cluster cluster = Cluster.build(3)
+ .withConfig(cfg ->
cfg.with(Feature.NETWORK).with(Feature.GOSSIP)
+
.set("mutation_tracking.journal_promotion_threshold", "1KiB"))
+ .start())
+ {
+ cluster.schemaChange(withKeyspace(CREATE_KEYSPACE));
+ cluster.schemaChange(String.format(CREATE_TABLE, KEYSPACE));
+
+ // Never let compaction promote the sstables to repaired; only the
size-triggered promotion can.
+ cluster.forEach(i -> i.nodetoolResult("disableautocompaction",
KEYSPACE, "tbl").asserts().success());
+
+ for (int i = 0; i < 50; i++)
+ {
+ cluster.coordinator(1)
+ .execute(withKeyspace("INSERT INTO %s.tbl (pk, val)
VALUES (?, ?)"),
+ ConsistencyLevel.QUORUM, i, "v" + i);
+ }
+
+ cluster.forEach(i -> i.nodetoolResult("flush",
KEYSPACE).asserts().success());
+ cluster.forEach(i -> i.runOnInstance(() ->
MutationJournal.instance().closeCurrentSegmentForTestingIfNonEmpty()));
+
+ // The journal exceeds the tiny threshold and the (unrepaired)
flushed sstables are holding segments.
+ cluster.forEach(i -> i.runOnInstance(() -> {
+ long threshold =
DatabaseDescriptor.getMutationTrackingConfig().getJournalPromotionThresholdBytes();
+ assertTrue("promotion threshold must be configured (>0), was "
+ threshold, threshold > 0);
+ assertTrue("journal size must exceed the threshold",
+ MutationJournal.instance().getDiskSpaceUsed() >
threshold);
+ assertTrue("segments must be held by the unrepaired sstables",
+
MutationJournal.instance().countStaticSegmentsForTesting() > 0);
+ }));
+
+ // Reconcile (persist then exchange offsets) and trigger
size-based promotion until the reconciled
+ // sstables are flipped to repaired and their segments reclaimed.
Promotion only flips sstables that are
+ // already durably reconciled, which requires the offset exchange
to have propagated.
+ boolean converged = false;
+ for (int round = 0; round < 30 && !converged; round++)
+ {
+ for (int n = 1; n <= cluster.size(); n++)
+ cluster.get(n).runOnInstance(() ->
MutationTrackingService.instance().persistLogStateForTesting());
+ for (int n = 1; n <= cluster.size(); n++)
+ cluster.get(n).runOnInstance(() ->
MutationTrackingService.instance().broadcastOffsetsForTesting());
+ for (int n = 1; n <= cluster.size(); n++)
+ cluster.get(n).runOnInstance(() -> {
+
MutationTrackingService.instance().maybePromoteReconciledSSTablesForTesting();
+
MutationTrackingService.instance().persistLogStateForTesting();
+ });
+
+ converged = true;
+ for (int n = 1; n <= cluster.size(); n++)
+ converged &= cluster.get(n).callOnInstance(() ->
MutationJournal.instance().countStaticSegmentsForTesting() == 0);
+ }
+
+ cluster.forEach(i -> i.runOnInstance(() -> {
+ int remaining =
MutationJournal.instance().countStaticSegmentsForTesting();
+ assertEquals("Static segments must be dropped after
size-triggered promotion of reconciled sstables",
+ 0, remaining);
+ }));
+ }
+ }
+
+ /**
+ * A transient (witness) replica journals witnessed writes but never
applies them to a memtable, so it never
+ * flushes an sstable for them. Such witness-only journal data must not
pin {@code needsReplay} (only a flush
+ * clears it, so it would otherwise stay set forever) and cannot be
reclaimed by the reference path (there
+ * is no sstable to promote). Its journal segments are instead governed
purely by the reconciliation gate.
+ * This exercises the full transient-replication path end to end on a
{@code '3/1'} keyspace where every
+ * node is a full replica for some ranges and a witness for others: once
every offset is durably
+ * reconciled and the full-replica sstables are compacted to repaired,
every node reclaims all of its static
+ * segments, including those holding only witnessed data.
+ */
+ @Test(timeout = 120_000)
+ public void testWitnessSegmentsReleasedOnceReconciled() throws Throwable
+ {
+ try (Cluster cluster = Cluster.build(3)
+ .withConfig(cfg ->
cfg.with(Feature.NETWORK).with(Feature.GOSSIP)
+
.set("transient_replication_enabled", "true"))
+ .start())
+ {
+ cluster.schemaChange(withKeyspace(
+ "CREATE KEYSPACE %s WITH replication = {'class':
'SimpleStrategy', 'replication_factor': '3/1'} " +
+ "AND replication_type = 'tracked'"));
+ cluster.schemaChange(String.format(CREATE_TABLE, KEYSPACE));
+
+ cluster.forEach(i -> i.nodetoolResult("disableautocompaction",
KEYSPACE, "tbl").asserts().success());
+
+ for (int i = 0; i < 50; i++)
+ {
+ cluster.coordinator(1)
+ .execute(withKeyspace("INSERT INTO %s.tbl (pk, val)
VALUES (?, ?)"),
+ ConsistencyLevel.QUORUM, i, "v" + i);
+ }
+
+ cluster.forEach(i -> i.nodetoolResult("flush",
KEYSPACE).asserts().success());
+ cluster.forEach(i -> i.runOnInstance(() ->
MutationJournal.instance().closeCurrentSegmentForTestingIfNonEmpty()));
+
+ cluster.forEach(i -> i.runOnInstance(() ->
+ assertTrue("Expected static segments after flush+close",
+
MutationJournal.instance().countStaticSegmentsForTesting() > 0)));
+
+ // Reconcile to convergence across full replicas and witnesses.
+ boolean reconciled = false;
+ for (int round = 0; round < 30 && !reconciled; round++)
+ {
+ for (int n = 1; n <= cluster.size(); n++)
+ cluster.get(n).runOnInstance(() ->
MutationTrackingService.instance().persistLogStateForTesting());
+ for (int n = 1; n <= cluster.size(); n++)
+ cluster.get(n).runOnInstance(() ->
MutationTrackingService.instance().broadcastOffsetsForTesting());
+ for (int n = 1; n <= cluster.size(); n++)
+ cluster.get(n).runOnInstance(() ->
MutationTrackingService.instance().persistLogStateForTesting());
+
+ reconciled = true;
+ for (int n = 1; n <= cluster.size(); n++)
+ reconciled &= cluster.get(n).callOnInstance(() ->
+
MutationTrackingService.instance().allStaticSegmentsDurablyReconciledForTesting());
+ }
+ assertTrue("Reconciliation must converge across all replicas and
witnesses", reconciled);
+
+ // Compact so the full-replica sstables are promoted to repaired
and release their segment references.
+ // Witness-only segments have no sstable and no pinned
needsReplay, so the reconciliation above already
+ // made them reclaimable.
+ cluster.forEach(i -> i.nodetoolResult("compact", KEYSPACE,
"tbl").asserts().success());
+
+ cluster.forEach(i -> i.runOnInstance(() ->
MutationTrackingService.instance().persistLogStateForTesting()));
+ cluster.forEach(i -> i.runOnInstance(() ->
+ assertEquals("All static segments (including witness-only
ranges) must be reclaimed once reconciled",
+ 0,
MutationJournal.instance().countStaticSegmentsForTesting())));
+ }
+ }
+}
diff --git
a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingLogPersisterTest.java
b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingLogPersisterTest.java
index 0437aa7514..4fa1fbfb70 100644
---
a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingLogPersisterTest.java
+++
b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingLogPersisterTest.java
@@ -74,6 +74,10 @@ public class MutationTrackingLogPersisterTest extends
FuzzTestBase
.build(schema, hb, cluster)));
}
+ // Keep background compaction from promoting sstables to
repaired before we explicitly compact
+ // below, so the segment-drop timing is deterministic.
+ cluster.forEach(i -> i.nodetoolResult("disableautocompaction",
KEYSPACE).asserts().success());
+
int counter = 0;
for (int pk = 0; pk < pks; pk++)
{
@@ -86,13 +90,22 @@ public class MutationTrackingLogPersisterTest extends
FuzzTestBase
}
cluster.forEach(i -> i.nodetoolResult("flush",
KEYSPACE).asserts().success());
+
cluster.forEach(i -> i.runOnInstance(() ->
MutationTrackingService.instance().persistLogStateForTesting()));
cluster.forEach(i -> i.runOnInstance(() ->
MutationTrackingService.instance().broadcastOffsetsForTesting()));
cluster.forEach(i -> i.runOnInstance(() ->
MutationTrackingService.instance().persistLogStateForTesting()));
+ // The flushed sstables are unrepaired and therefore still
reference their journal segments
+ // (CASSANDRA-21406). Now that their mutations are durably
reconciled, compaction rewrites them into
+ // repaired outputs (SSTableWriter.finalizeMetadata),
releasing the references so the persister can
+ // finally drop the static segments.
+ cluster.forEach(i -> i.nodetoolResult("compact",
KEYSPACE).asserts().success());
+ cluster.forEach(i -> i.runOnInstance(() ->
MutationTrackingService.instance().persistLogStateForTesting()));
+
cluster.forEach(i -> i.runOnInstance(() -> {
int staticSegments =
MutationJournal.instance().countStaticSegmentsForTesting();
- Assert.assertEquals("Expected no static segments after log
persister runs", 0, staticSegments);
+ Assert.assertEquals("Expected no static segments once
reconciled sstables are compacted to repaired",
+ 0, staticSegments);
}));
});
}
diff --git
a/test/unit/org/apache/cassandra/db/virtual/MutationJournalTableTest.java
b/test/unit/org/apache/cassandra/db/virtual/MutationJournalTableTest.java
index 9bfd68e2b9..c9b75a52d7 100644
--- a/test/unit/org/apache/cassandra/db/virtual/MutationJournalTableTest.java
+++ b/test/unit/org/apache/cassandra/db/virtual/MutationJournalTableTest.java
@@ -81,7 +81,8 @@ public class MutationJournalTableTest extends CQLTester
"written_to",
"fsynced_to",
"needs_replay",
- "file_path"
+ "file_path",
+ "referring_sstables"
));
boolean foundSegments = false;
@@ -99,7 +100,8 @@ public class MutationJournalTableTest extends CQLTester
int writtenTo = r.getInt("written_to");
int fsyncedTo = r.getInt("fsynced_to");
r.getBool("needs_replay"); // Just verify it's accessible
- String filePath = r.getString("file_path");
+ r.getString("file_path"); // Just verify it's accessible
+ r.getList("referring_sstables", String.class);// Just verify it's
accessible
assertThat(segmentId).isGreaterThan(0L);
diff --git
a/test/unit/org/apache/cassandra/replication/MutationJournalTest.java
b/test/unit/org/apache/cassandra/replication/MutationJournalTest.java
index 54f69d34f0..b9805b01df 100644
--- a/test/unit/org/apache/cassandra/replication/MutationJournalTest.java
+++ b/test/unit/org/apache/cassandra/replication/MutationJournalTest.java
@@ -22,15 +22,18 @@ import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.util.ArrayList;
import java.util.List;
+import java.util.function.Supplier;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
+import org.mockito.Mockito;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.RowUpdateBuilder;
import org.apache.cassandra.db.marshal.UTF8Type;
+import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.journal.Descriptor;
@@ -56,6 +59,7 @@ public class MutationJournalTest
private static final String KEYSPACE = "mjtks";
private static final String TABLE = "mjtt";
+ private static TestDurablyReconciledOffsetsSupplier
durablyReconciledOffsetsSupplier;
private static MutationJournal journal;
private static File directory;
@@ -73,7 +77,9 @@ public class MutationJournalTest
directory = new
File(Files.createTempDirectory("mutation-journal-test-simple"));
directory.deleteRecursiveOnExit();
- journal = new MutationJournal(directory, TestParams.MUTATION_JOURNAL);
+ durablyReconciledOffsetsSupplier = new
TestDurablyReconciledOffsetsSupplier();
+
+ journal = new MutationJournal(directory, TestParams.MUTATION_JOURNAL,
durablyReconciledOffsetsSupplier);
journal.startInternal();
}
@@ -181,19 +187,25 @@ public class MutationJournalTest
}
@Test
- public void testDropReconcileSegments()
+ public void testDropSegments()
{
- ShortMutationId id1 = id(100L, 0);
- ShortMutationId id2 = id(100L, 1);
- ShortMutationId id3 = id(200L, 2);
- ShortMutationId id4 = id(200L, 3);
+ // Distinct logIds from the read tests (which use logId 100) so writes
to the shared journal don't collide.
+ ShortMutationId id1 = id(1000L, 0);
+ ShortMutationId id2 = id(1000L, 1);
+ ShortMutationId id3 = id(2000L, 2);
+ ShortMutationId id4 = id(2000L, 3);
Mutation mutation1 = mutation("key1", "ck1", "value1");
Mutation mutation2 = mutation("key2", "ck2", "value2");
Mutation mutation3 = mutation("key3", "ck3", "value3");
Mutation mutation4 = mutation("key4", "ck4", "value4");
+ SegmentReferenceTracker refs = journal.segmentReferenceTracker();
+
+ int baseline = journal.countStaticSegmentsForTesting();
+
// write two mutations to the first segment and flush it to make static
+ long firstSegment = journal.getCurrentPosition().segmentId;
journal.write(id1, mutation1);
journal.write(id2, mutation2);
journal.closeCurrentSegmentForTestingIfNonEmpty();
@@ -203,51 +215,96 @@ public class MutationJournalTest
journal.write(id4, mutation4);
journal.closeCurrentSegmentForTestingIfNonEmpty();
+ // durably-reconciled offsets covering every mutation written above
(logId 1000 -> {0,1}, 2000 -> {2,3})
+ Log2OffsetsMap.Mutable allReconciled = new Log2OffsetsMap.Mutable();
+ for (ShortMutationId id : List.of(id1, id2, id3, id4))
+ allReconciled.add(id);
+
{
- // call dropReconciledSegments() with a log2offsets map that
covers both segments fully
- // *BUT* with the segments still marked as needing replay nothing
should be dropped
- Log2OffsetsMap.Immutable.Builder builder = new
Log2OffsetsMap.Immutable.Builder();
- builder.add(id1);
- builder.add(id2);
- builder.add(id3);
- builder.add(id4);
- assertEquals(0, journal.dropReconciledSegments(builder.build()));
- // confirm that no static segments have been dropped
- assertEquals(2, journal.countStaticSegmentsForTesting());
+ // Both segments still need replay; even fully reconciled and
unreferenced they must be retained.
+
durablyReconciledOffsetsSupplier.setDurablyReconciledOffsetsSupplierForTesting(()
-> allReconciled);
+ journal.runCompactionBlocking();
+ assertEquals(baseline + 2,
journal.countStaticSegmentsForTesting());
}
- // mark both segments as not needing replay
+ // mark both segments as not needing replay (simulate their memtables
having been flushed)
journal.clearNeedsReplayForTesting();
{
- // call dropReconciledSegments() with a log2offsets map that
doesn't cover any segments fully
- Log2OffsetsMap.Immutable.Builder builder = new
Log2OffsetsMap.Immutable.Builder();
- builder.add(id1);
- assertEquals(0, journal.dropReconciledSegments(builder.build()));
- // confirm that no static segments got dropped
- assertEquals(2, journal.countStaticSegmentsForTesting());
+ // Not reconciled -> retained even when !needsReplay and
unreferenced (the witness gate).
+
durablyReconciledOffsetsSupplier.setDurablyReconciledOffsetsSupplierForTesting(Log2OffsetsMap.Mutable::new);
+ journal.runCompactionBlocking();
+ assertEquals(baseline + 2,
journal.countStaticSegmentsForTesting());
+ }
+
+
durablyReconciledOffsetsSupplier.setDurablyReconciledOffsetsSupplierForTesting(()
-> allReconciled);
+
+ {
+ // An unrepaired sstable referencing the first segment retains it;
the second (unreferenced,
+ // reconciled, !needsReplay) is dropped.
+ SSTableReader referrer = Mockito.mock(SSTableReader.class);
+ refs.addReferenceForTesting(firstSegment, referrer);
+ journal.runCompactionBlocking();
+ assertEquals(baseline + 1,
journal.countStaticSegmentsForTesting());
+
+ // Releasing the last reference allows the first segment to drop
too.
+ refs.removeReferenceForTesting(firstSegment, referrer);
+ journal.runCompactionBlocking();
+ assertEquals(baseline, journal.countStaticSegmentsForTesting());
}
+ }
+
+ @Test
+ public void testWitnessOnlyWritesDoNotPinNeedsReplay()
+ {
+ // Isolate a fresh segment holding only the witness-only writes below.
+ journal.closeCurrentSegmentForTestingIfNonEmpty();
+ long witnessSegment = journal.getCurrentPosition().segmentId;
+
+ // Witnessed-only writes (fullReplica=false): journaled but never
applied to a memtable, so never marked
+ // dirty. A segment holding only such data has nothing to flush.
+ journal.write(id(300L, 0), mutation("wk1", "ck", "v1"), false);
+ journal.write(id(300L, 1), mutation("wk2", "ck", "v2"), false);
+ journal.closeCurrentSegmentForTestingIfNonEmpty();
+
+ assertTrue("witness-only segment should be eligible to clear
needsReplay without a flush to prevent " +
+ "the journal to grow unbounded on witness-only nodes",
+
journal.pendingCleanupForTesting().contains(witnessSegment));
+ }
+
+ @Test
+ public void testFullReplicaWritesPinNeedsReplayUntilFlushed()
+ {
+ // Isolate a fresh segment holding only the full-replica write below.
+ journal.closeCurrentSegmentForTestingIfNonEmpty();
+ long fullSegment = journal.getCurrentPosition().segmentId;
+
+ // Full-replica write marks the segment dirty; without a flush it
still needs replay.
+ journal.write(id(400L, 0), mutation("fk1", "ck", "v1"), /* fullReplica
= */ true);
+ journal.closeCurrentSegmentForTestingIfNonEmpty();
+
+ assertFalse("full-replica segment must not clear needsReplay before
its memtable is flushed",
+ journal.pendingCleanupForTesting().contains(fullSegment));
+ }
+
+ private static class TestDurablyReconciledOffsetsSupplier implements
Supplier<Log2OffsetsMap<?>>
+ {
+ private Supplier<Log2OffsetsMap<?>> nonDefaultSupplier;
+ @Override
+ public Log2OffsetsMap<?> get()
{
- // call dropReconciledSegments() with a log2offsets map that
covers only the first segment fully
- Log2OffsetsMap.Immutable.Builder builder = new
Log2OffsetsMap.Immutable.Builder();
- builder.add(id1);
- builder.add(id2);
- assertEquals(1, journal.dropReconciledSegments(builder.build()));
- // confirm that only one static segment got dropped
- assertEquals(1, journal.countStaticSegmentsForTesting());
+ if (nonDefaultSupplier != null)
+ return nonDefaultSupplier.get();
+ Log2OffsetsMap.Mutable durablyReconciled = new
Log2OffsetsMap.Mutable();
+ if (MutationTrackingService.isEnabled())
+
MutationTrackingService.instance().collectDurablyReconciledOffsets(durablyReconciled);
+ return durablyReconciled;
}
+ public void
setDurablyReconciledOffsetsSupplierForTesting(Supplier<Log2OffsetsMap<?>>
nonDefaultSupplier)
{
- // call dropReconciledSegments() with a log2offsets map that
covers both segments fully
- Log2OffsetsMap.Immutable.Builder builder = new
Log2OffsetsMap.Immutable.Builder();
- builder.add(id1);
- builder.add(id2);
- builder.add(id3);
- builder.add(id4);
- assertEquals(1, journal.dropReconciledSegments(builder.build()));
- // confirm that all static segments have now been dropped
- assertEquals(0, journal.countStaticSegmentsForTesting());
+ this.nonDefaultSupplier = nonDefaultSupplier;
}
}
diff --git
a/test/unit/org/apache/cassandra/replication/SegmentReferenceTrackerTest.java
b/test/unit/org/apache/cassandra/replication/SegmentReferenceTrackerTest.java
new file mode 100644
index 0000000000..c1504ae205
--- /dev/null
+++
b/test/unit/org/apache/cassandra/replication/SegmentReferenceTrackerTest.java
@@ -0,0 +1,517 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.replication;
+
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BooleanSupplier;
+
+import org.junit.Test;
+import org.mockito.Mockito;
+
+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.notifications.InitialSSTableAddedNotification;
+import org.apache.cassandra.notifications.SSTableAddedNotification;
+import org.apache.cassandra.notifications.SSTableListChangedNotification;
+import org.apache.cassandra.notifications.SSTableRepairStatusChanged;
+import org.apache.cassandra.schema.ReplicationType;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.service.ActiveRepairService;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.EstimatedHistogram;
+import org.apache.cassandra.utils.streamhist.TombstoneHistogram;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class SegmentReferenceTrackerTest
+{
+ private static final UUID LOCAL_HOST = UUID.randomUUID();
+ private static final UUID REMOTE_HOST = UUID.randomUUID();
+
+ @Test
+ public void testInitialAddRefsEverySegmentInTheInterval()
+ {
+ int startSegment = 5;
+ int endSegment = 7;
+ SegmentReferenceTracker tracker = newTracker();
+ SSTableReader sstable = unrepaired(intervals(startSegment, 0,
endSegment, 100));
+
+ tracker.handleNotification(new
InitialSSTableAddedNotification(List.of(sstable)), null);
+
+ // Coarse range covers segments 5..7 inclusive.
+ for (long segment = startSegment; segment <= endSegment; segment++)
+ assertEquals("segment " + segment, 1L,
tracker.referenceCountForTesting(segment));
+ assertFalse(tracker.isReferenced(4));
+ assertFalse(tracker.isReferenced(8));
+ assertEquals(1, tracker.trackedSstableCountForTesting());
+ }
+
+ @Test
+ public void testAddedRepairedSSTableHoldsNoRefs()
+ {
+ int startSegment = 5;
+ int endSegment = 7;
+ SegmentReferenceTracker tracker = newTracker();
+ SSTableReader sstable = repaired(intervals(startSegment, 0,
endSegment, 100));
+
+ tracker.handleNotification(new
SSTableAddedNotification(List.of(sstable), null), null);
+
+ for (long segment = startSegment; segment <= endSegment; segment++)
+ assertFalse("segment " + segment, tracker.isReferenced(segment));
+ assertEquals(0, tracker.trackedSstableCountForTesting());
+ }
+
+ @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));
+
+ tracker.handleNotification(new
SSTableAddedNotification(List.of(sstable), null), null);
+
+ for (long segment = 5; segment <= 7; segment++)
+ assertFalse("segment " + segment, tracker.isReferenced(segment));
+ 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);
+
+ tracker.handleNotification(new
SSTableAddedNotification(List.of(sstable), null), null);
+
+ for (long segment = 5; segment <= 7; segment++)
+ assertFalse("segment " + segment, tracker.isReferenced(segment));
+ 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);
+
+ tracker.handleNotification(new
SSTableAddedNotification(List.of(sstable), null), null);
+
+ for (long segment = 5; segment <= 7; segment++)
+ assertFalse("segment " + segment, tracker.isReferenced(segment));
+ assertEquals(0, tracker.trackedSstableCountForTesting());
+ }
+
+ @Test
+ public void testEvictReleasesReferencesAndFiresCallback()
+ {
+ int[] calls = { 0 };
+ SegmentReferenceTracker tracker = newTracker(() -> calls[0]++);
+ SSTableReader sstable = unrepaired(intervals(5, 0, 7, 0));
+
+ 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++)
+ assertFalse("segment " + segment, tracker.isReferenced(segment));
+ assertEquals(0, tracker.trackedSstableCountForTesting());
+ assertEquals("evicting the last referrer fires the unreferenced
callback", 1, calls[0]);
+ }
+
+ @Test
+ public void testEvictOnlyReleasesGivenSstables()
+ {
+ SegmentReferenceTracker tracker = newTracker();
+ SSTableReader a = unrepaired(intervals(5, 0, 5, 100));
+ SSTableReader b = unrepaired(intervals(5, 0, 5, 200));
+
+ 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));
+
+ tracker.evict(List.of(b));
+ assertFalse(tracker.isReferenced(5));
+ assertEquals(0, tracker.trackedSstableCountForTesting());
+ }
+
+ @Test
+ public void testAddIsIdempotent()
+ {
+ int startSegment = 5;
+ int endSegment = 7;
+ SegmentReferenceTracker tracker = newTracker();
+ SSTableReader sstable = unrepaired(intervals(startSegment, 0,
endSegment, 100));
+
+ tracker.handleNotification(new
SSTableAddedNotification(List.of(sstable), null), null);
+ tracker.handleNotification(new
SSTableAddedNotification(List.of(sstable), null), null);
+
+ assertEquals(1L, tracker.referenceCountForTesting(5));
+ assertEquals(1, tracker.trackedSstableCountForTesting());
+ }
+
+ @Test
+ public void testMultipleDisjointIntervalsRefEachContainedSegment()
+ {
+ SegmentReferenceTracker tracker = newTracker();
+ // Two disjoint intervals: [3:0..3:100] and [9:0..10:50].
+ IntervalSet.Builder<CommitLogPosition> 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());
+
+ 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));
+ }
+
+ @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<CommitLogPosition> 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());
+ }
+
+ @Test
+ public void testEmptyIntervalsHoldNoRefs()
+ {
+ SegmentReferenceTracker tracker = newTracker();
+ SSTableReader sstable = unrepaired(IntervalSet.empty());
+
+ tracker.handleNotification(new
SSTableAddedNotification(List.of(sstable), null), null);
+
+ // Tracked sstable but no segments to ref.
+ assertEquals(1, tracker.trackedSstableCountForTesting());
+ assertFalse(tracker.isReferenced(0));
+ }
+
+ @Test
+ public void
testCompactionPreservesRefsWhenInputAndOutputOverlapSameSegment()
+ {
+ SegmentReferenceTracker tracker = newTracker();
+ SSTableReader input = unrepaired(intervals(5, 0, 5, 100));
+ SSTableReader output = unrepaired(intervals(5, 0, 5, 200));
+
+ 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);
+
+ // Net: still one unrepaired sstable referencing segment 5.
+ assertEquals(1L, tracker.referenceCountForTesting(5));
+ assertEquals(1, tracker.trackedSstableCountForTesting());
+ }
+
+ @Test
+ public void testCompactionToRepairedOutputReleasesRefs()
+ {
+ SegmentReferenceTracker tracker = newTracker();
+ SSTableReader input = unrepaired(intervals(5, 0, 5, 100));
+ SSTableReader output = repaired(intervals(5, 0, 5, 200));
+
+ 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);
+
+ assertFalse(tracker.isReferenced(5));
+ assertEquals(0, tracker.trackedSstableCountForTesting());
+ }
+
+ @Test
+ public void testRepairPromotionReleasesRefs()
+ {
+ SegmentReferenceTracker tracker = newTracker();
+ AtomicReference<Boolean> repaired = new AtomicReference<>(false);
+ SSTableReader sstable = sstableWithRepairSupplier(intervals(5, 0, 7,
0), repaired::get);
+
+ 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.
+ 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 testRepairStatusFlippingBackToUnrepairedReAcquires()
+ {
+ SegmentReferenceTracker tracker = newTracker();
+ AtomicReference<Boolean> repaired = new AtomicReference<>(true);
+ SSTableReader sstable = sstableWithRepairSupplier(intervals(5, 0, 5,
100), 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);
+
+ assertEquals(1L, tracker.referenceCountForTesting(5));
+ assertEquals(1, tracker.trackedSstableCountForTesting());
+ }
+
+ @Test
+ public void testMultipleSstablesAccumulateRefsOnSharedSegments()
+ {
+ SegmentReferenceTracker tracker = newTracker();
+ SSTableReader a = unrepaired(intervals(5, 0, 6, 0));
+ SSTableReader b = unrepaired(intervals(6, 0, 7, 0));
+
+ tracker.handleNotification(new SSTableAddedNotification(List.of(a, b),
null), null);
+
+ assertEquals(1L, tracker.referenceCountForTesting(5));
+ assertEquals(2L, tracker.referenceCountForTesting(6));
+ assertEquals(1L, tracker.referenceCountForTesting(7));
+
+ 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(1L, tracker.referenceCountForTesting(7));
+ }
+
+ @Test
+ public void testReleaseOfUntrackedSstableIsNoOp()
+ {
+ SegmentReferenceTracker tracker = newTracker();
+ SSTableReader sstable = unrepaired(intervals(5, 0, 5, 100));
+
+ assertFalse(tracker.isReferenced(5));
+
+ 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));
+ }
+
+ @Test
+ public void testCallbackFiredWhenLastReferenceReleased()
+ {
+ int[] calls = { 0 };
+ SegmentReferenceTracker tracker = newTracker(() -> calls[0]++);
+ SSTableReader sstable = unrepaired(intervals(5, 0, 7, 0));
+
+ tracker.handleNotification(new
SSTableAddedNotification(List.of(sstable), null), null);
+ assertEquals("adding never fires 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);
+
+ assertFalse(tracker.isReferenced(5));
+ assertEquals("releasing the last reference fires the unreferenced
callback", 1, calls[0]);
+ }
+
+ @Test
+ 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));
+
+ 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);
+
+ assertTrue(tracker.isReferenced(5));
+ assertEquals("callback must not fire while the segment is still
referenced", 0, calls[0]);
+ }
+
+ // -- helpers ---------------------------------------------------------
+
+ private static SegmentReferenceTracker newTracker()
+ {
+ return newTracker(() -> {});
+ }
+
+ private static SegmentReferenceTracker newTracker(Runnable
onSegmentsUnreferenced)
+ {
+ // Fixed local host id so the mock sstables (originating from
LOCAL_HOST) are treated as locally originated.
+ return new SegmentReferenceTracker(onSegmentsUnreferenced, () ->
LOCAL_HOST);
+ }
+
+ private static IntervalSet<CommitLogPosition> intervals(long startSegment,
int startPosition, long endSegment, int endPosition)
+ {
+ 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));
+ }
+
+ private static SSTableReader unrepaired(IntervalSet<CommitLogPosition>
intervals)
+ {
+ return stub(intervals, false);
+ }
+
+ private static SSTableReader repaired(IntervalSet<CommitLogPosition>
intervals)
+ {
+ return stub(intervals, true);
+ }
+
+ private static SSTableReader stub(IntervalSet<CommitLogPosition>
intervals, boolean isRepaired)
+ {
+ return sstableWithRepairSupplier(intervals, () -> isRepaired);
+ }
+
+ private static SSTableReader
sstableWithRepairSupplier(IntervalSet<CommitLogPosition> intervals,
+ BooleanSupplier
isRepairedSupplier)
+ {
+ return sstable(intervals, isRepairedSupplier,
coordinatorLogOffsets(true));
+ }
+
+ private static SSTableReader sstable(IntervalSet<CommitLogPosition>
intervals,
+ BooleanSupplier isRepairedSupplier,
+ ImmutableCoordinatorLogOffsets
coordinatorLogOffsets)
+ {
+ return sstable(intervals, isRepairedSupplier, coordinatorLogOffsets,
LOCAL_HOST);
+ }
+
+ private static SSTableReader sstable(IntervalSet<CommitLogPosition>
intervals,
+ BooleanSupplier isRepairedSupplier,
+ ImmutableCoordinatorLogOffsets
coordinatorLogOffsets,
+ UUID originatingHostId)
+ {
+ return sstable(intervals, isRepairedSupplier, coordinatorLogOffsets,
originatingHostId, ReplicationType.tracked);
+ }
+
+ private static SSTableReader sstable(IntervalSet<CommitLogPosition>
intervals,
+ BooleanSupplier isRepairedSupplier,
+ ImmutableCoordinatorLogOffsets
coordinatorLogOffsets,
+ UUID originatingHostId,
+ ReplicationType replicationType)
+ {
+ 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);
+ TableMetadata tableMetadata = Mockito.mock(TableMetadata.class);
+
Mockito.when(tableMetadata.replicationType()).thenReturn(replicationType);
+ Mockito.when(reader.metadata()).thenReturn(tableMetadata);
+ return reader;
+ }
+
+ 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;
+ }
+
+ private static StatsMetadata stats(IntervalSet<CommitLogPosition>
intervals, 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
+ TombstoneHistogram.createDefault(),
+ 0,
// sstableLevel
+ List.of(),
// clusteringTypes
+ Slice.ALL,
// coveredClustering
+ false,
// hasLegacyCounterShards
+ ActiveRepairService.UNREPAIRED_SSTABLE,
+ 0L,
// totalColumnsSet
+ 0L,
// totalRows
+ Double.NaN,
// tokenSpaceCoverage
+ originatingHostId,
+ ActiveRepairService.NO_PENDING_REPAIR,
+ false,
// hasPartitionLevelDeletions
+ ImmutableCoordinatorLogOffsets.NONE,
+ ByteBufferUtil.EMPTY_BYTE_BUFFER,
+ ByteBufferUtil.EMPTY_BYTE_BUFFER);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
