This is an automated email from the ASF dual-hosted git repository.
bdeggleston 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 d8ee241e87 CEP-45: Handle tombstone purging correctly in preview
repair, fix validation timeouts, and add trace logging
d8ee241e87 is described below
commit d8ee241e87f8678fb602829a75f3a92b22777320
Author: Blake Eggleston <[email protected]>
AuthorDate: Tue Nov 3 13:23:18 2026 -0800
CEP-45: Handle tombstone purging correctly in preview repair, fix
validation timeouts, and add trace logging
Patch by Blake Eggleston; reviewed by Caleb Rackliffe for CASSANDRA-21850
---
src/java/org/apache/cassandra/config/Config.java | 2 -
.../cassandra/config/DatabaseDescriptor.java | 10 -
.../db/repair/CassandraTableRepairManager.java | 4 +-
.../db/repair/CassandraValidationIterator.java | 38 +++-
.../repair/JournalMergingValidationIterator.java | 52 ++++-
.../repair/MutationTrackingPreviewRepairTask.java | 35 +++-
.../MutationTrackingValidationCoordinator.java | 11 +-
.../cassandra/replication/ActivatedTransfers.java | 34 ++-
.../ImmutableCoordinatorLogOffsets.java | 30 ++-
.../org/apache/cassandra/replication/Offsets.java | 24 +++
.../cassandra/replication/ValidationOffsets.java | 28 ++-
.../apache/cassandra/service/StorageService.java | 12 --
.../cassandra/service/StorageServiceMBean.java | 3 -
.../test/tracking/TrackedPreviewRepairTest.java | 17 +-
.../JournalMergingValidationIteratorTest.java | 228 +++++++++++++++++++++
15 files changed, 451 insertions(+), 77 deletions(-)
diff --git a/src/java/org/apache/cassandra/config/Config.java
b/src/java/org/apache/cassandra/config/Config.java
index 527f75c79a..614691056b 100644
--- a/src/java/org/apache/cassandra/config/Config.java
+++ b/src/java/org/apache/cassandra/config/Config.java
@@ -175,8 +175,6 @@ public class Config
public volatile DurationSpec.LongMillisecondsBound
mutation_tracking_sync_timeout = new DurationSpec.LongMillisecondsBound("2m");
- public volatile DurationSpec.LongMillisecondsBound
mutation_tracking_validation_timeout = new
DurationSpec.LongMillisecondsBound("2m");
-
public Integer streaming_connections_per_host = 1;
@Replaces(oldName = "streaming_keep_alive_period_in_secs", converter =
Converters.SECONDS_DURATION, deprecated = true)
public DurationSpec.IntSecondsBound streaming_keep_alive_period = new
DurationSpec.IntSecondsBound("300s");
diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index f88406a388..86ca89b02c 100644
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@ -2695,16 +2695,6 @@ public class DatabaseDescriptor
conf.mutation_tracking_sync_timeout = new
DurationSpec.LongMillisecondsBound(timeoutInMillis);
}
- public static long getMutationTrackingValidationTimeout(TimeUnit unit)
- {
- return conf.mutation_tracking_validation_timeout.to(unit);
- }
-
- public static void setMutationTrackingValidationTimeout(long
timeoutInMillis)
- {
- conf.mutation_tracking_validation_timeout = new
DurationSpec.LongMillisecondsBound(timeoutInMillis);
- }
-
public static boolean hasCrossNodeTimeout()
{
return conf.internode_timeout;
diff --git
a/src/java/org/apache/cassandra/db/repair/CassandraTableRepairManager.java
b/src/java/org/apache/cassandra/db/repair/CassandraTableRepairManager.java
index e4232071f8..21c2dee9af 100644
--- a/src/java/org/apache/cassandra/db/repair/CassandraTableRepairManager.java
+++ b/src/java/org/apache/cassandra/db/repair/CassandraTableRepairManager.java
@@ -62,14 +62,14 @@ public class CassandraTableRepairManager implements
TableRepairManager
public ValidationPartitionIterator
getValidationIterator(Collection<Range<Token>> ranges, TimeUUID parentId,
TimeUUID sessionID, boolean isIncremental, long nowInSec, boolean
dontPurgeTombstones, ValidationOffsets validationOffsets,
TopPartitionTracker.Collector topPartitionCollector) throws IOException,
NoSuchRepairSessionException
{
CassandraValidationIterator sstableIterator =
- new CassandraValidationIterator(cfs, ctx, ranges, parentId,
sessionID, isIncremental, nowInSec, dontPurgeTombstones, validationOffsets,
topPartitionCollector);
+ new CassandraValidationIterator(cfs, ctx, ranges, parentId,
sessionID, isIncremental, nowInSec, validationOffsets != null ||
dontPurgeTombstones, validationOffsets, topPartitionCollector);
if (validationOffsets == null)
return sstableIterator;
try
{
- return new JournalMergingValidationIterator(cfs, sstableIterator,
ranges, validationOffsets);
+ return new JournalMergingValidationIterator(cfs, sstableIterator,
ranges, validationOffsets, nowInSec, dontPurgeTombstones);
}
catch (Throwable t)
{
diff --git
a/src/java/org/apache/cassandra/db/repair/CassandraValidationIterator.java
b/src/java/org/apache/cassandra/db/repair/CassandraValidationIterator.java
index e09269c70f..9489254828 100644
--- a/src/java/org/apache/cassandra/db/repair/CassandraValidationIterator.java
+++ b/src/java/org/apache/cassandra/db/repair/CassandraValidationIterator.java
@@ -61,6 +61,7 @@ import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.ActiveRepairService;
import org.apache.cassandra.service.snapshot.SnapshotManager;
import org.apache.cassandra.service.snapshot.TableSnapshot;
+import org.apache.cassandra.streaming.PreviewKind;
import org.apache.cassandra.utils.TimeUUID;
import org.apache.cassandra.utils.concurrent.Refs;
@@ -160,10 +161,15 @@ public class CassandraValidationIterator extends
ValidationPartitionIterator
{
for (SSTableReader sstable : sstableCandidates.sstables)
{
- if (new Bounds<>(sstable.getFirst().getToken(),
sstable.getLast().getToken()).intersects(ranges) && predicate.apply(sstable))
- {
+ boolean intersects = new
Bounds<>(sstable.getFirst().getToken(),
sstable.getLast().getToken()).intersects(ranges);
+ boolean selected = intersects && predicate.apply(sstable);
+ if (selected)
sstablesToValidate.add(sstable);
- }
+
+ if (validationOffsets != null && logger.isTraceEnabled())
+ logger.trace("{} Validation selection for {}: {} - {}",
+ prs.previewKind.logPrefix(parentId), sstable,
selected ? "selected" : "skipped",
+ selectionReason(sstable, intersects,
selected, validationOffsets));
}
sstables = Refs.tryRef(sstablesToValidate);
@@ -177,7 +183,31 @@ public class CassandraValidationIterator extends
ValidationPartitionIterator
return sstables;
}
+ private static String selectionReason(SSTableReader sstable, boolean
intersects, boolean selected, ValidationOffsets validationOffsets)
+ {
+ String offsets = sstable.getCoordinatorLogOffsets().toString(true);
+ if (!intersects)
+ return "out of range, offsets=" + offsets;
+ if (sstable.isRepaired())
+ return (selected ? "" : "not selected despite being repaired, ") +
"repaired, no offset check, offsets=" + offsets;
+ if (selected)
+ return "unrepaired, offsets covered, offsets=" + offsets;
+ return "unrepaired, offsets not covered by " +
validationOffsets.toString(true) + ", offsets=" + offsets;
+ }
+
+ TimeUUID parentId()
+ {
+ return parentId;
+ }
+
+ PreviewKind previewKind()
+ {
+ return previewKind;
+ }
+
private final ColumnFamilyStore cfs;
+ private final TimeUUID parentId;
+ private final PreviewKind previewKind;
private final Refs<SSTableReader> sstables;
private final String snapshotName;
private final boolean isGlobalSnapshotValidation;
@@ -195,6 +225,7 @@ public class CassandraValidationIterator extends
ValidationPartitionIterator
public CassandraValidationIterator(ColumnFamilyStore cfs, SharedContext
ctx, Collection<Range<Token>> ranges, TimeUUID parentId, TimeUUID sessionID,
boolean isIncremental, long nowInSec, boolean dontPurgeTombstones,
ValidationOffsets validationOffsets, TopPartitionTracker.Collector
topPartitionCollector) throws IOException, NoSuchRepairSessionException
{
this.cfs = cfs;
+ this.parentId = parentId;
isGlobalSnapshotValidation =
SnapshotManager.instance.exists(cfs.getKeyspaceName(), cfs.getTableName(),
parentId.toString());
if (isGlobalSnapshotValidation)
@@ -234,6 +265,7 @@ public class CassandraValidationIterator extends
ValidationPartitionIterator
Preconditions.checkArgument(sstables != null);
ActiveRepairService.ParentRepairSession prs =
ctx.repair().getParentRepairSession(parentId);
+ previewKind = prs.previewKind;
logger.info("{}, parentSessionId={}: Performing validation compaction
on {} sstables in {}.{}",
prs.previewKind.logPrefix(sessionID),
parentId,
diff --git
a/src/java/org/apache/cassandra/db/repair/JournalMergingValidationIterator.java
b/src/java/org/apache/cassandra/db/repair/JournalMergingValidationIterator.java
index a2d6c537b9..c9877fcda6 100644
---
a/src/java/org/apache/cassandra/db/repair/JournalMergingValidationIterator.java
+++
b/src/java/org/apache/cassandra/db/repair/JournalMergingValidationIterator.java
@@ -28,6 +28,7 @@ import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.TreeMap;
+import java.util.function.LongPredicate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -38,21 +39,28 @@ import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.RegularAndStaticColumns;
import org.apache.cassandra.db.partitions.AbstractUnfilteredPartitionIterator;
import org.apache.cassandra.db.partitions.PartitionUpdate;
+import org.apache.cassandra.db.partitions.PurgeFunction;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
import org.apache.cassandra.db.rows.DeserializationHelper;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.db.rows.UnfilteredRowIterators;
import org.apache.cassandra.db.rows.WrappingUnfilteredRowIterator;
+import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.repair.ValidationPartitionIterator;
+import org.apache.cassandra.replication.ImmutableCoordinatorLogOffsets;
import org.apache.cassandra.replication.MutationJournal;
import org.apache.cassandra.replication.ShortMutationId;
import org.apache.cassandra.replication.ValidationOffsets;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.streaming.PreviewKind;
+import org.apache.cassandra.utils.TimeUUID;
+
+import static
org.apache.cassandra.db.repair.CassandraValidationIterator.getDefaultGcBefore;
/**
* Merges a {@link CassandraValidationIterator}'s SSTable-sourced partitions
with
@@ -73,14 +81,27 @@ public class JournalMergingValidationIterator extends
ValidationPartitionIterato
public JournalMergingValidationIterator(ColumnFamilyStore cfs,
CassandraValidationIterator
sstableIterator,
Collection<Range<Token>> ranges,
- ValidationOffsets
validationOffsets)
+ ValidationOffsets
validationOffsets,
+ long nowInSec,
+ boolean dontPurgeTombstones)
{
this.sstableIterator = sstableIterator;
// Narrow the offsets for the journal to exclude anything the selected
SSTables already cover.
// The journal isn't pruned on flush, so without this, we could merge
the same data from both sources.
- ValidationOffsets journalOffsets =
validationOffsets.without(sstableIterator.sstableCoordinatorLogOffsets());
+ ImmutableCoordinatorLogOffsets sstableOffsets =
sstableIterator.sstableCoordinatorLogOffsets();
+ ValidationOffsets journalOffsets =
validationOffsets.without(sstableOffsets);
- JournalPartitionAccessor journalAccessor =
JournalPartitionAccessor.build(cfs, ranges, journalOffsets);
+ if (logger.isTraceEnabled())
+ {
+ logger.trace("{} Journal validtion iterator segments selected for
{}.{}: agreed={}, sstableOffsets={}, journalRequest={}",
+
sstableIterator.previewKind().logPrefix(sstableIterator.parentId()),
+ cfs.getKeyspaceName(), cfs.getTableName(),
+ validationOffsets.toString(true),
sstableOffsets.toString(true), journalOffsets.toString(true));
+ }
+
+ JournalPartitionAccessor journalAccessor =
JournalPartitionAccessor.build(cfs, ranges, journalOffsets,
+
sstableIterator.previewKind(),
+
sstableIterator.parentId());
try
{
JournalPartitionIterator journalIterator = new
JournalPartitionIterator(journalAccessor);
@@ -90,7 +111,18 @@ public class JournalMergingValidationIterator extends
ValidationPartitionIterato
this.rangePartitionCounts = new
HashMap<>(sstableIterator.getRangePartitionCounts());
for (Map.Entry<Range<Token>, Long> entry :
journalAccessor.rangeCounts.entrySet())
rangePartitionCounts.merge(entry.getKey(), entry.getValue(),
Long::sum);
- this.merged =
UnfilteredPartitionIterators.merge(List.of(sstableIterator, journalIterator),
UnfilteredPartitionIterators.MergeListener.NOOP);
+
+ UnfilteredPartitionIterator iter =
UnfilteredPartitionIterators.merge(List.of(sstableIterator, journalIterator),
UnfilteredPartitionIterators.MergeListener.NOOP);
+ long gcBefore = dontPurgeTombstones ? Long.MIN_VALUE :
getDefaultGcBefore(cfs, nowInSec);
+ PurgeFunction purger = new PurgeFunction(nowInSec, gcBefore,
Long.MIN_VALUE, false, cfs.metadata().enforceStrictLiveness())
+ {
+ @Override
+ protected LongPredicate getPurgeEvaluator()
+ {
+ return time -> true;
+ }
+ };
+ this.merged = Transformation.apply(iter, purger);
}
catch (Throwable t)
{
@@ -170,12 +202,12 @@ public class JournalMergingValidationIterator extends
ValidationPartitionIterato
this.snapshot = snapshot;
}
- static JournalPartitionAccessor build(ColumnFamilyStore cfs,
Collection<Range<Token>> ranges, ValidationOffsets validationOffsets)
+ static JournalPartitionAccessor build(ColumnFamilyStore cfs,
Collection<Range<Token>> ranges, ValidationOffsets validationOffsets,
PreviewKind previewKind, TimeUUID parentId)
{
MutationJournal.Snapshot snapshot =
MutationJournal.instance().snapshot(validationOffsets);
try
{
- return buildFromSnapshot(snapshot, cfs, ranges,
validationOffsets);
+ return buildFromSnapshot(snapshot, cfs, ranges,
validationOffsets, previewKind, parentId);
}
catch (Throwable t)
{
@@ -184,7 +216,7 @@ public class JournalMergingValidationIterator extends
ValidationPartitionIterato
}
}
- private static JournalPartitionAccessor
buildFromSnapshot(MutationJournal.Snapshot snapshot, ColumnFamilyStore cfs,
Collection<Range<Token>> ranges, ValidationOffsets validationOffsets)
+ private static JournalPartitionAccessor
buildFromSnapshot(MutationJournal.Snapshot snapshot, ColumnFamilyStore cfs,
Collection<Range<Token>> ranges, ValidationOffsets validationOffsets,
PreviewKind previewKind, TimeUUID parentId)
{
TableMetadata metadata = cfs.metadata();
TableId tableId = metadata.id;
@@ -232,7 +264,11 @@ public class JournalMergingValidationIterator extends
ValidationPartitionIterato
}
});
- logger.info("Performing journal validation on {} partitions ({}
bytes) in {}.{}", byKey.size(), estimatedBytes[0], metadata.keyspace,
metadata.name);
+ String prefix = previewKind.logPrefix(parentId);
+ logger.info("{} Performing journal validation on {} partitions ({}
bytes) in {}.{}", prefix, byKey.size(), estimatedBytes[0], metadata.keyspace,
metadata.name);
+ if (logger.isTraceEnabled())
+ logger.trace("{} Journal scan for {}.{}: journalRequest={},
rangeCounts={}",
+ prefix, metadata.keyspace, metadata.name,
validationOffsets.toString(true), rangeCounts);
return new JournalPartitionAccessor(metadata, byKey,
estimatedBytes[0], rangeCounts, snapshot);
}
diff --git
a/src/java/org/apache/cassandra/repair/MutationTrackingPreviewRepairTask.java
b/src/java/org/apache/cassandra/repair/MutationTrackingPreviewRepairTask.java
index 0ed0c0741b..078292455d 100644
---
a/src/java/org/apache/cassandra/repair/MutationTrackingPreviewRepairTask.java
+++
b/src/java/org/apache/cassandra/repair/MutationTrackingPreviewRepairTask.java
@@ -30,6 +30,9 @@ import java.util.concurrent.atomic.AtomicInteger;
import com.google.common.base.Preconditions;
import com.google.common.collect.Sets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import org.apache.cassandra.concurrent.ExecutorPlus;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
@@ -63,6 +66,8 @@ import org.apache.cassandra.utils.concurrent.Future;
*/
public class MutationTrackingPreviewRepairTask extends AbstractRepairTask
{
+ private static final Logger logger =
LoggerFactory.getLogger(MutationTrackingPreviewRepairTask.class);
+
public static final String ESTABLISHING_OFFSETS = "Establishing validation
offsets";
public static final String OFFSETS_ESTABLISHED = "Validation offsets
established";
public static final String DISPATCHING_VALIDATION = "Dispatching
validation";
@@ -130,10 +135,28 @@ public class MutationTrackingPreviewRepairTask extends
AbstractRepairTask
// Extract captured targets from every sync coordinator IR
ran, merge into one
// flat ValidationOffsets, and pass to the validation dispatch.
List<Map<CoordinatorLogId, Offsets.Immutable>> targetsList =
new ArrayList<>();
+ List<String> targetDescriptions = logger.isTraceEnabled() ?
new ArrayList<>() : null;
for (MutationTrackingSyncCoordinator syncCoordinator :
irTask.getSyncCoordinators())
- targetsList.add(syncCoordinator.getCapturedTargets());
+ {
+ Map<CoordinatorLogId, Offsets.Immutable> capturedTargets =
syncCoordinator.getCapturedTargets();
+ targetsList.add(capturedTargets);
+ if (targetDescriptions != null)
+ {
+ targetDescriptions.add(String.format("%s/%s: %s",
+
syncCoordinator.getKeyspace(),
+
syncCoordinator.getRange(),
+ new
ValidationOffsets(capturedTargets).toString(true)));
+ }
+ }
ValidationOffsets validationOffsets =
ValidationOffsets.flatten(targetsList);
+ if (targetDescriptions != null)
+ {
+ logger.trace("[#{}] Captured validation targets from {}
sync coordinators: {}",
+ parentSession, targetsList.size(),
targetDescriptions);
+ logger.trace("[#{}] Agreed validation target: {}",
parentSession, validationOffsets.toString(true));
+ }
+
runValidation(promise, validationOffsets, executor,
validationScheduler);
},
failure -> promise.tryFailure(new RuntimeException("unable to
capture offset: " + failure.getMessage(), failure))
@@ -176,6 +199,11 @@ public class MutationTrackingPreviewRepairTask extends
AbstractRepairTask
if (!participants.isEmpty())
allParticipants.addAll(participants);
+ if (logger.isTraceEnabled())
+ logger.trace("[#{}] Dispatching tracked validation for
{}.{} over {} to {} participants: {}",
+ parentSession, keyspace, String.join(",",
tableNames), commonRange.ranges,
+ participants.size(), participants);
+
for (String tableName : tableNames)
{
RepairJobDesc desc = new RepairJobDesc(parentSession,
TimeUUID.Generator.nextTimeUUID(), keyspace, tableName, commonRange.ranges);
@@ -211,9 +239,8 @@ public class MutationTrackingPreviewRepairTask extends
AbstractRepairTask
emitValidationComplete(promise, mismatchesByTable,
allParticipants, offset);
},
failure -> {
- // Any validation failure fails the whole preview. Cancel
siblings so they
- // don't keep running to completion or their own timeout,
holding merkle
- // trees and registry entries for a result nobody will
consume.
+ // Any validation failure fails the whole preview. Cancel
siblings so they don't keep running
+ // to completion, holding merkle trees and registry
entries for a result nobody will consume.
promise.tryFailure(new RuntimeException("Tracked
validation failed: " + failure.getMessage(), failure));
for (ValidationHandle sibling : handles)
sibling.coordinator.cancel();
diff --git
a/src/java/org/apache/cassandra/repair/MutationTrackingValidationCoordinator.java
b/src/java/org/apache/cassandra/repair/MutationTrackingValidationCoordinator.java
index f876ebb47b..93feb5ff0a 100644
---
a/src/java/org/apache/cassandra/repair/MutationTrackingValidationCoordinator.java
+++
b/src/java/org/apache/cassandra/repair/MutationTrackingValidationCoordinator.java
@@ -24,8 +24,6 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableSet;
@@ -33,7 +31,6 @@ import com.google.common.collect.ImmutableSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.RequestFailure;
@@ -105,12 +102,6 @@ public class MutationTrackingValidationCoordinator
}
future.addListener(() -> REGISTRY.remove(desc, this));
- long timeout =
DatabaseDescriptor.getMutationTrackingValidationTimeout(TimeUnit.SECONDS);
- ctx.scheduledTasks().schedule(() -> {
- if (!future.isDone())
- fail(new TimeoutException("Tracked validation timed out after
" + timeout + "s for " + desc));
- }, timeout, TimeUnit.SECONDS);
-
ValidationRequest request = new ValidationRequest(desc,
FBUtilities.nowInSeconds(), false, offset);
for (InetAddressAndPort participant : participants)
{
@@ -141,7 +132,7 @@ public class MutationTrackingValidationCoordinator
/**
* Invoked by {@link RepairMessageVerbHandler} on receipt of {@link
Verb#MT_VALIDATION_RSP}.
* Trees may be null if the participant reported validation failure.
Synchronized with
- * {@link #complete()} and {@link #fail(Throwable)} so a concurrent
timeout can't race
+ * {@link #complete()} and {@link #fail(Throwable)} so a concurrent
failure cannot race
* a tree being stored.
*/
synchronized void onTreesReceived(InetAddressAndPort from, MerkleTrees
trees)
diff --git a/src/java/org/apache/cassandra/replication/ActivatedTransfers.java
b/src/java/org/apache/cassandra/replication/ActivatedTransfers.java
index 701038894b..1ab3f56648 100644
--- a/src/java/org/apache/cassandra/replication/ActivatedTransfers.java
+++ b/src/java/org/apache/cassandra/replication/ActivatedTransfers.java
@@ -184,6 +184,11 @@ public class ActivatedTransfers implements
Iterable<ShortMutationId>
return transfers.isEmpty();
}
+ public int size()
+ {
+ return transfers.size();
+ }
+
public static Bounds<Token> covering(Collection<SSTableReader> sstables)
{
Preconditions.checkArgument(!sstables.isEmpty());
@@ -251,9 +256,32 @@ public class ActivatedTransfers implements
Iterable<ShortMutationId>
@Override
public String toString()
{
- return "ActivatedTransfers{" +
- "transfers=" + transfers +
- '}';
+ return toString(false);
+ }
+
+ /**
+ * Transfers have no range structure to collapse, so the verbose form
lists ids. Ids are rendered as
+ * {@code hostId:hostLogId:offset} rather than via {@link
ShortMutationId#toString()} to keep the line
+ * readable when several transfers are present. Only enable verbose output
for trace logging that is expected
+ * to be verbose.
+ */
+ public String toString(boolean verbose)
+ {
+ StringBuilder builder = new
StringBuilder("ActivatedTransfers{transfers=").append(size());
+ if (!verbose || transfers.isEmpty())
+ return builder.append('}').toString();
+
+ boolean first = true;
+ for (ActivatedTransfer transfer : transfers)
+ {
+ if (!first)
+ builder.append(',');
+ first = false;
+ builder.append(transfer.id.hostId())
+ .append(':').append(transfer.id.hostLogId())
+ .append(':').append(transfer.id.offset());
+ }
+ return builder.append('}').toString();
}
public static final VersionedSerializer<ActivatedTransfers> serializer =
new VersionedSerializer<>()
diff --git
a/src/java/org/apache/cassandra/replication/ImmutableCoordinatorLogOffsets.java
b/src/java/org/apache/cassandra/replication/ImmutableCoordinatorLogOffsets.java
index df851339b0..e5c00c2784 100644
---
a/src/java/org/apache/cassandra/replication/ImmutableCoordinatorLogOffsets.java
+++
b/src/java/org/apache/cassandra/replication/ImmutableCoordinatorLogOffsets.java
@@ -129,10 +129,32 @@ public class ImmutableCoordinatorLogOffsets implements
CoordinatorLogOffsets<Off
@Override
public String toString()
{
- return "ImmutableCoordinatorLogOffsets{" +
- "mutations=" + mutations +
- ", transfers=" + transfers +
- '}';
+ return toString(false);
+ }
+
+ /**
+ * The default form reports counts only; the verbose form renders per-log
offset ranges and transfer ids. Use
+ * {@link ValidationOffsets#toString(boolean)} when comparing an SSTable's
offsets with an agreed target, so the
+ * two renderings match. Only enable verbose output for trace logging that
is expected to be verbose.
+ */
+ public String toString(boolean verbose)
+ {
+ StringBuilder builder = new
StringBuilder("ImmutableCoordinatorLogOffsets{logs=").append(mutations.size());
+ if (!verbose)
+ return builder.append(",
transfers=").append(transfers.size()).append('}').toString();
+
+ boolean first = true;
+ for (Long logId : mutations)
+ {
+ builder.append(first ? "" : ", ");
+ first = false;
+
+ Offsets.Immutable offsets = mutations.getOffsets(logId);
+ builder.append(CoordinatorLogId.hostId(logId))
+
.append(':').append(CoordinatorLogId.hostLogId(logId)).append(':');
+ Offsets.appendRanges(builder, offsets);
+ }
+ return builder.append(",
transfers=").append(transfers.toString(true)).append('}').toString();
}
@NotThreadSafe
diff --git a/src/java/org/apache/cassandra/replication/Offsets.java
b/src/java/org/apache/cassandra/replication/Offsets.java
index bb98620eee..f8960e2dbb 100644
--- a/src/java/org/apache/cassandra/replication/Offsets.java
+++ b/src/java/org/apache/cassandra/replication/Offsets.java
@@ -23,6 +23,8 @@ import java.util.Collection;
import java.util.Iterator;
import java.util.List;
+import javax.annotation.Nullable;
+
import com.google.common.base.Function;
import com.google.common.base.Objects;
import com.google.common.base.Preconditions;
@@ -180,6 +182,28 @@ public abstract class Offsets implements
Iterable<ShortMutationId>
return builder.append('}').toString();
}
+ static void appendRanges(StringBuilder builder, Offsets.RangeIterator
ranges)
+ {
+ builder.append('{');
+ for (int rendered = 0; ranges.tryAdvance(); rendered++)
+ {
+ if (rendered > 0)
+ builder.append(',');
+
builder.append('[').append(ranges.start()).append(',').append(ranges.end()).append(']');
+ }
+ builder.append('}');
+ }
+
+ static void appendRanges(StringBuilder builder, @Nullable Immutable
offsets)
+ {
+ if (offsets == null)
+ {
+ builder.append("{}");
+ return;
+ }
+ appendRanges(builder, offsets.rangeIterator());
+ }
+
public boolean contains(int offset)
{
if (size == 0)
diff --git a/src/java/org/apache/cassandra/replication/ValidationOffsets.java
b/src/java/org/apache/cassandra/replication/ValidationOffsets.java
index 801d0ff1aa..a22dd35559 100644
--- a/src/java/org/apache/cassandra/replication/ValidationOffsets.java
+++ b/src/java/org/apache/cassandra/replication/ValidationOffsets.java
@@ -149,7 +149,33 @@ public final class ValidationOffsets
@Override
public String toString()
{
- return "ValidationOffsets{logs=" + perLog.size() + '}';
+ return toString(false);
+ }
+
+ /**
+ * Offsets are rendered as ranges, never as expanded ids: a single range
can span thousands of offsets and
+ * {@link Offsets#collectIds} allocates one {@link ShortMutationId} per
offset. Only enable this for trace
+ * logging that is expected to be verbose.
+ */
+ public String toString(boolean verbose)
+ {
+ StringBuilder builder = new
StringBuilder("ValidationOffsets{logs=").append(perLog.size());
+ if (!verbose || perLog.isEmpty())
+ return builder.append('}').toString();
+
+ builder.append(", ");
+ boolean first = true;
+ for (Map.Entry<Long, Offsets.Immutable> entry : perLog.entrySet())
+ {
+ if (!first)
+ builder.append(", ");
+ first = false;
+
+ CoordinatorLogId logId = CoordinatorLogId.fromLong(entry.getKey());
+
builder.append(logId.hostId()).append(':').append(logId.hostLogId()).append(':');
+ Offsets.appendRanges(builder, entry.getValue());
+ }
+ return builder.append('}').toString();
}
public static final UnversionedSerializer<ValidationOffsets> serializer =
new UnversionedSerializer<>()
diff --git a/src/java/org/apache/cassandra/service/StorageService.java
b/src/java/org/apache/cassandra/service/StorageService.java
index 5cb67e0879..2ae74635fa 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -5589,18 +5589,6 @@ public class StorageService extends
NotificationBroadcasterSupport implements IE
logger.info("MutationTrackingSyncTimeout set to {}ms via JMX",
timeoutInMillis);
}
- public long getMutationTrackingValidationTimeout()
- {
- return
DatabaseDescriptor.getMutationTrackingValidationTimeout(MILLISECONDS);
- }
-
- public void setMutationTrackingValidationTimeout(long timeoutInMillis)
- {
- checkArgument(timeoutInMillis > 0);
-
DatabaseDescriptor.setMutationTrackingValidationTimeout(timeoutInMillis);
- logger.info("MutationTrackingValidationTimeout set to {}ms via JMX",
timeoutInMillis);
- }
-
public void evictHungRepairs()
{
logger.info("StorageService#clearPaxosRateLimiters called via jmx");
diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java
b/src/java/org/apache/cassandra/service/StorageServiceMBean.java
index d881d7de5f..277096353f 100644
--- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java
+++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java
@@ -1361,9 +1361,6 @@ public interface StorageServiceMBean extends
NotificationEmitter
long getMutationTrackingSyncTimeout();
void setMutationTrackingSyncTimeout(long timeoutInMillis);
- long getMutationTrackingValidationTimeout();
- void setMutationTrackingValidationTimeout(long timeoutInMillis);
-
public void evictHungRepairs();
public void clearPaxosRepairs();
public void setSkipPaxosRepairCompatibilityCheck(boolean v);
diff --git
a/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedPreviewRepairTest.java
b/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedPreviewRepairTest.java
index c01e0ab06d..3ff9b4e4c6 100644
---
a/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedPreviewRepairTest.java
+++
b/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedPreviewRepairTest.java
@@ -49,7 +49,6 @@ import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
-import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.CassandraWriteContext;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DecoratedKey;
@@ -1240,11 +1239,6 @@ public class TrackedPreviewRepairTest extends
TestBaseImpl
createTrackedSchema(CLUSTER, keyspace, table);
IInvokableInstance coordinator = CLUSTER.get(1);
- long originalTimeoutMillis = coordinator.callOnInstance(() ->
DatabaseDescriptor.getMutationTrackingValidationTimeout(TimeUnit.MILLISECONDS));
-
- // Make the validation timeout essentially infinite, so the only path
to resolution is an explicit failure.
- long longTimeoutMillis = TimeUnit.MINUTES.toMillis(10);
- coordinator.runOnInstance(() ->
DatabaseDescriptor.setMutationTrackingValidationTimeout(longTimeoutMillis));
coordinator.runOnInstance(ThrowOnTreeDifference::activate);
ExecutorService executor = Executors.newSingleThreadExecutor();
@@ -1253,13 +1247,12 @@ public class TrackedPreviewRepairTest extends
TestBaseImpl
Future<NodeToolResult> resultFuture = submitValidate(executor,
coordinator, keyspace);
// This should terminate immediately...
- NodeToolResult result = resultFuture.get(longTimeoutMillis / 2,
TimeUnit.MILLISECONDS);
+ NodeToolResult result = resultFuture.get(1, TimeUnit.MINUTES);
result.asserts().failure();
}
finally
{
coordinator.runOnInstance(ThrowOnTreeDifference::deactivate);
- coordinator.runOnInstance(() ->
DatabaseDescriptor.setMutationTrackingValidationTimeout(originalTimeoutMillis));
ExecutorUtils.shutdownAndWait(10, TimeUnit.SECONDS, executor);
}
}
@@ -1271,11 +1264,6 @@ public class TrackedPreviewRepairTest extends
TestBaseImpl
IInvokableInstance coordinator = CLUSTER.get(1);
IInvokableInstance node2 = CLUSTER.get(2);
- long originalTimeoutMillis = coordinator.callOnInstance(() ->
DatabaseDescriptor.getMutationTrackingValidationTimeout(TimeUnit.MILLISECONDS));
-
- // Make the validation timeout essentially infinite, so the only path
to resolution is an explicit failure.
- long longTimeoutMillis = TimeUnit.MINUTES.toMillis(10);
- coordinator.runOnInstance(() ->
DatabaseDescriptor.setMutationTrackingValidationTimeout(longTimeoutMillis));
node2.runOnInstance(PauseValidationRequest::activateToFail);
ExecutorService executor = Executors.newSingleThreadExecutor();
@@ -1284,13 +1272,12 @@ public class TrackedPreviewRepairTest extends
TestBaseImpl
Future<NodeToolResult> resultFuture = submitValidate(executor,
coordinator, keyspace);
// This should terminate immediately...
- NodeToolResult result = resultFuture.get(longTimeoutMillis / 2,
TimeUnit.MILLISECONDS);
+ NodeToolResult result = resultFuture.get(1, TimeUnit.MINUTES);
result.asserts().failure();
}
finally
{
node2.runOnInstance(PauseValidationRequest::releaseAndDeactivate);
- coordinator.runOnInstance(() ->
DatabaseDescriptor.setMutationTrackingValidationTimeout(originalTimeoutMillis));
ExecutorUtils.shutdownAndWait(10, TimeUnit.SECONDS, executor);
}
}
diff --git
a/test/unit/org/apache/cassandra/db/repair/JournalMergingValidationIteratorTest.java
b/test/unit/org/apache/cassandra/db/repair/JournalMergingValidationIteratorTest.java
new file mode 100644
index 0000000000..32d7cf2e9f
--- /dev/null
+++
b/test/unit/org/apache/cassandra/db/repair/JournalMergingValidationIteratorTest.java
@@ -0,0 +1,228 @@
+/*
+ * 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.db.repair;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.SchemaLoader;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.Mutation;
+import org.apache.cassandra.db.marshal.Int32Type;
+import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.repair.ValidationPartitionIterator;
+import org.apache.cassandra.replication.CoordinatorLogId;
+import org.apache.cassandra.replication.MutationId;
+import org.apache.cassandra.replication.MutationJournal;
+import org.apache.cassandra.replication.MutationTrackingService;
+import org.apache.cassandra.replication.Offsets;
+import org.apache.cassandra.replication.ShortMutationId;
+import org.apache.cassandra.replication.ValidationOffsets;
+import org.apache.cassandra.schema.KeyspaceParams;
+import org.apache.cassandra.schema.ReplicationType;
+import org.apache.cassandra.schema.Schema;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.schema.TableParams;
+import org.apache.cassandra.service.ActiveRepairService;
+import org.apache.cassandra.streaming.PreviewKind;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.TimeUUID;
+
+import static org.apache.cassandra.utils.TimeUUID.Generator.nextTimeUUID;
+
+public class JournalMergingValidationIteratorTest
+{
+ private static final AtomicInteger counter = new AtomicInteger();
+ private static InetAddressAndPort coordinator;
+ private static Token MT;
+
+ @BeforeClass
+ public static void setupClass() throws Exception
+ {
+ SchemaLoader.prepareServer();
+ MutationJournal.start();
+ MutationTrackingService.start();
+ coordinator = InetAddressAndPort.getByName("10.0.0.1");
+ MT = DatabaseDescriptor.getPartitioner().getMinimumToken();
+ }
+
+ private ColumnFamilyStore createTable()
+ {
+ String ks = "ks_" + counter.incrementAndGet();
+ String tbl = "tbl";
+ TableMetadata metadata =
+ TableMetadata.builder(ks, tbl)
+ .addPartitionKeyColumn("k", Int32Type.instance)
+ .addRegularColumn("v", Int32Type.instance)
+
.params(TableParams.builder().gcGraceSeconds(0).build())
+ .build();
+ SchemaLoader.createKeyspace(ks, KeyspaceParams.simple(1,
ReplicationType.tracked), metadata);
+ ColumnFamilyStore cfs =
Schema.instance.getColumnFamilyStoreInstance(metadata.id);
+ cfs.disableAutoCompaction();
+ return cfs;
+ }
+
+ private void registerRepair(TimeUUID parentId, TimeUUID sessionID,
ColumnFamilyStore cfs, Range<Token> range)
+ {
+ ActiveRepairService.instance().registerParentRepairSession(parentId,
+ coordinator,
+
List.of(cfs),
+
Set.of(range),
+ false,
+
ActiveRepairService.UNREPAIRED_SSTABLE,
+ false,
+
PreviewKind.REPAIRED);
+ }
+
+ private Mutation createInsertMutation(TableMetadata tableMetadata, int k,
int v, long timestamp)
+ {
+ DecoratedKey key =
tableMetadata.partitioner.decorateKey(ByteBufferUtil.bytes(k));
+ MutationId mutationId =
MutationTrackingService.instance().nextMutationId(tableMetadata.keyspace,
key.getToken());
+ org.apache.cassandra.db.SimpleBuilders.MutationBuilder builder = new
org.apache.cassandra.db.SimpleBuilders.MutationBuilder(mutationId,
tableMetadata.keyspace, key);
+ builder.timestamp(timestamp);
+ builder.update(tableMetadata).row().add("v", v);
+ return builder.build();
+ }
+
+ private Mutation createDeleteMutation(TableMetadata tableMetadata, int k,
long timestamp, long nowInSec)
+ {
+ DecoratedKey key =
tableMetadata.partitioner.decorateKey(ByteBufferUtil.bytes(k));
+ MutationId mutationId =
MutationTrackingService.instance().nextMutationId(tableMetadata.keyspace,
key.getToken());
+ org.apache.cassandra.db.SimpleBuilders.MutationBuilder builder = new
org.apache.cassandra.db.SimpleBuilders.MutationBuilder(mutationId,
tableMetadata.keyspace, key);
+ builder.timestamp(timestamp);
+ org.apache.cassandra.db.partitions.PartitionUpdate.SimpleBuilder
update = builder.update(tableMetadata);
+ update.nowInSec(nowInSec);
+ update.delete();
+ return builder.build();
+ }
+
+ @Test
+ public void testTombstoneInSSTableShadowsOlderWriteInJournal() throws
Exception
+ {
+ ColumnFamilyStore cfs = createTable();
+ Range<Token> range = new Range<>(MT, MT);
+
+ // Write a deletion for key 1 with timestamp 2000L and
localDeletionTime in the past into the table and flush to SSTable.
+ long deleteTimestamp = 2000L;
+ long localDeletionTime = 100;
+ Mutation delete = createDeleteMutation(cfs.metadata(), 1,
deleteTimestamp, localDeletionTime);
+ delete.apply();
+ cfs.forceBlockingFlush(ColumnFamilyStore.FlushReason.UNIT_TESTS);
+
+ Assert.assertEquals(1, cfs.getLiveSSTables().size());
+
+ // Write an insert into the journal with timestamp 1000L.
+ // It is not flushed, so it only exists in the journal.
+ Mutation insert = createInsertMutation(cfs.metadata(), 1, 1, 1000L);
+ CoordinatorLogId coordLogId =
CoordinatorLogId.fromLong(insert.id().logId());
+ ShortMutationId journalId = new ShortMutationId(coordLogId.asLong(),
insert.id().offset());
+ MutationJournal.instance().write(journalId, insert);
+ MutationJournal.instance().closeCurrentSegmentForTestingIfNonEmpty();
+
+ int maxOffset = Math.max(delete.id().offset(), insert.id().offset());
+ ValidationOffsets validationOffsets = new
ValidationOffsets(Map.of(coordLogId, new Offsets.Immutable(coordLogId, new
int[]{ 0, maxOffset })));
+
+ TimeUUID parentId = nextTimeUUID();
+ TimeUUID sessionID = nextTimeUUID();
+ registerRepair(parentId, sessionID, cfs, range);
+
+ long nowInSec = 200; // nowInSec > localDeletionTime (100)
+ boolean dontPurgeTombstones = false;
+
+ try (ValidationPartitionIterator vi =
cfs.getRepairManager().getValidationIterator(Collections.singleton(range),
parentId, sessionID, false, nowInSec, dontPurgeTombstones, validationOffsets,
null))
+ {
+ // If the SSTable iterator prematurely purges the tombstone before
merging,
+ // the tombstone will fail to shadow the insert from the journal,
and the insert will be resurrected.
+ // With correct purging on the final iterator, the tombstone
shadows the insert, then is purged,
+ // leaving no live or resurrected data.
+ Assert.assertFalse("Deleted row was resurrected because tombstone
was purged before merge", vi.hasNext());
+ }
+ }
+
+ @Test
+ public void testTombstoneInJournalIsPurgedOnFinalIterator() throws
Exception
+ {
+ ColumnFamilyStore cfs = createTable();
+ Range<Token> range = new Range<>(MT, MT);
+
+ // Write an expired deletion into the journal.
+ long deleteTimestamp = 2000L;
+ long localDeletionTime = 100;
+ Mutation delete = createDeleteMutation(cfs.metadata(), 2,
deleteTimestamp, localDeletionTime);
+ CoordinatorLogId coordLogId =
CoordinatorLogId.fromLong(delete.id().logId());
+ ShortMutationId journalId = new ShortMutationId(coordLogId.asLong(),
delete.id().offset());
+ MutationJournal.instance().write(journalId, delete);
+ MutationJournal.instance().closeCurrentSegmentForTestingIfNonEmpty();
+
+ ValidationOffsets validationOffsets = new
ValidationOffsets(Map.of(coordLogId, new Offsets.Immutable(coordLogId, new
int[]{ 0, delete.id().offset() })));
+
+ TimeUUID parentId = nextTimeUUID();
+ TimeUUID sessionID = nextTimeUUID();
+ registerRepair(parentId, sessionID, cfs, range);
+
+ long nowInSec = 200;
+ boolean dontPurgeTombstones = false;
+
+ try (ValidationPartitionIterator vi =
cfs.getRepairManager().getValidationIterator(Collections.singleton(range),
parentId, sessionID, false, nowInSec, dontPurgeTombstones, validationOffsets,
null))
+ {
+ // With final iterator purging, expired tombstones from the
journal must be purged when dontPurgeTombstones is false.
+ Assert.assertFalse("Expired tombstone from journal must be purged
on final iterator", vi.hasNext());
+ }
+ }
+
+ @Test
+ public void testTombstonesRetainedWhenDontPurgeTombstonesIsTrue() throws
Exception
+ {
+ ColumnFamilyStore cfs = createTable();
+ Range<Token> range = new Range<>(MT, MT);
+
+ long deleteTimestamp = 2000L;
+ long localDeletionTime = 100;
+ Mutation delete = createDeleteMutation(cfs.metadata(), 3,
deleteTimestamp, localDeletionTime);
+ CoordinatorLogId coordLogId =
CoordinatorLogId.fromLong(delete.id().logId());
+ ShortMutationId journalId = new ShortMutationId(coordLogId.asLong(),
delete.id().offset());
+ MutationJournal.instance().write(journalId, delete);
+ MutationJournal.instance().closeCurrentSegmentForTestingIfNonEmpty();
+
+ ValidationOffsets validationOffsets = new
ValidationOffsets(Map.of(coordLogId, new Offsets.Immutable(coordLogId, new
int[]{ 0, delete.id().offset() })));
+
+ TimeUUID parentId = nextTimeUUID();
+ TimeUUID sessionID = nextTimeUUID();
+ registerRepair(parentId, sessionID, cfs, range);
+
+ long nowInSec = 200;
+ boolean dontPurgeTombstones = true;
+
+ try (ValidationPartitionIterator vi =
cfs.getRepairManager().getValidationIterator(Collections.singleton(range),
parentId, sessionID, false, nowInSec, dontPurgeTombstones, validationOffsets,
null))
+ {
+ Assert.assertTrue("Tombstones must be retained when
dontPurgeTombstones is true", vi.hasNext());
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
