natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

IR may leak SSTables with pending repair when coming from streaming 

patch by David Capwell; reviewed by Blake Eggleston for CASSANDRA-19182
This commit is contained in:
David Capwell 2024-05-01 09:47:06 -07:00 committed by ci worker
parent 9157d98e4c
commit c14abb40b2
10 changed files with 89 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
4.0.13
* IR may leak SSTables with pending repair when coming from streaming (CASSANDRA-19182)
* Streaming exception race creates corrupt transaction log files that prevent restart (CASSANDRA-18736)
* Fix CQL tojson timestamp output on negative timestamp values before Gregorian calendar reform in 1582 (CASSANDRA-19566)
* Fix few types issues and implement types compatibility tests (CASSANDRA-19479)

1
src/java/org/apache/cassandra/db/compaction/AbstractStrategyHolder.java
View File

@ -181,6 +181,7 @@ public abstract class AbstractStrategyHolder
return new GroupedSSTableContainer(this);
}
public abstract void addSSTable(SSTableReader sstable);
public abstract void addSSTables(GroupedSSTableContainer sstables);
public abstract void removeSSTables(GroupedSSTableContainer sstables);

6
src/java/org/apache/cassandra/db/compaction/CompactionStrategyHolder.java
View File

@ -138,6 +138,12 @@ public class CompactionStrategyHolder extends AbstractStrategyHolder
return tasks;
}
@Override
public void addSSTable(SSTableReader sstable)
{
getStrategyFor(sstable).addSSTable(sstable);
}
@Override
public void addSSTables(GroupedSSTableContainer sstables)
{

18
src/java/org/apache/cassandra/db/compaction/CompactionStrategyManager.java
View File

@ -434,6 +434,20 @@ public class CompactionStrategyManager implements INotificationConsumer
}
}
@VisibleForTesting
public boolean hasPendingRepairSSTable(UUID sessionID, SSTableReader sstable)
{
readLock.lock();
try
{
return pendingRepairs.hasPendingRepairSSTable(sessionID, sstable) || transientRepairs.hasPendingRepairSSTable(sessionID, sstable);
}
finally
{
readLock.unlock();
}
}
public void shutdown()
{
writeLock.lock();
@ -608,7 +622,7 @@ public class CompactionStrategyManager implements INotificationConsumer
private void handleFlushNotification(Iterable<SSTableReader> added)
{
for (SSTableReader sstable : added)
compactionStrategyFor(sstable).addSSTable(sstable);
getHolder(sstable).addSSTable(sstable);
}
private int getHolderIndex(SSTableReader sstable)
@ -1147,6 +1161,8 @@ public class CompactionStrategyManager implements INotificationConsumer
*/
public void mutateRepaired(Collection<SSTableReader> sstables, long repairedAt, UUID pendingRepair, boolean isTransient) throws IOException
{
if (sstables.isEmpty())
return;
Set<SSTableReader> changed = new HashSet<>();
writeLock.lock();

13
src/java/org/apache/cassandra/db/compaction/PendingRepairHolder.java
View File

@ -30,7 +30,6 @@ import com.google.common.collect.Iterables;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.SerializationHeader;
import org.apache.cassandra.db.lifecycle.LifecycleNewTracker;
import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.index.Index;
@ -149,6 +148,13 @@ public class PendingRepairHolder extends AbstractStrategyHolder
return tasks;
}
@Override
public void addSSTable(SSTableReader sstable)
{
Preconditions.checkArgument(managesSSTable(sstable), "Attempting to add sstable from wrong holder");
managers.get(router.getIndexForSSTable(sstable)).addSSTable(sstable);
}
AbstractCompactionTask getNextRepairFinishedTask()
{
List<TaskSupplier> repairFinishedSuppliers = getRepairFinishedTaskSuppliers();
@ -282,4 +288,9 @@ public class PendingRepairHolder extends AbstractStrategyHolder
{
return Iterables.any(managers, prm -> prm.containsSSTable(sstable));
}
public boolean hasPendingRepairSSTable(UUID sessionID, SSTableReader sstable)
{
return Iterables.any(managers, prm -> prm.hasPendingRepairSSTable(sessionID, sstable));
}
}

11
src/java/org/apache/cassandra/db/compaction/PendingRepairManager.java
View File

@ -464,7 +464,7 @@ class PendingRepairManager
public synchronized boolean hasDataForSession(UUID sessionID)
{
return strategies.keySet().contains(sessionID);
return strategies.containsKey(sessionID);
}
boolean containsSSTable(SSTableReader sstable)
@ -482,6 +482,15 @@ class PendingRepairManager
return group.entrySet().stream().map(g -> strategies.get(g.getKey()).getUserDefinedTask(g.getValue(), gcBefore)).collect(Collectors.toList());
}
@VisibleForTesting
public synchronized boolean hasPendingRepairSSTable(UUID sessionID, SSTableReader sstable)
{
AbstractCompactionStrategy strat = strategies.get(sessionID);
if (strat == null)
return false;
return strat.getSSTables().contains(sstable);
}
/**
* promotes/demotes sstables involved in a consistent repair that has been finalized, or failed
*/

2
src/java/org/apache/cassandra/db/lifecycle/Tracker.java
View File

