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 cassandra-3.7

This commit is contained in:
Stefania Alborghetti 2016-05-26 10:51:46 +08:00
parent f39a352fdf 85cc390189
commit 68d20edb2b
18 changed files with 439 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,6 +1,7 @@
3.7
* Don't use static dataDirectories field in Directories instances (CASSANDRA-11647)
Merged from 3.0:
* Prevent OOM failures on SSTable corruption, improve tests for corruption detection (CASSANDRA-9530)
* Use CFS.initialDirectories when clearing snapshots (CASSANDRA-11705)
* Allow compaction strategies to disable early open (CASSANDRA-11754)
* Refactor Materialized View code (CASSANDRA-11475)

11
NEWS.txt
View File

@ -13,6 +13,17 @@ restore snapshots created with the previous major version using the
'sstableloader' tool. You can upgrade the file format of your snapshots
using the provided 'sstableupgrade' tool.
3.7
===
Upgrading
---------
- A maximum size for SSTables values has been introduced, to prevent out of memory
exceptions when reading corrupt SSTables. This maximum size can be set via
max_value_size_in_mb in cassandra.yaml. The default is 256MB, which matches the default
value of native_transport_max_frame_size_in_mb. SSTables will be considered corrupt if
they contain values whose size exceeds this limit. See CASSANDRA-9530 for more details.
3.6
=====

8
conf/cassandra.yaml
View File

@ -564,7 +564,8 @@ native_transport_port: 9042
# native_transport_max_threads: 128
#
# The maximum size of allowed frame. Frame (requests) larger than this will
# be rejected as invalid. The default is 256MB.
# be rejected as invalid. The default is 256MB. If you're changing this parameter,
# you may want to adjust max_value_size_in_mb accordingly.
# native_transport_max_frame_size_in_mb: 256
# The maximum number of concurrent client connections.
@ -1041,3 +1042,8 @@ compaction_large_partition_warning_threshold_mb: 100
# Adjust the threshold based on your application throughput requirement
# By default, Cassandra logs GC Pauses greater than 200 ms at INFO level
gc_warn_threshold_in_ms: 1000
# Maximum size of any value in SSTables. Safety measure to detect SSTable corruption
# early. Any value size larger than this threshold will result into marking an SSTable
# as corrupted.
# max_value_size_in_mb: 256

6
src/java/org/apache/cassandra/config/Config.java
View File

@ -155,6 +155,12 @@ public class Config
@Deprecated
public Integer thrift_max_message_length_in_mb = 16;
/**
* Max size of values in SSTables, in MegaBytes.
* Default is the same as the native protocol frame limit: 256Mb.
* See AbstractType for how it is used.
*/
public Integer max_value_size_in_mb = 256;
public Integer thrift_framed_transport_size_in_mb = 15;
public Boolean snapshot_before_compaction = false;

8
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -764,6 +764,9 @@ public class DatabaseDescriptor
{
throw new ConfigurationException("Encryption must be enabled in client_encryption_options for native_transport_port_ssl", false);
}
if (conf.max_value_size_in_mb == null || conf.max_value_size_in_mb <= 0)
throw new ConfigurationException("max_value_size_in_mb must be positive", false);
}
private static FileStore guessFileStore(String dir) throws IOException
@ -909,6 +912,11 @@ public class DatabaseDescriptor
return conf.thrift_framed_transport_size_in_mb * 1024 * 1024;
}
public static int getMaxValueSize()
{
return conf.max_value_size_in_mb * 1024 * 1024;
}
/**
* Creates all storage-related directories.
*/

4
src/java/org/apache/cassandra/db/ClusteringPrefix.java
View File

