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

better handling for amid compaction failure; patch by yukim reviewed by slebresne for CASSANDRA-5137

This commit is contained in:
Yuki Morishita 2013-01-11 12:32:59 -06:00
parent 1cbbba095a
commit 3cc8656f8f
3 changed files with 43 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -9,6 +9,7 @@
* fix user defined compaction to run against 1.1 data directory (CASSANDRA-5118)
* Fix CQL3 BATCH authorization caching (CASSANDRA-5145)
* fix get_count returns incorrect value with TTL (CASSANDRA-5099)
* better handling for amid compaction failure (CASSANDRA-5137)
1.1.8

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

@ -244,20 +244,33 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
Directories.SSTableLister sstableFiles = directories.sstableLister().skipCompacted(true).skipTemporary(true);
Collection<SSTableReader> sstables = SSTableReader.batchOpen(sstableFiles.list().entrySet(), savedKeys, data, metadata, this.partitioner);
// Filter non-compacted sstables, remove compacted ones
Set<Integer> compactedSSTables = new HashSet<Integer>();
for (SSTableReader sstable : sstables)
compactedSSTables.addAll(sstable.getAncestors());
Set<SSTableReader> liveSSTables = new HashSet<SSTableReader>();
for (SSTableReader sstable : sstables)
if (metadata.getDefaultValidator().isCommutative())
{
if (compactedSSTables.contains(sstable.descriptor.generation))
sstable.releaseReference(); // this amount to deleting the sstable
else
liveSSTables.add(sstable);
// Filter non-compacted sstables, remove compacted ones
Set<Integer> compactedSSTables = new HashSet<Integer>();
for (SSTableReader sstable : sstables)
compactedSSTables.addAll(sstable.getAncestors());
Set<SSTableReader> liveSSTables = new HashSet<SSTableReader>();
for (SSTableReader sstable : sstables)
{
if (compactedSSTables.contains(sstable.descriptor.generation))
{
logger.info("{} is already compacted and will be removed.", sstable);
sstable.markCompacted(); // we need to mark as compacted to be deleted
sstable.releaseReference(); // this amount to deleting the sstable
}
else
{
liveSSTables.add(sstable);
}
}
data.addInitialSSTables(liveSSTables);
}
else
{
data.addInitialSSTables(sstables);
}
data.addInitialSSTables(liveSSTables);
}
// compaction strategy should be created after the CFS has been prepared

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

@ -32,9 +32,7 @@ import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.compaction.CompactionManager.CompactionExecutorStatsCollector;
import org.apache.cassandra.io.sstable.SSTable;
import org.apache.cassandra.io.sstable.SSTableReader;
import org.apache.cassandra.io.sstable.SSTableWriter;
import org.apache.cassandra.io.sstable.*;
import org.apache.cassandra.utils.CloseableIterator;
import org.apache.cassandra.utils.FBUtilities;
@ -127,7 +125,7 @@ public class CompactionTask extends AbstractCompactionTask
// we can't preheat until the tracker has been set. This doesn't happen until we tell the cfs to
// replace the old entries. Track entries to preheat here until then.
Map<SSTableReader, Map<DecoratedKey, Long>> cachedKeyMap = new HashMap<SSTableReader, Map<DecoratedKey, Long>>();
Map<Descriptor, Map<DecoratedKey, Long>> cachedKeyMap = new HashMap<Descriptor, Map<DecoratedKey, Long>>();
Collection<SSTableReader> sstables = new ArrayList<SSTableReader>();
Collection<SSTableWriter> writers = new ArrayList<SSTableWriter>();
@ -175,9 +173,8 @@ public class CompactionTask extends AbstractCompactionTask
}
if (!nni.hasNext() || newSSTableSegmentThresholdReached(writer))
{
SSTableReader toIndex = writer.closeAndOpenReader(getMaxDataAge(toCompact));
cachedKeyMap.put(toIndex, cachedKeys);
sstables.add(toIndex);
// tmp = false because later we want to query it with descriptor from SSTableReader
cachedKeyMap.put(writer.descriptor.asTemporary(false), cachedKeys);
if (nni.hasNext())
{
writer = cfs.createCompactionWriter(keysPerSSTable, compactionFileLocation, toCompact);
@ -186,11 +183,21 @@ public class CompactionTask extends AbstractCompactionTask
}
}
}
long maxAge = getMaxDataAge(toCompact);
for (SSTableWriter completedWriter : writers)
sstables.add(completedWriter.closeAndOpenReader(maxAge));
}
catch (Exception e)
{
for (SSTableWriter writer : writers)
writer.abort();
// also remove already completed SSTables
for (SSTableReader sstable : sstables)
{
sstable.markCompacted();
sstable.releaseReference();
}
throw FBUtilities.unchecked(e);
}
finally
@ -202,11 +209,10 @@ public class CompactionTask extends AbstractCompactionTask
cfs.replaceCompactedSSTables(toCompact, sstables, compactionType);
// TODO: this doesn't belong here, it should be part of the reader to load when the tracker is wired up
for (Map.Entry<SSTableReader, Map<DecoratedKey, Long>> ssTableReaderMapEntry : cachedKeyMap.entrySet())
for (SSTableReader sstable : sstables)
{
SSTableReader key = ssTableReaderMapEntry.getKey();
for (Map.Entry<DecoratedKey, Long> entry : ssTableReaderMapEntry.getValue().entrySet())
key.cacheKey(entry.getKey(), entry.getValue());
for (Map.Entry<DecoratedKey, Long> entry : cachedKeyMap.get(sstable.descriptor).entrySet())
sstable.cacheKey(entry.getKey(), entry.getValue());
}
long dTime = System.currentTimeMillis() - startTime;