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

Merge branch 'cassandra-3.0' into trunk

This commit is contained in:
Marcus Eriksson 2015-09-10 14:41:52 +02:00
parent c18ab38578 9f0509ee35
commit 55b6e8a072
2 changed files with 20 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -460,12 +460,18 @@ public class CompactionManager implements CompactionManagerMBean
Iterator<SSTableReader> sstableIterator = sstables.iterator();
try
{
List<Range<Token>> normalizedRanges = Range.normalize(ranges);
while (sstableIterator.hasNext())
{
SSTableReader sstable = sstableIterator.next();
for (Range<Token> r : Range.normalize(ranges))
Range<Token> sstableRange = new Range<>(sstable.first.getToken(), sstable.last.getToken());
boolean shouldAnticompact = false;
for (Range<Token> r : normalizedRanges)
{
Range<Token> sstableRange = new Range<>(sstable.first.getToken(), sstable.last.getToken());
if (r.contains(sstableRange))
{
logger.info("SSTable {} fully contained in range {}, mutating repairedAt instead of anticompacting", sstable, r);
@ -473,20 +479,22 @@ public class CompactionManager implements CompactionManagerMBean
sstable.reloadSSTableMetadata();
mutatedRepairStatuses.add(sstable);
sstableIterator.remove();
shouldAnticompact = true;
break;
}
else if (!sstableRange.intersects(r))
{
logger.info("SSTable {} ({}) does not intersect repaired range {}, not touching repairedAt.", sstable, sstableRange, r);
nonAnticompacting.add(sstable);
sstableIterator.remove();
break;
}
else
else if (sstableRange.intersects(r))
{
logger.info("SSTable {} ({}) will be anticompacted on range {}", sstable, sstableRange, r);
shouldAnticompact = true;
}
}
if (!shouldAnticompact)
{
logger.info("SSTable {} ({}) does not intersect repaired ranges {}, not touching repairedAt.", sstable, sstableRange, normalizedRanges);
nonAnticompacting.add(sstable);
sstableIterator.remove();
}
}
cfs.getTracker().notifySSTableRepairedStatusChanged(mutatedRepairStatuses);
txn.cancel(Sets.union(nonAnticompacting, mutatedRepairStatuses));

4
test/unit/org/apache/cassandra/db/compaction/AntiCompactionTest.java
View File

@ -302,14 +302,14 @@ public class AntiCompactionTest
Collection<SSTableReader> sstables = getUnrepairedSSTables(store);
assertEquals(store.getLiveSSTables().size(), sstables.size());
Range<Token> range = new Range<Token>(new BytesToken("-10".getBytes()), new BytesToken("-1".getBytes()));
Range<Token> range = new Range<Token>(new BytesToken("-1".getBytes()), new BytesToken("-10".getBytes()));
List<Range<Token>> ranges = Arrays.asList(range);
try (LifecycleTransaction txn = store.getTracker().tryModify(sstables, OperationType.ANTICOMPACTION);
Refs<SSTableReader> refs = Refs.ref(sstables))
{
CompactionManager.instance.performAnticompaction(store, ranges, refs, txn, 0);
CompactionManager.instance.performAnticompaction(store, ranges, refs, txn, 1);
}
assertThat(store.getLiveSSTables().size(), is(10));