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

Merge branch 'cassandra-2.0' into cassandra-2.1 

Conflicts:
	CHANGES.txt
	src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java
	test/unit/org/apache/cassandra/db/compaction/BlacklistingCompactionsTest.java
This commit is contained in:
Sylvain Lebresne 2015-06-02 14:51:20 +02:00
parent 14a332470c 9b10928c15
commit c0f96e1d46
4 changed files with 69 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -32,6 +32,7 @@
* Fix streaming not holding ref when stream error (CASSANDRA-9295)
* Fix canonical view returning early opened SSTables (CASSANDRA-9396)
Merged from 2.0:
* Always mark sstable suspect when corrupted (CASSANDRA-9478)
* Add database users and permissions to CQL3 documentation (CASSANDRA-7558)
* Allow JVM_OPTS to be passed to standalone tools (CASSANDRA-5969)
* Fix bad condition in RangeTombstoneList (CASSANDRA-9485)

10
src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java
View File

@ -38,6 +38,9 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
private final boolean validateColumns;
private final String filename;
// Not every SSTableIdentifyIterator is attached to a sstable, so this can be null.
private final SSTableReader sstable;
/**
* Used to iterate through the columns of a row.
* @param sstable SSTable we are reading ffrom.
@ -81,6 +84,7 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
this.dataSize = dataSize;
this.flag = flag;
this.validateColumns = checkData;
this.sstable = sstable;
Descriptor.Version dataVersion = sstable == null ? Descriptor.Version.CURRENT : sstable.descriptor.version;
int expireBefore = (int) (System.currentTimeMillis() / 1000);
@ -119,9 +123,15 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
{
// catch here b/c atomIterator is an AbstractIterator; hasNext reads the value
if (e.getCause() instanceof IOException)
{
if (sstable != null)
sstable.markSuspect();
throw new CorruptSSTableException((IOException)e.getCause(), filename);
}
else
{
throw e;
}
}
}

68
src/java/org/apache/cassandra/io/sstable/SSTableReader.java
View File

@ -956,14 +956,55 @@ public class SSTableReader extends SSTable implements SelfRefCounted<SSTableRead
}
}
/**
* Clone this reader with the provided start and open reason, and set the clone as replacement.
*
* @param newFirst the first key for the replacement (which can be different from the original due to the pre-emptive
* opening of compaction results).
* @param reason the {@code OpenReason} for the replacement.
*
* @return the cloned reader. That reader is set as a replacement by the method.
*/
private SSTableReader cloneAndReplace(DecoratedKey newFirst, OpenReason reason)
{
return cloneAndReplace(newFirst, reason, indexSummary.sharedCopy());
}
/**
* Clone this reader with the new values and set the clone as replacement.
*
* @param newFirst the first key for the replacement (which can be different from the original due to the pre-emptive
* opening of compaction results).
* @param reason the {@code OpenReason} for the replacement.
* @param newSummary the index summary for the replacement.
*
* @return the cloned reader. That reader is set as a replacement by the method.
*/
private SSTableReader cloneAndReplace(DecoratedKey newFirst, OpenReason reason, IndexSummary newSummary)
{
SSTableReader replacement = internalOpen(descriptor,
components,
metadata,
partitioner,
ifile.sharedCopy(),
dfile.sharedCopy(),
newSummary,
bf.sharedCopy(),
maxDataAge,
sstableMetadata,
reason);
replacement.first = newFirst;
replacement.last = last;
replacement.isSuspect.set(isSuspect.get());
setReplacedBy(replacement);
return replacement;
}
public SSTableReader cloneWithNewStart(DecoratedKey newStart, final Runnable runOnClose)
{
synchronized (tidy.global)
{
assert openReason != OpenReason.EARLY;
SSTableReader replacement = new SSTableReader(descriptor, components, metadata, partitioner, ifile.sharedCopy(),
dfile.sharedCopy(), indexSummary.sharedCopy(), bf.sharedCopy(),
maxDataAge, sstableMetadata, OpenReason.MOVED_START);
// TODO: make data/index start accurate for compressed files
// TODO: merge with caller's firstKeyBeyond() work,to save time
if (newStart.compareTo(first) > 0)
@ -982,10 +1023,7 @@ public class SSTableReader extends SSTable implements SelfRefCounted<SSTableRead
};
}
replacement.first = newStart;
replacement.last = this.last;
setReplacedBy(replacement);
return replacement;
return cloneAndReplace(newStart, OpenReason.MOVED_START);
}
}
@ -1004,13 +1042,7 @@ public class SSTableReader extends SSTable implements SelfRefCounted<SSTableRead
}
};
SSTableReader replacement = new SSTableReader(descriptor, components, metadata, partitioner, ifile.sharedCopy(),
dfile.sharedCopy(), indexSummary.sharedCopy(), bf.sharedCopy(),
maxDataAge, sstableMetadata, OpenReason.SHADOWED);
replacement.first = first;
replacement.last = last;
setReplacedBy(replacement);
return replacement;
return cloneAndReplace(first, OpenReason.SHADOWED);
}
}
@ -1067,13 +1099,7 @@ public class SSTableReader extends SSTable implements SelfRefCounted<SSTableRead
StorageMetrics.load.inc(newSize - oldSize);
parent.metric.liveDiskSpaceUsed.inc(newSize - oldSize);
SSTableReader replacement = new SSTableReader(descriptor, components, metadata, partitioner, ifile.sharedCopy(),
dfile.sharedCopy(), newSummary, bf.sharedCopy(), maxDataAge,
sstableMetadata, OpenReason.METADATA_CHANGE);
replacement.first = this.first;
replacement.last = this.last;
setReplacedBy(replacement);
return replacement;
return cloneAndReplace(first, OpenReason.METADATA_CHANGE, newSummary);
}
}

17
test/unit/org/apache/cassandra/db/compaction/BlacklistingCompactionsTest.java
View File

@ -22,9 +22,7 @@ package org.apache.cassandra.db.compaction;
import java.io.RandomAccessFile;
import java.util.Collection;
import java.util.HashSet;
import java.util.Set;
import java.util.*;
import org.junit.After;
import org.junit.AfterClass;
@ -40,6 +38,7 @@ import org.apache.cassandra.utils.ByteBufferUtil;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.apache.cassandra.Util.cellname;
public class BlacklistingCompactionsTest extends SchemaLoader
@ -128,7 +127,14 @@ public class BlacklistingCompactionsTest extends SchemaLoader
{
raf = new RandomAccessFile(sstable.getFilename(), "rw");
assertNotNull(raf);
raf.write(0xFFFFFF);
assertTrue(raf.length() > 20);
raf.seek(new Random().nextInt((int)(raf.length() - 20)));
// We want to write something large enough that the corruption cannot get undetected
// (even without compression)
byte[] corruption = new byte[20];
Arrays.fill(corruption, (byte)0xFF);
raf.write(corruption);
}
finally
{
@ -160,8 +166,7 @@ public class BlacklistingCompactionsTest extends SchemaLoader
break;
}
cfs.truncateBlocking();
assertEquals(failures, sstablesToCorrupt);
assertEquals(sstablesToCorrupt, failures);
}
}