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

Pick sstables to validate as late as possible with inc repairs 

Patch by marcuse; reviewed by yukim for CASSANDRA-8366
This commit is contained in:
Marcus Eriksson 2015-02-17 16:20:35 +01:00
parent 33279dd8c5
commit 2f7077c06c
4 changed files with 42 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
2.1.4
* Pick sstables for validation as late as possible inc repairs (CASSANDRA-8366)
* Fix commitlog getPendingTasks to not increment (CASSANDRA-8856)
* Fix parallelism adjustment in range and secondary index queries
when the first fetch does not satisfy the limit (CASSANDRA-8856)

14
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -2926,4 +2926,18 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
return new ArrayList<>(view.sstables);
}
};
public static final Function<DataTracker.View, List<SSTableReader>> UNREPAIRED_SSTABLES = new Function<DataTracker.View, List<SSTableReader>>()
{
public List<SSTableReader> apply(DataTracker.View view)
{
List<SSTableReader> sstables = new ArrayList<>();
for (SSTableReader sstable : view.sstables)
{
if (!sstable.isRepaired())
sstables.add(sstable);
}
return sstables;
}
};
}

14
src/java/org/apache/cassandra/db/compaction/CompactionManager.java
View File

@ -956,7 +956,19 @@ public class CompactionManager implements CompactionManagerMBean
if (validator.desc.parentSessionId == null || ActiveRepairService.instance.getParentRepairSession(validator.desc.parentSessionId) == null)
sstables = cfs.selectAndReference(ColumnFamilyStore.ALL_SSTABLES).refs;
else
sstables = ActiveRepairService.instance.getParentRepairSession(validator.desc.parentSessionId).getAndReferenceSSTables(cfs.metadata.cfId);
{
ColumnFamilyStore.RefViewFragment refView = cfs.selectAndReference(ColumnFamilyStore.UNREPAIRED_SSTABLES);
sstables = refView.refs;
Set<SSTableReader> currentlyRepairing = ActiveRepairService.instance.currentlyRepairing(cfs.metadata.cfId, validator.desc.parentSessionId);
if (!Sets.intersection(currentlyRepairing, Sets.newHashSet(refView.sstables)).isEmpty())
{
logger.error("Cannot start multiple repair sessions over the same sstables");
throw new RuntimeException("Cannot start multiple repair sessions over the same sstables");
}
ActiveRepairService.instance.getParentRepairSession(validator.desc.parentSessionId).addSSTables(cfs.metadata.cfId, refView.sstables);
}
if (validator.gcBefore > 0)
gcBefore = validator.gcBefore;

41
src/java/org/apache/cassandra/service/ActiveRepairService.java
View File

@ -305,38 +305,16 @@ public class ActiveRepairService
public synchronized void registerParentRepairSession(UUID parentRepairSession, List<ColumnFamilyStore> columnFamilyStores, Collection<Range<Token>> ranges)
{
Map<UUID, Set<SSTableReader>> sstablesToRepair = new HashMap<>();
for (ColumnFamilyStore cfs : columnFamilyStores)
{
Set<SSTableReader> sstables = new HashSet<>();
Set<SSTableReader> currentlyRepairing = currentlyRepairing(cfs.metadata.cfId);
for (SSTableReader sstable : cfs.getSSTables())
{
if (new Bounds<>(sstable.first.getToken(), sstable.last.getToken()).intersects(ranges))
{
if (!sstable.isRepaired())
{
if (currentlyRepairing.contains(sstable))
{
logger.error("Already repairing "+sstable+", can not continue.");
throw new RuntimeException("Already repairing "+sstable+", can not continue.");
}
sstables.add(sstable);
}
}
}
sstablesToRepair.put(cfs.metadata.cfId, sstables);
}
parentRepairSessions.put(parentRepairSession, new ParentRepairSession(columnFamilyStores, ranges, sstablesToRepair, System.currentTimeMillis()));
parentRepairSessions.put(parentRepairSession, new ParentRepairSession(columnFamilyStores, ranges, System.currentTimeMillis()));
}
private Set<SSTableReader> currentlyRepairing(UUID cfId)
public Set<SSTableReader> currentlyRepairing(UUID cfId, UUID parentRepairSession)
{
Set<SSTableReader> repairing = new HashSet<>();
for (Map.Entry<UUID, ParentRepairSession> entry : parentRepairSessions.entrySet())
{
Collection<SSTableReader> sstables = entry.getValue().sstableMap.get(cfId);
if (sstables != null)
if (sstables != null && !entry.getKey().equals(parentRepairSession))
repairing.addAll(sstables);
}
return repairing;
@ -419,12 +397,12 @@ public class ActiveRepairService
public final Map<UUID, Set<SSTableReader>> sstableMap;
public final long repairedAt;
public ParentRepairSession(List<ColumnFamilyStore> columnFamilyStores, Collection<Range<Token>> ranges, Map<UUID, Set<SSTableReader>> sstables, long repairedAt)
public ParentRepairSession(List<ColumnFamilyStore> columnFamilyStores, Collection<Range<Token>> ranges, long repairedAt)
{
for (ColumnFamilyStore cfs : columnFamilyStores)
this.columnFamilyStores.put(cfs.metadata.cfId, cfs);
this.ranges = ranges;
this.sstableMap = sstables;
this.sstableMap = new HashMap<>();
this.repairedAt = repairedAt;
}
@ -452,6 +430,15 @@ public class ActiveRepairService
return new Refs<>(references.build());
}
public void addSSTables(UUID cfId, Collection<SSTableReader> sstables)
{
Set<SSTableReader> existingSSTables = this.sstableMap.get(cfId);
if (existingSSTables == null)
existingSSTables = new HashSet<>();
existingSSTables.addAll(sstables);
this.sstableMap.put(cfId, existingSSTables);
}
@Override
public String toString()
{