@ -449,7 +449,9 @@ public interface ClusteringPrefix extends IMeasurableMemory, Clusterable
public void prepare(int flags, int extendedFlags) throws IOException
{
assert !UnfilteredSerializer.isStatic(extendedFlags) : "Flags = " + flags;
if (UnfilteredSerializer.isStatic(extendedFlags))
throw new IOException("Corrupt flags value for clustering prefix (isStatic flag set): " + flags);
this.nextIsRow = UnfilteredSerializer.kind(flags) == Unfiltered.Kind.ROW;
this.nextKind = nextIsRow ? Kind.CLUSTERING : ClusteringPrefix.Kind.values()[in.readByte()];
this.nextSize = nextIsRow ? comparator.size() : in.readUnsignedShort();

2
src/java/org/apache/cassandra/db/columniterator/AbstractSSTableIterator.java
View File

@ -365,7 +365,7 @@ public abstract class AbstractSSTableIterator implements UnfilteredRowIterator
{
return hasNextInternal();
}
catch (IOException e)
catch (IOException | IndexOutOfBoundsException e)
{
try
{

46
src/java/org/apache/cassandra/db/marshal/AbstractType.java
View File

@ -27,9 +27,11 @@ import java.util.Comparator;
import java.util.List;
import java.util.Map;
import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.CQL3Type;
import org.apache.cassandra.cql3.Term;
import org.apache.cassandra.db.TypeSizes;
@ -41,11 +43,9 @@ import org.apache.cassandra.utils.FastByteOperations;
import org.github.jamm.Unmetered;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.ByteBufferUtil;
import static org.apache.cassandra.db.marshal.AbstractType.ComparisonType.CUSTOM;
import static org.apache.cassandra.db.marshal.AbstractType.ComparisonType.NOT_COMPARABLE;
/**
* Specifies a Comparator for a specific type of ByteBuffer.
@ -62,7 +62,7 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>
public final Comparator<ByteBuffer> reverseComparator;
public static enum ComparisonType
public enum ComparisonType
{
/**
* This type should never be compared
@ -82,10 +82,23 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>
public final ComparisonType comparisonType;
public final boolean isByteOrderComparable;
/**
* The maximum size of values for this type, used when some values are not of fixed length,
* that is valueLengthIfFixed() returns -1.
*/
public int maxValueSize;
protected AbstractType(ComparisonType comparisonType)
{
this(comparisonType, DatabaseDescriptor.getMaxValueSize());
}
protected AbstractType(ComparisonType comparisonType, int maxValueSize)
{
this.comparisonType = comparisonType;
this.isByteOrderComparable = comparisonType == ComparisonType.BYTE_ORDER;
this.maxValueSize = maxValueSize;
reverseComparator = (o1, o2) -> AbstractType.this.compare(o2, o1);
try
{
@ -101,6 +114,17 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>
}
}
/**
* Change the maximum value size, this should only be called for testing.
* Unfortunately, ensuring we use a type created with a different maxValueSize
* is too hard at the moment, due to the pervasive use of the type's singleton instances.
*/
@VisibleForTesting
public void setMaxValueSize(int maxValueSize)
{
this.maxValueSize = maxValueSize;
}
public static List<String> asCQLTypeStringList(List<AbstractType<?>> abstractTypes)
{
List<String> r = new ArrayList<>(abstractTypes.size());
@ -380,7 +404,7 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>
}
/**
* The length of values for this type if all values are of fixed length, -1 otherwise.
* The length of values for this type if all values are of fixed length, -1 otherwise.
*/
protected int valueLengthIfFixed()
{
@ -408,10 +432,22 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>
public ByteBuffer readValue(DataInputPlus in) throws IOException
{
int length = valueLengthIfFixed();
if (length >= 0)
return ByteBufferUtil.read(in, length);
else
return ByteBufferUtil.readWithVIntLength(in);
{
int l = (int)in.readUnsignedVInt();
if (l < 0)
throw new IOException("Corrupt (negative) value length encountered");
if (l > maxValueSize)
throw new IOException(String.format("Corrupt value length %d encountered, as it exceeds the maximum of %d, " +
"which is set via max_value_size_in_mb in cassandra.yaml",
l, maxValueSize));
return ByteBufferUtil.read(in, l);
}
}
public void skipValue(DataInputPlus in) throws IOException

2
src/java/org/apache/cassandra/db/rows/Rows.java
View File

@ -59,7 +59,7 @@ public abstract class Rows
}
/**
* Collect statistics ont a given row.
* Collect statistics on a given row.
*
* @param row the row for which to collect stats.
* @param collector the stats collector.

13
src/java/org/apache/cassandra/db/rows/UnfilteredSerializer.java
View File

@ -375,9 +375,18 @@ public class UnfilteredSerializer
}
else
{
assert !isStatic(extendedFlags); // deserializeStaticRow should be used for that.
// deserializeStaticRow should be used for that.
if (isStatic(extendedFlags))
throw new IOException("Corrupt flags value for unfiltered partition (isStatic flag set): " + flags);
builder.newRow(Clustering.serializer.deserialize(in, helper.version, header.clusteringTypes()));
return deserializeRowBody(in, header, helper, flags, extendedFlags, builder);
Row row = deserializeRowBody(in, header, helper, flags, extendedFlags, builder);
// we do not write empty rows because Rows.collectStats(), called by BTW.applyToRow(), asserts that rows are not empty
// if we don't throw here, then later the very same assertion in Rows.collectStats() will fail compactions
// see BlackListingCompactionsTest and CASSANDRA-9530 for details
if (row.isEmpty())
throw new IOException("Corrupt empty row found in unfiltered partition");
return row;
}
}

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

@ -99,6 +99,11 @@ public class SSTableIdentityIterator extends AbstractIterator<Unfiltered> implem
{
return iterator.hasNext() ? iterator.next() : endOfData();
}
catch (IndexOutOfBoundsException e)
{
sstable.markSuspect();
throw new CorruptSSTableException(e, filename);
}
catch (IOError e)
{
if (e.getCause() instanceof IOException)

2
src/java/org/apache/cassandra/io/sstable/format/big/BigTableScanner.java
View File

@ -330,7 +330,7 @@ public class BigTableScanner implements ISSTableScanner
if (dataRange == null)
{
dfile.seek(currentEntry.position);
ByteBufferUtil.readWithShortLength(dfile); // key
ByteBufferUtil.skipShortLength(dfile); // key
return new SSTableIdentityIterator(sstable, dfile, partitionKey());
}

2
src/java/org/apache/cassandra/io/util/RebufferingInputStream.java
View File

@ -65,7 +65,7 @@ public abstract class RebufferingInputStream extends InputStream implements Data
{
int read = read(b, off, len);
if (read < len)
throw new EOFException();
throw new EOFException("EOF after " + read + " bytes out of " + len);
}
@Override

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

@ -28,26 +28,33 @@ import org.junit.After;
import org.junit.BeforeClass;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.Util;
import org.apache.cassandra.cache.ChunkCache;
import org.apache.cassandra.config.*;
import org.apache.cassandra.db.*;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.schema.*;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
public class BlacklistingCompactionsTest
{
private static final Logger logger = LoggerFactory.getLogger(BlacklistingCompactionsTest.class);
private static Random random;
private static final String KEYSPACE1 = "BlacklistingCompactionsTest";
private static final String CF_STANDARD1 = "Standard1";
private static final String STANDARD_STCS = "Standard_STCS";
private static final String STANDARD_LCS = "Standard_LCS";
@After
public void leakDetect() throws InterruptedException
@ -61,10 +68,26 @@ public class BlacklistingCompactionsTest
@BeforeClass
public static void defineSchema() throws ConfigurationException
{
long seed = System.nanoTime();
//long seed = 754271160974509L; // CASSANDRA-9530: use this seed to reproduce compaction failures if reading empty rows
logger.info("Seed {}", seed);
random = new Random(seed);
SchemaLoader.prepareServer();
SchemaLoader.createKeyspace(KEYSPACE1,
KeyspaceParams.simple(1),
SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARD1));
SchemaLoader.standardCFMD(KEYSPACE1, STANDARD_STCS).compaction(CompactionParams.DEFAULT),
SchemaLoader.standardCFMD(KEYSPACE1, STANDARD_LCS).compaction(CompactionParams.lcs(Collections.emptyMap())));
Keyspace keyspace = Keyspace.open(KEYSPACE1);
for (String tableName : new String[] {STANDARD_STCS, STANDARD_LCS})
{
final ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(tableName);
for (ColumnDefinition cd : cfs.metadata.allColumns())
cd.type.setMaxValueSize(1024 * 1024); // set max value size to 1MB
}
closeStdErr();
}
@ -81,20 +104,20 @@ public class BlacklistingCompactionsTest
@Test
public void testBlacklistingWithSizeTieredCompactionStrategy() throws Exception
{
testBlacklisting(SizeTieredCompactionStrategy.class.getCanonicalName());
testBlacklisting(STANDARD_STCS);
}
@Test
public void testBlacklistingWithLeveledCompactionStrategy() throws Exception
{
testBlacklisting(LeveledCompactionStrategy.class.getCanonicalName());
testBlacklisting(STANDARD_LCS);
}
public void testBlacklisting(String compactionStrategy) throws Exception
private void testBlacklisting(String tableName) throws Exception
{
// this test does enough rows to force multiple block indexes to be used
Keyspace keyspace = Keyspace.open(KEYSPACE1);
final ColumnFamilyStore cfs = keyspace.getColumnFamilyStore("Standard1");
final ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(tableName);
final int ROWS_PER_SSTABLE = 10;
final int SSTABLES = cfs.metadata.params.minIndexInterval * 2 / ROWS_PER_SSTABLE;
@ -143,11 +166,11 @@ public class BlacklistingCompactionsTest
raf = new RandomAccessFile(sstable.getFilename(), "rw");
assertNotNull(raf);
assertTrue(raf.length() > corruptionSize);
raf.seek(new Random().nextInt((int)(raf.length() - corruptionSize)));
raf.seek(random.nextInt((int)(raf.length() - corruptionSize)));
// We want to write something large enough that the corruption cannot get undetected
// (even without compression)
byte[] corruption = new byte[corruptionSize];
Arrays.fill(corruption, (byte)0xFF);
random.nextBytes(corruption);
raf.write(corruption);
if (ChunkCache.instance != null)
ChunkCache.instance.invalidateFile(sstable.getFilename());

8
test/unit/org/apache/cassandra/index/sasi/SASIIndexTest.java
View File

@ -76,7 +76,12 @@ import org.junit.*;
public class SASIIndexTest
{
private static final IPartitioner PARTITIONER = new Murmur3Partitioner();
private static final IPartitioner PARTITIONER;
static {
System.setProperty("cassandra.config", "cassandra-murmur.yaml");
PARTITIONER = Murmur3Partitioner.instance;
}
private static final String KS_NAME = "sasi";
private static final String CF_NAME = "test_cf";
@ -87,7 +92,6 @@ public class SASIIndexTest
@BeforeClass
public static void loadSchema() throws ConfigurationException
{
System.setProperty("cassandra.config", "cassandra-murmur.yaml");
SchemaLoader.loadSchema();
MigrationManager.announceNewKeyspace(KeyspaceMetadata.create(KS_NAME,
KeyspaceParams.simpleTransient(1),

247
test/unit/org/apache/cassandra/io/sstable/SSTableCorruptionDetectionTest.java Normal file
View File

@ -0,0 +1,247 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.io.sstable;
import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.function.*;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.*;
import org.apache.cassandra.cache.*;
import org.apache.cassandra.config.*;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.compaction.OperationType;
import org.apache.cassandra.db.filter.*;
import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
import org.apache.cassandra.db.marshal.*;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.sstable.format.SSTableWriter;
import org.apache.cassandra.io.util.*;
import org.apache.cassandra.schema.*;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
public class SSTableCorruptionDetectionTest extends SSTableWriterTestBase
{
private static final Logger logger = LoggerFactory.getLogger(SSTableCorruptionDetectionTest.class);
private static final int numberOfPks = 1000;
private static final int numberOfRuns = 100;
private static final int valueSize = 512 * 1024;
// Set corruption size larger or in comparable size to value size, otherwise
// chance for corruption to land in the middle of value is quite high.
private static final int maxCorruptionSize = 2 * 1024 * 1024;
private static final String keyspace = "SSTableCorruptionDetectionTest";
private static final String table = "corrupted_table";
private static Random random;
private static SSTableWriter writer;
private static LifecycleTransaction txn;
private static ColumnFamilyStore cfs;
private static SSTableReader ssTableReader;
@BeforeClass
public static void setUp()
{
CFMetaData cfm = CFMetaData.Builder.create(keyspace, table)
.addPartitionKey("pk", AsciiType.instance)
.addClusteringColumn("ck1", AsciiType.instance)
.addClusteringColumn("ck2", AsciiType.instance)
.addRegularColumn("reg1", BytesType.instance)
.addRegularColumn("reg2", BytesType.instance)
.build();
cfm.compression(CompressionParams.noCompression());
SchemaLoader.createKeyspace(keyspace,
KeyspaceParams.simple(1),
cfm);
cfs = Keyspace.open(keyspace).getColumnFamilyStore(table);
cfs.disableAutoCompaction();
for (ColumnDefinition cd : cfs.metadata.allColumns())
cd.type.setMaxValueSize(1024 * 1024);
long seed = System.nanoTime();
logger.info("Seed {}", seed);
random = new Random(seed);
truncate(cfs);
File dir = cfs.getDirectories().getDirectoryForNewSSTables();
txn = LifecycleTransaction.offline(OperationType.WRITE);
// Setting up/writing large values is an expensive operation, we only want to do it once per run
writer = getWriter(cfs, dir, txn);
for (int i = 0; i < numberOfPks; i++)
{
UpdateBuilder builder = UpdateBuilder.create(cfs.metadata, String.format("pkvalue_%07d", i)).withTimestamp(1);
byte[] reg1 = new byte[valueSize];
random.nextBytes(reg1);
byte[] reg2 = new byte[valueSize];
random.nextBytes(reg2);
builder.newRow("clustering_" + i, "clustering_" + (i + 1))
.add("reg1", ByteBuffer.wrap(reg1))
.add("reg2", ByteBuffer.wrap(reg2));
writer.append(builder.build().unfilteredIterator());
}
cfs.forceBlockingFlush();
ssTableReader = writer.finish(true);
txn.update(ssTableReader, false);
LifecycleTransaction.waitForDeletions();
}
@AfterClass
public static void tearDown()
{
txn.abort();
writer.close();
}
@Test
public void testSinglePartitionIterator() throws Throwable
{
bruteForceCorruptionTest(ssTableReader, partitionIterator());
}
@Test
public void testSSTableScanner() throws Throwable
{
bruteForceCorruptionTest(ssTableReader, sstableScanner());
}
private void bruteForceCorruptionTest(SSTableReader ssTableReader, Consumer<SSTableReader> walker) throws Throwable
{
RandomAccessFile raf = new RandomAccessFile(ssTableReader.getFilename(), "rw");
int corruptedCounter = 0;
int fileLength = (int)raf.length(); // in current test, it does fit into int
for (int i = 0; i < numberOfRuns; i++)
{
final int corruptionPosition = random.nextInt(fileLength - 1); //ensure at least one byte will be corrupted
// corrupt max from position to end of file
final int corruptionSize = Math.min(maxCorruptionSize, random.nextInt(fileLength - corruptionPosition));
byte[] backup = corruptSstable(raf, corruptionPosition, corruptionSize);
try
{
walker.accept(ssTableReader);
}
catch (CorruptSSTableException t)
{
corruptedCounter++;
}
finally
{
if (ChunkCache.instance != null)
ChunkCache.instance.invalidateFile(ssTableReader.getFilename());
restore(raf, corruptionPosition, backup);
}
}
assertTrue(corruptedCounter > 0);
FileUtils.closeQuietly(raf);
}
private Consumer<SSTableReader> sstableScanner()
{
return (SSTableReader sstable) -> {
try (ISSTableScanner scanner = sstable.getScanner())
{
while (scanner.hasNext())
{
try (UnfilteredRowIterator rowIter = scanner.next())
{
if (rowIter.hasNext())
{
Unfiltered unfiltered = rowIter.next();
if (unfiltered.isRow())
{
Row row = (Row) unfiltered;
assertEquals(2, row.clustering().size());
// no-op read
}
}
}
}
}
};
}
private Consumer<SSTableReader> partitionIterator()
{
return (SSTableReader sstable) -> {
for (int i = 0; i < numberOfPks; i++)
{
DecoratedKey dk = Util.dk(String.format("pkvalue_%07d", i));
try (UnfilteredRowIterator rowIter = sstable.iterator(dk, Slices.ALL, ColumnFilter.all(cfs.metadata), false, false))
{
while (rowIter.hasNext())
{
Unfiltered unfiltered = rowIter.next();
if (unfiltered.isRow())
{
Row row = (Row) unfiltered;
assertEquals(2, row.clustering().size());
// no-op read
}
}
rowIter.close();
}
}
};
}
private byte[] corruptSstable(RandomAccessFile raf, int position, int corruptionSize) throws IOException
{
byte[] backup = new byte[corruptionSize];
raf.seek(position);
raf.read(backup);
raf.seek(position);
byte[] corruption = new byte[corruptionSize];
random.nextBytes(corruption);
raf.write(corruption);
return backup;
}
private void restore(RandomAccessFile raf, int position, byte[] backup) throws IOException
{
raf.seek(position);
raf.write(backup);
}
}

49
test/unit/org/apache/cassandra/io/sstable/SSTableWriterTest.java
View File

@ -23,15 +23,17 @@ import java.nio.ByteBuffer;
import org.junit.Test;
import org.apache.cassandra.UpdateBuilder;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.*;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.compaction.OperationType;
import org.apache.cassandra.db.filter.*;
import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.sstable.format.SSTableWriter;
import org.apache.cassandra.utils.FBUtilities;
import static junit.framework.Assert.fail;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@ -197,4 +199,45 @@ public class SSTableWriterTest extends SSTableWriterTestBase
writer2.close();
}
}
@Test
public void testValueTooBigCorruption() throws InterruptedException
{
Keyspace keyspace = Keyspace.open(KEYSPACE);
ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF_SMALL_MAX_VALUE);
truncate(cfs);
File dir = cfs.getDirectories().getDirectoryForNewSSTables();
LifecycleTransaction txn = LifecycleTransaction.offline(OperationType.STREAM);
try (SSTableWriter writer1 = getWriter(cfs, dir, txn))
{
UpdateBuilder largeValue = UpdateBuilder.create(cfs.metadata, "large_value").withTimestamp(1);
largeValue.newRow("clustering").add("val", ByteBuffer.allocate(2 * 1024 * 1024));
writer1.append(largeValue.build().unfilteredIterator());
SSTableReader sstable = writer1.finish(true);
txn.update(sstable, false);
try
{
DecoratedKey dk = Util.dk("large_value");
UnfilteredRowIterator rowIter = sstable.iterator(dk, Slices.ALL, ColumnFilter.all(cfs.metadata), false, false);
while (rowIter.hasNext())
{
rowIter.next();
// no-op read, as values may not appear expected
}
fail("Expected a CorruptSSTableException to be thrown");
}
catch (CorruptSSTableException e)
{
}
txn.abort();
LifecycleTransaction.waitForDeletions();
}
}
}

10
test/unit/org/apache/cassandra/io/sstable/SSTableWriterTestBase.java
View File

@ -31,6 +31,7 @@ import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
@ -54,6 +55,7 @@ public class SSTableWriterTestBase extends SchemaLoader
protected static final String KEYSPACE = "SSTableRewriterTest";
protected static final String CF = "Standard1";
protected static final String CF_SMALL_MAX_VALUE = "Standard_SmallMaxValue";
private static Config.DiskAccessMode standardMode;
private static Config.DiskAccessMode indexMode;
@ -73,7 +75,13 @@ public class SSTableWriterTestBase extends SchemaLoader
SchemaLoader.prepareServer();
SchemaLoader.createKeyspace(KEYSPACE,
KeyspaceParams.simple(1),
SchemaLoader.standardCFMD(KEYSPACE, CF));
SchemaLoader.standardCFMD(KEYSPACE, CF),
SchemaLoader.standardCFMD(KEYSPACE, CF_SMALL_MAX_VALUE));
Keyspace keyspace = Keyspace.open(KEYSPACE);
ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF_SMALL_MAX_VALUE);
for (ColumnDefinition cd : cfs.metadata.allColumns())
cd.type.setMaxValueSize(1024 * 1024); // set max value size to 1MB
}
@AfterClass