@ -460,6 +460,8 @@ public class Tracker
public void notifySSTableRepairedStatusChanged(Collection<SSTableReader> repairStatusesChanged)
{
if (repairStatusesChanged.isEmpty())
return;
INotification notification = new SSTableRepairStatusChanged(repairStatusesChanged);
for (INotificationConsumer subscriber : subscribers)
subscriber.handleNotification(notification, this);

34
test/distributed/org/apache/cassandra/distributed/test/DistributedRepairUtils.java
View File

@ -20,12 +20,15 @@ package org.apache.cassandra.distributed.test;
import java.util.Collections;
import java.util.Set;
import java.util.UUID;
import java.util.function.Consumer;
import com.google.common.collect.ImmutableSet;
import org.apache.commons.lang3.ArrayUtils;
import org.junit.Assert;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.ICluster;
import org.apache.cassandra.distributed.api.IInvokableInstance;
@ -33,7 +36,10 @@ import org.apache.cassandra.distributed.api.NodeToolResult;
import org.apache.cassandra.distributed.api.QueryResult;
import org.apache.cassandra.distributed.api.Row;
import org.apache.cassandra.distributed.impl.AbstractCluster;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.metrics.StorageMetrics;
import org.apache.cassandra.repair.consistent.LocalSession;
import org.apache.cassandra.service.ActiveRepairService;
import static org.apache.cassandra.utils.Retry.retryWithBackoffBlocking;
@ -167,6 +173,34 @@ public final class DistributedRepairUtils
Assert.assertFalse("Only one repair expected, but found more than one", rs.hasNext());
}
public static void assertNoSSTableLeak(ICluster<IInvokableInstance> cluster, String ks, String table)
{
cluster.forEach(i -> {
String name = "node" + i.config().num();
i.forceCompact(ks, table); // cleanup happens in compaction, so run before checking
i.runOnInstance(() -> {
ColumnFamilyStore cfs = Keyspace.open(ks).getColumnFamilyStore(table);
for (SSTableReader sstable : cfs.getTracker().getView().liveSSTables())
{
UUID pendingRepair = sstable.getSSTableMetadata().pendingRepair;
if (pendingRepair == null)
continue;
LocalSession session = ActiveRepairService.instance.consistent.local.getSession(pendingRepair);
// repair maybe async, so some participates may still think the repair is active, which means the sstable SHOULD link to it
if (session != null && !session.isCompleted())
continue;
// The session is complete, yet the sstable is not updated... is this still pending in compaction?
if (cfs.getCompactionStrategyManager().hasPendingRepairSSTable(pendingRepair, sstable))
continue;
// compaction does not know about the pending repair... race condition since this check started?
if (sstable.getSSTableMetadata().pendingRepair == null)
continue; // yep, race condition... ignore
throw new AssertionError(String.format("%s had leak detected on sstable %s", name, sstable.descriptor));
}
});
});
}
public enum RepairType {
FULL {
public String[] append(String... args)

3
test/distributed/org/apache/cassandra/distributed/test/RepairCoordinatorFast.java
View File

@ -40,6 +40,7 @@ import org.apache.cassandra.service.StorageService;
import static java.lang.String.format;
import static org.apache.cassandra.distributed.api.IMessageFilters.Matcher.of;
import static org.apache.cassandra.distributed.test.DistributedRepairUtils.assertNoSSTableLeak;
import static org.apache.cassandra.distributed.test.DistributedRepairUtils.assertParentRepairFailedWithMessageContains;
import static org.apache.cassandra.distributed.test.DistributedRepairUtils.assertParentRepairNotExist;
import static org.apache.cassandra.distributed.test.DistributedRepairUtils.assertParentRepairSuccess;
@ -82,6 +83,7 @@ public abstract class RepairCoordinatorFast extends RepairCoordinatorBase
}
Assert.assertEquals(repairExceptions, getRepairExceptions(CLUSTER, 2));
assertNoSSTableLeak(CLUSTER, KEYSPACE, table);
});
}
@ -398,6 +400,7 @@ public abstract class RepairCoordinatorFast extends RepairCoordinatorBase
{
assertParentRepairNotExist(CLUSTER, KEYSPACE, table);
}
assertNoSSTableLeak(CLUSTER, KEYSPACE, table);
}
finally
{

3
test/distributed/org/apache/cassandra/distributed/test/RepairCoordinatorNeighbourDown.java
View File

@ -40,6 +40,7 @@ import org.apache.cassandra.utils.FBUtilities;
import static java.lang.String.format;
import static org.apache.cassandra.distributed.api.IMessageFilters.Matcher.of;
import static org.apache.cassandra.distributed.test.DistributedRepairUtils.assertNoSSTableLeak;
import static org.apache.cassandra.distributed.test.DistributedRepairUtils.assertParentRepairFailedWithMessageContains;
import static org.apache.cassandra.distributed.test.DistributedRepairUtils.assertParentRepairNotExist;
import static org.apache.cassandra.distributed.test.DistributedRepairUtils.getRepairExceptions;
@ -125,6 +126,7 @@ public abstract class RepairCoordinatorNeighbourDown extends RepairCoordinatorBa
{
assertParentRepairNotExist(CLUSTER, KEYSPACE, table);
}
assertNoSSTableLeak(CLUSTER, KEYSPACE, table);
});
}
@ -184,6 +186,7 @@ public abstract class RepairCoordinatorNeighbourDown extends RepairCoordinatorBa
{
assertParentRepairNotExist(CLUSTER, KEYSPACE, table);
}
assertNoSSTableLeak(CLUSTER, KEYSPACE, table);
});
}
}