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:
Benjamin Lerer 2016-06-02 12:47:03 +02:00
parent dbf03103cc 1e82695115
commit dc6ffc25a8
14 changed files with 407 additions and 225 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -26,6 +26,7 @@ Merged from 2.2:
* Prohibit Reversed Counter type as part of the PK (CASSANDRA-9395)
* cqlsh: correctly handle non-ascii chars in error messages (CASSANDRA-11626)
Merged from 2.1:
* Run CommitLog tests with different compression settings (CASSANDRA-9039)
* cqlsh: apply current keyspace to source command (CASSANDRA-11152)
* Clear out parent repair session if repair coordinator dies (CASSANDRA-11824)
* Set default streaming_socket_timeout_in_ms to 24 hours (CASSANDRA-11840)

102
src/java/org/apache/cassandra/db/commitlog/CommitLog.java
View File

@ -72,9 +72,7 @@ public class CommitLog implements CommitLogMBean
final CommitLogMetrics metrics;
final AbstractCommitLogService executor;
final ICompressor compressor;
public ParameterizedClass compressorClass;
public EncryptionContext encryptionContext;
volatile Configuration configuration;
final public String location;
private static CommitLog construct()
@ -96,13 +94,11 @@ public class CommitLog implements CommitLogMBean
@VisibleForTesting
CommitLog(String location, CommitLogArchiver archiver)
{
compressorClass = DatabaseDescriptor.getCommitLogCompression();
this.location = location;
ICompressor compressor = compressorClass != null ? CompressionParams.createCompressor(compressorClass) : null;
this.configuration = new Configuration(DatabaseDescriptor.getCommitLogCompression(),
DatabaseDescriptor.getEncryptionContext());
DatabaseDescriptor.createAllDirectories();
encryptionContext = DatabaseDescriptor.getEncryptionContext();
this.compressor = compressor;
this.archiver = archiver;
metrics = new CommitLogMetrics();
@ -146,7 +142,8 @@ public class CommitLog implements CommitLogMBean
};
// submit all existing files in the commit log dir for archiving prior to recovery - CASSANDRA-6904
for (File file : new File(DatabaseDescriptor.getCommitLogLocation()).listFiles(unmanagedFilesFilter))
File[] listFiles = new File(DatabaseDescriptor.getCommitLogLocation()).listFiles(unmanagedFilesFilter);
for (File file : listFiles)
{
archiver.maybeArchive(file.getPath(), file.getName());
archiver.maybeWaitForArchiving(file.getName());
@ -416,9 +413,19 @@ public class CommitLog implements CommitLogMBean
public int resetUnsafe(boolean deleteSegments) throws IOException
{
stopUnsafe(deleteSegments);
resetConfiguration();
return restartUnsafe();
}
/**
* FOR TESTING PURPOSES.
*/
public void resetConfiguration()
{
configuration = new Configuration(DatabaseDescriptor.getCommitLogCompression(),
DatabaseDescriptor.getEncryptionContext());
}
/**
* FOR TESTING PURPOSES. See CommitLogAllocator.
*/
@ -492,4 +499,83 @@ public class CommitLog implements CommitLogMBean
throw new AssertionError(DatabaseDescriptor.getCommitFailurePolicy());
}
}
public static final class Configuration
{
/**
* The compressor class.
*/
private final ParameterizedClass compressorClass;
/**
* The compressor used to compress the segments.
*/
private final ICompressor compressor;
/**
* The encryption context used to encrypt the segments.
*/
private EncryptionContext encryptionContext;
public Configuration(ParameterizedClass compressorClass, EncryptionContext encryptionContext)
{
this.compressorClass = compressorClass;
this.compressor = compressorClass != null ? CompressionParams.createCompressor(compressorClass) : null;
this.encryptionContext = encryptionContext;
}
/**
* Checks if the segments must be compressed.
* @return <code>true</code> if the segments must be compressed, <code>false</code> otherwise.
*/
public boolean useCompression()
{
return compressor != null;
}
/**
* Checks if the segments must be encrypted.
* @return <code>true</code> if the segments must be encrypted, <code>false</code> otherwise.
*/
public boolean useEncryption()
{
return encryptionContext.isEnabled();
}
/**
* Returns the compressor used to compress the segments.
* @return the compressor used to compress the segments
*/
public ICompressor getCompressor()
{
return compressor;
}
/**
* Returns the compressor class.
* @return the compressor class
*/
public ParameterizedClass getCompressorClass()
{
return compressorClass;
}
/**
* Returns the compressor name.
* @return the compressor name.
*/
public String getCompressorName()
{
return useCompression() ? compressor.getClass().getSimpleName() : "none";
}
/**
* Returns the encryption context used to encrypt the segments.
* @return the encryption context used to encrypt the segments
*/
public EncryptionContext getEncryptionContext()
{
return encryptionContext;
}
}
}

15
src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java
View File

@ -46,6 +46,7 @@ import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.commitlog.CommitLog.Configuration;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.io.FSWriteError;
import org.apache.cassandra.io.util.FileUtils;
@ -122,9 +123,10 @@ public abstract class CommitLogSegment
static CommitLogSegment createSegment(CommitLog commitLog, Runnable onClose)
{
CommitLogSegment segment = commitLog.encryptionContext.isEnabled() ? new EncryptedSegment(commitLog, commitLog.encryptionContext, onClose) :
commitLog.compressor != null ? new CompressedSegment(commitLog, onClose) :
new MemoryMappedSegment(commitLog);
Configuration config = commitLog.configuration;
CommitLogSegment segment = config.useEncryption() ? new EncryptedSegment(commitLog, onClose)
: config.useCompression() ? new CompressedSegment(commitLog, onClose)
: new MemoryMappedSegment(commitLog);
segment.writeLogHeader();
return segment;
}
@ -137,7 +139,8 @@ public abstract class CommitLogSegment
*/
static boolean usesBufferPool(CommitLog commitLog)
{
return commitLog.encryptionContext.isEnabled() || commitLog.compressor != null;
Configuration config = commitLog.configuration;
return config.useEncryption() || config.useCompression();
}
static long getNextId()
@ -152,7 +155,9 @@ public abstract class CommitLogSegment
{
this.commitLog = commitLog;
id = getNextId();
descriptor = new CommitLogDescriptor(id, commitLog.compressorClass, commitLog.encryptionContext);
descriptor = new CommitLogDescriptor(id,
commitLog.configuration.getCompressorClass(),
commitLog.configuration.getEncryptionContext());
logFile = new File(commitLog.location, descriptor.fileName());
try

17
src/java/org/apache/cassandra/db/commitlog/CommitLogSegmentManager.java
View File

@ -21,11 +21,9 @@ import java.io.File;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentLinkedQueue;
@ -488,13 +486,16 @@ public class CommitLogSegmentManager
throw new RuntimeException(e);
}
for (CommitLogSegment segment : activeSegments)
closeAndDeleteSegmentUnsafe(segment, deleteSegments);
activeSegments.clear();
synchronized (this)
{
for (CommitLogSegment segment : activeSegments)
closeAndDeleteSegmentUnsafe(segment, deleteSegments);
activeSegments.clear();
for (CommitLogSegment segment : availableSegments)
closeAndDeleteSegmentUnsafe(segment, deleteSegments);
availableSegments.clear();
for (CommitLogSegment segment : availableSegments)
closeAndDeleteSegmentUnsafe(segment, deleteSegments);
availableSegments.clear();
}
allocatingFrom = null;

4
src/java/org/apache/cassandra/db/commitlog/CompressedSegment.java
View File

@ -47,7 +47,7 @@ public class CompressedSegment extends FileDirectSegment
CompressedSegment(CommitLog commitLog, Runnable onClose)
{
super(commitLog, onClose);
this.compressor = commitLog.compressor;
this.compressor = commitLog.configuration.getCompressor();
}
ByteBuffer allocate(int size)
@ -57,7 +57,7 @@ public class CompressedSegment extends FileDirectSegment
ByteBuffer createBuffer(CommitLog commitLog)
{
return createBuffer(commitLog.compressor.preferredBufferType());
return createBuffer(commitLog.configuration.getCompressor().preferredBufferType());
}
@Override

4
src/java/org/apache/cassandra/db/commitlog/EncryptedSegment.java
View File

@ -65,10 +65,10 @@ public class EncryptedSegment extends FileDirectSegment
private final EncryptionContext encryptionContext;
private final Cipher cipher;
public EncryptedSegment(CommitLog commitLog, EncryptionContext encryptionContext, Runnable onClose)
public EncryptedSegment(CommitLog commitLog, Runnable onClose)
{
super(commitLog, onClose);
this.encryptionContext = encryptionContext;
this.encryptionContext = commitLog.configuration.getEncryptionContext();
try
{

12
test/long/org/apache/cassandra/db/commitlog/CommitLogStressTest.java
View File

@ -229,8 +229,8 @@ public class CommitLogStressTest
public void testLog(CommitLog commitLog) throws IOException, InterruptedException {
System.out.format("\nTesting commit log size %.0fmb, compressor: %s, encryption enabled: %b, sync %s%s%s\n",
mb(DatabaseDescriptor.getCommitLogSegmentSize()),
commitLog.compressor != null ? commitLog.compressor.getClass().getSimpleName() : "none",
commitLog.encryptionContext.isEnabled(),
commitLog.configuration.getCompressorName(),
commitLog.configuration.useEncryption(),
commitLog.executor.getClass().getSimpleName(),
randomSize ? " random size" : "",
discardedRun ? " with discarded run" : "");
@ -295,14 +295,14 @@ public class CommitLogStressTest
if (hash == repl.hash && cells == repl.cells)
System.out.format("Test success. compressor = %s, encryption enabled = %b; discarded = %d, skipped = %d\n",
commitLog.compressor != null ? commitLog.compressor.getClass().getSimpleName() : "none",
commitLog.encryptionContext.isEnabled(),
commitLog.configuration.getCompressorName(),
commitLog.configuration.useEncryption(),
repl.discarded, repl.skipped);
else
{
System.out.format("Test failed (compressor = %s, encryption enabled = %b). Cells %d, expected %d, diff %d; discarded = %d, skipped = %d - hash %d expected %d.\n",
commitLog.compressor != null ? commitLog.compressor.getClass().getSimpleName() : "none",
commitLog.encryptionContext.isEnabled(),
commitLog.configuration.getCompressorName(),
commitLog.configuration.useEncryption(),
repl.cells, cells, cells - repl.cells, repl.discarded, repl.skipped,
repl.hash, hash);
failed = true;

40
test/unit/org/apache/cassandra/db/RecoveryManagerFlushedTest.java
View File

@ -18,20 +18,37 @@
*/
package org.apache.cassandra.db;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameters;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.ParameterizedClass;
import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.compress.DeflateCompressor;
import org.apache.cassandra.io.compress.LZ4Compressor;
import org.apache.cassandra.io.compress.SnappyCompressor;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.schema.SchemaKeyspace;
import org.apache.cassandra.security.EncryptionContext;
import org.apache.cassandra.security.EncryptionContextGenerator;
import org.apache.cassandra.utils.FBUtilities;
@RunWith(Parameterized.class)
public class RecoveryManagerFlushedTest
{
private static Logger logger = LoggerFactory.getLogger(RecoveryManagerFlushedTest.class);
@ -40,6 +57,29 @@ public class RecoveryManagerFlushedTest
private static final String CF_STANDARD1 = "Standard1";
private static final String CF_STANDARD2 = "Standard2";
public RecoveryManagerFlushedTest(ParameterizedClass commitLogCompression, EncryptionContext encryptionContext)
{
DatabaseDescriptor.setCommitLogCompression(commitLogCompression);
DatabaseDescriptor.setEncryptionContext(encryptionContext);
}
@Parameters()
public static Collection<Object[]> generateData()
{
return Arrays.asList(new Object[][]{
{null, EncryptionContextGenerator.createDisabledContext()}, // No compression, no encryption
{null, EncryptionContextGenerator.createContext(true)}, // Encryption
{new ParameterizedClass(LZ4Compressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()},
{new ParameterizedClass(SnappyCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()},
{new ParameterizedClass(DeflateCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}});
}
@Before
public void setUp() throws IOException
{
CommitLog.instance.resetUnsafe(true);
}
@BeforeClass
public static void defineSchema() throws ConfigurationException
{

38
test/unit/org/apache/cassandra/db/RecoveryManagerMissingHeaderTest.java
View File

@ -20,21 +20,34 @@ package org.apache.cassandra.db;
import java.io.File;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameters;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.Util;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.rows.AbstractUnfilteredRowIterator;
import org.apache.cassandra.config.ParameterizedClass;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.compress.DeflateCompressor;
import org.apache.cassandra.io.compress.LZ4Compressor;
import org.apache.cassandra.io.compress.SnappyCompressor;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.security.EncryptionContext;
import org.apache.cassandra.security.EncryptionContextGenerator;
@RunWith(Parameterized.class)
public class RecoveryManagerMissingHeaderTest
{
private static final String KEYSPACE1 = "RecoveryManager3Test1";
@ -43,6 +56,29 @@ public class RecoveryManagerMissingHeaderTest
private static final String KEYSPACE2 = "RecoveryManager3Test2";
private static final String CF_STANDARD3 = "Standard3";
public RecoveryManagerMissingHeaderTest(ParameterizedClass commitLogCompression, EncryptionContext encryptionContext)
{
DatabaseDescriptor.setCommitLogCompression(commitLogCompression);
DatabaseDescriptor.setEncryptionContext(encryptionContext);
}
@Parameters()
public static Collection<Object[]> generateData()
{
return Arrays.asList(new Object[][]{
{null, EncryptionContextGenerator.createDisabledContext()}, // No compression, no encryption
{null, EncryptionContextGenerator.createContext(true)}, // Encryption
{new ParameterizedClass(LZ4Compressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()},
{new ParameterizedClass(SnappyCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()},
{new ParameterizedClass(DeflateCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}});
}
@Before
public void setUp() throws IOException
{
CommitLog.instance.resetUnsafe(true);
}
@BeforeClass
public static void defineSchema() throws ConfigurationException
{

167
test/unit/org/apache/cassandra/db/RecoveryManagerTest.java
View File

@ -19,6 +19,9 @@
package org.apache.cassandra.db;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
@ -30,18 +33,24 @@ import java.util.concurrent.atomic.AtomicReference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.OrderedJUnit4ClassRunner;
import org.apache.cassandra.Util;
import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.ParameterizedClass;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.db.context.CounterContext;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.compress.DeflateCompressor;
import org.apache.cassandra.io.compress.LZ4Compressor;
import org.apache.cassandra.io.compress.SnappyCompressor;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameters;
import static org.junit.Assert.assertEquals;
@ -49,72 +58,15 @@ import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.db.commitlog.CommitLogArchiver;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.security.EncryptionContext;
import org.apache.cassandra.security.EncryptionContextGenerator;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.db.commitlog.CommitLogReplayer;
@RunWith(OrderedJUnit4ClassRunner.class)
@RunWith(Parameterized.class)
public class RecoveryManagerTest
{
private static Logger logger = LoggerFactory.getLogger(RecoveryManagerTest.class);
static final Semaphore blocker = new Semaphore(0);
static final Semaphore blocked = new Semaphore(0);
static CommitLogReplayer.MutationInitiator originalInitiator = null;
static final CommitLogReplayer.MutationInitiator mockInitiator = new CommitLogReplayer.MutationInitiator()
{
@Override
protected Future<Integer> initiateMutation(final Mutation mutation,
final long segmentId,
final int serializedSize,
final int entryLocation,
final CommitLogReplayer clr)
{
final Future<Integer> toWrap = super.initiateMutation(mutation,
segmentId,
serializedSize,
entryLocation,
clr);
return new Future<Integer>()
{
@Override
public boolean cancel(boolean mayInterruptIfRunning)
{
throw new UnsupportedOperationException();
}
@Override
public boolean isCancelled()
{
throw new UnsupportedOperationException();
}
@Override
public boolean isDone()
{
return blocker.availablePermits() > 0 && toWrap.isDone();
}
@Override
public Integer get() throws InterruptedException, ExecutionException
{
System.out.println("Got blocker once");
blocked.release();
blocker.acquire();
return toWrap.get();
}
@Override
public Integer get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException
{
blocked.release();
blocker.tryAcquire(1, timeout, unit);
return toWrap.get(timeout, unit);
}
};
}
};
private static final String KEYSPACE1 = "RecoveryManagerTest1";
private static final String CF_STANDARD1 = "Standard1";
@ -123,6 +75,29 @@ public class RecoveryManagerTest
private static final String KEYSPACE2 = "RecoveryManagerTest2";
private static final String CF_STANDARD3 = "Standard3";
public RecoveryManagerTest(ParameterizedClass commitLogCompression, EncryptionContext encryptionContext)
{
DatabaseDescriptor.setCommitLogCompression(commitLogCompression);
DatabaseDescriptor.setEncryptionContext(encryptionContext);
}
@Parameters()
public static Collection<Object[]> generateData()
{
return Arrays.asList(new Object[][]{
{null, EncryptionContextGenerator.createDisabledContext()}, // No compression, no encryption
{null, EncryptionContextGenerator.createContext(true)}, // Encryption
{new ParameterizedClass(LZ4Compressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()},
{new ParameterizedClass(SnappyCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()},
{new ParameterizedClass(DeflateCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}});
}
@Before
public void setUp() throws IOException
{
CommitLog.instance.resetUnsafe(true);
}
@BeforeClass
public static void defineSchema() throws ConfigurationException
{
@ -139,6 +114,7 @@ public class RecoveryManagerTest
@Before
public void clearData()
{
// clear data
Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_STANDARD1).truncateBlocking();
Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_COUNTER1).truncateBlocking();
Keyspace.open(KEYSPACE2).getColumnFamilyStore(CF_STANDARD3).truncateBlocking();
@ -156,6 +132,7 @@ public class RecoveryManagerTest
long originalMaxOutstanding = CommitLogReplayer.MAX_OUTSTANDING_REPLAY_BYTES;
CommitLogReplayer.MAX_OUTSTANDING_REPLAY_BYTES = 1;
CommitLogReplayer.MutationInitiator originalInitiator = CommitLogReplayer.mutationInitiator;
MockInitiator mockInitiator = new MockInitiator();
CommitLogReplayer.mutationInitiator = mockInitiator;
try
{
@ -194,10 +171,10 @@ public class RecoveryManagerTest
}
};
t.start();
Assert.assertTrue(blocked.tryAcquire(1, 20, TimeUnit.SECONDS));
Assert.assertTrue(mockInitiator.blocked.tryAcquire(1, 20, TimeUnit.SECONDS));
Thread.sleep(100);
Assert.assertTrue(t.isAlive());
blocker.release(Integer.MAX_VALUE);
mockInitiator.blocker.release(Integer.MAX_VALUE);
t.join(20 * 1000);
if (err.get() != null)
@ -273,8 +250,8 @@ public class RecoveryManagerTest
@Test
public void testRecoverPIT() throws Exception
{
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_STANDARD1);
CommitLog.instance.resetUnsafe(true);
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_STANDARD1);
Date date = CommitLogArchiver.format.parse("2112:12:12 12:12:12");
long timeMS = date.getTime() - 5000;
@ -301,8 +278,8 @@ public class RecoveryManagerTest
@Test
public void testRecoverPITUnordered() throws Exception
{
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_STANDARD1);
CommitLog.instance.resetUnsafe(true);
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_STANDARD1);
Date date = CommitLogArchiver.format.parse("2112:12:12 12:12:12");
long timeMS = date.getTime();
@ -332,4 +309,64 @@ public class RecoveryManagerTest
assertEquals(2, Util.getAll(Util.cmd(cfs).build()).size());
}
private static class MockInitiator extends CommitLogReplayer.MutationInitiator
{
final Semaphore blocker = new Semaphore(0);
final Semaphore blocked = new Semaphore(0);
@Override
protected Future<Integer> initiateMutation(final Mutation mutation,
final long segmentId,
final int serializedSize,
final int entryLocation,
final CommitLogReplayer clr)
{
final Future<Integer> toWrap = super.initiateMutation(mutation,
segmentId,
serializedSize,
entryLocation,
clr);
return new Future<Integer>()
{
@Override
public boolean cancel(boolean mayInterruptIfRunning)
{
throw new UnsupportedOperationException();
}
@Override
public boolean isCancelled()
{
throw new UnsupportedOperationException();
}
@Override
public boolean isDone()
{
return blocker.availablePermits() > 0 && toWrap.isDone();
}
@Override
public Integer get() throws InterruptedException, ExecutionException
{
System.out.println("Got blocker once");
blocked.release();
blocker.acquire();
return toWrap.get();
}
@Override
public Integer get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException
{
blocked.release();
blocker.tryAcquire(1, timeout, unit);
return toWrap.get(timeout, unit);
}
};
}
};
}

38
test/unit/org/apache/cassandra/db/RecoveryManagerTruncateTest.java
View File

@ -19,26 +19,64 @@
package org.apache.cassandra.db;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.Util;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.ParameterizedClass;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.compress.DeflateCompressor;
import org.apache.cassandra.io.compress.LZ4Compressor;
import org.apache.cassandra.io.compress.SnappyCompressor;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.security.EncryptionContext;
import org.apache.cassandra.security.EncryptionContextGenerator;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameters;
import static org.junit.Assert.*;
/**
* Test for the truncate operation.
*/
@RunWith(Parameterized.class)
public class RecoveryManagerTruncateTest
{
private static final String KEYSPACE1 = "RecoveryManagerTruncateTest";
private static final String CF_STANDARD1 = "Standard1";
public RecoveryManagerTruncateTest(ParameterizedClass commitLogCompression, EncryptionContext encryptionContext)
{
DatabaseDescriptor.setCommitLogCompression(commitLogCompression);
DatabaseDescriptor.setEncryptionContext(encryptionContext);
}
@Parameters()
public static Collection<Object[]> generateData()
{
return Arrays.asList(new Object[][]{
{null, EncryptionContextGenerator.createDisabledContext()}, // No compression, no encryption
{null, EncryptionContextGenerator.createContext(true)}, // Encryption
{new ParameterizedClass(LZ4Compressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()},
{new ParameterizedClass(SnappyCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()},
{new ParameterizedClass(DeflateCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}});
}
@Before
public void setUp() throws IOException
{
CommitLog.instance.resetUnsafe(true);
}
@BeforeClass
public static void defineSchema() throws ConfigurationException
{

3
test/unit/org/apache/cassandra/db/commitlog/CommitLogDescriptorTest.java
View File

@ -15,6 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.db.commitlog;
import java.io.IOException;
@ -125,6 +126,7 @@ public class CommitLogDescriptorTest
21,
new ParameterizedClass("LZ4Compressor", params),
neverEnabledEncryption);
ByteBuffer buf = ByteBuffer.allocate(1024000);
CommitLogDescriptor.writeHeader(buf, desc);
Assert.fail("Parameter object too long should fail on writing descriptor.");
@ -307,5 +309,4 @@ public class CommitLogDescriptorTest
CommitLogDescriptor desc2 = new CommitLogDescriptor(CommitLogDescriptor.current_version, 1, compression, enabledEncryption);
Assert.assertEquals(desc1, desc2);
}
}

187
test/unit/org/apache/cassandra/db/commitlog/CommitLogTest.java
View File

@ -29,12 +29,18 @@ import java.util.zip.Checksum;
import com.google.common.collect.Iterables;
import org.junit.*;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameters;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.Util;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.ParameterizedClass;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.RowUpdateBuilder;
import org.apache.cassandra.db.commitlog.CommitLogReplayer.CommitLogReplayException;
import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.db.marshal.AsciiType;
@ -52,13 +58,17 @@ import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.security.EncryptionContext;
import org.apache.cassandra.security.EncryptionContextGenerator;
import org.apache.cassandra.utils.*;
import org.apache.cassandra.utils.Hex;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.KillerForTests;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.vint.VIntCoding;
import static org.apache.cassandra.utils.ByteBufferUtil.bytes;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@RunWith(Parameterized.class)
public class CommitLogTest
{
private static final String KEYSPACE1 = "CommitLogTest";
@ -66,7 +76,22 @@ public class CommitLogTest
private static final String STANDARD1 = "Standard1";
private static final String STANDARD2 = "Standard2";
String logDirectory;
public CommitLogTest(ParameterizedClass commitLogCompression, EncryptionContext encryptionContext)
{
DatabaseDescriptor.setCommitLogCompression(commitLogCompression);
DatabaseDescriptor.setEncryptionContext(encryptionContext);
}
@Parameters()
public static Collection<Object[]> generateData()
{
return Arrays.asList(new Object[][]{
{null, EncryptionContextGenerator.createDisabledContext()}, // No compression, no encryption
{null, EncryptionContextGenerator.createContext(true)}, // Encryption
{new ParameterizedClass(LZ4Compressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()},
{new ParameterizedClass(SnappyCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()},
{new ParameterizedClass(DeflateCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}});
}
@BeforeClass
public static void defineSchema() throws ConfigurationException
@ -86,8 +111,7 @@ public class CommitLogTest
@Before
public void setup() throws IOException
{
logDirectory = DatabaseDescriptor.getCommitLogLocation() + "/unit";
new File(logDirectory).mkdirs();
CommitLog.instance.resetUnsafe(true);
}
@Test
@ -178,7 +202,6 @@ public class CommitLogTest
@Test
public void testDontDeleteIfDirty() throws Exception
{
CommitLog.instance.resetUnsafe(true);
ColumnFamilyStore cfs1 = Keyspace.open(KEYSPACE1).getColumnFamilyStore(STANDARD1);
ColumnFamilyStore cfs2 = Keyspace.open(KEYSPACE1).getColumnFamilyStore(STANDARD2);
@ -214,8 +237,6 @@ public class CommitLogTest
@Test
public void testDeleteIfNotDirty() throws Exception
{
DatabaseDescriptor.getCommitLogSegmentSize();
CommitLog.instance.resetUnsafe(true);
ColumnFamilyStore cfs1 = Keyspace.open(KEYSPACE1).getColumnFamilyStore(STANDARD1);
ColumnFamilyStore cfs2 = Keyspace.open(KEYSPACE1).getColumnFamilyStore(STANDARD2);
@ -293,7 +314,6 @@ public class CommitLogTest
@Test
public void testEqualRecordLimit() throws Exception
{
CommitLog.instance.resetUnsafe(true);
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(STANDARD1);
Mutation rm = new RowUpdateBuilder(cfs.metadata, 0, "k")
.clustering("bytes")
@ -305,7 +325,6 @@ public class CommitLogTest
@Test(expected = IllegalArgumentException.class)
public void testExceedRecordLimit() throws Exception
{
CommitLog.instance.resetUnsafe(true);
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(STANDARD1);
Mutation rm = new RowUpdateBuilder(cfs.metadata, 0, "k")
.clustering("bytes")
@ -347,22 +366,13 @@ public class CommitLogTest
DatabaseDescriptor.getCommitLogCompression(),
encryptionContext);
// if we're testing encryption, we need to write out a cipher IV to the descriptor headers
Map<String, String> additionalHeaders = new HashMap<>();
if (encryptionContext.isEnabled())
{
byte[] buf = new byte[16];
new Random().nextBytes(buf);
additionalHeaders.put(EncryptionContext.ENCRYPTION_IV, Hex.bytesToHex(buf));
}
ByteBuffer buf = ByteBuffer.allocate(1024);
CommitLogDescriptor.writeHeader(buf, desc, additionalHeaders);
CommitLogDescriptor.writeHeader(buf, desc, getAdditionalHeaders(encryptionContext));
buf.flip();
int positionAfterHeader = buf.limit() + 1;
File logFile = new File(logDirectory, desc.fileName());
logFile.deleteOnExit();
File logFile = new File(DatabaseDescriptor.getCommitLogLocation(), desc.fileName());
try (OutputStream lout = new FileOutputStream(logFile))
{
@ -372,10 +382,20 @@ public class CommitLogTest
return Pair.create(logFile, positionAfterHeader);
}
private Map<String, String> getAdditionalHeaders(EncryptionContext encryptionContext)
{
if (!encryptionContext.isEnabled())
return Collections.emptyMap();
// if we're testing encryption, we need to write out a cipher IV to the descriptor headers
byte[] buf = new byte[16];
new Random().nextBytes(buf);
return Collections.singletonMap(EncryptionContext.ENCRYPTION_IV, Hex.bytesToHex(buf));
}
protected File tmpFile(int version) throws IOException
{
File logFile = File.createTempFile("CommitLog-" + version + "-", ".log");
logFile.deleteOnExit();
assert logFile.length() == 0;
return logFile;
}
@ -399,7 +419,7 @@ public class CommitLogTest
// Change id to match file.
desc = new CommitLogDescriptor(desc.version, fromFile.id, desc.compression, desc.getEncryptionContext());
ByteBuffer buf = ByteBuffer.allocate(1024);
CommitLogDescriptor.writeHeader(buf, desc);
CommitLogDescriptor.writeHeader(buf, desc, getAdditionalHeaders(desc.getEncryptionContext()));
try (OutputStream lout = new FileOutputStream(logFile))
{
lout.write(buf.array(), 0, buf.position());
@ -440,11 +460,8 @@ public class CommitLogTest
protected void runExpecting(Callable<Void> r, Class<?> expected)
{
JVMStabilityInspector.Killer originalKiller;
KillerForTests killerForTests;
killerForTests = new KillerForTests();
originalKiller = JVMStabilityInspector.replaceKiller(killerForTests);
KillerForTests killerForTests = new KillerForTests();
JVMStabilityInspector.Killer originalKiller = JVMStabilityInspector.replaceKiller(killerForTests);
Throwable caught = null;
try
@ -466,8 +483,10 @@ public class CommitLogTest
protected void testRecovery(final byte[] logData, Class<?> expected) throws Exception
{
ParameterizedClass commitLogCompression = DatabaseDescriptor.getCommitLogCompression();
EncryptionContext encryptionContext = DatabaseDescriptor.getEncryptionContext();
runExpecting(() -> testRecovery(logData, CommitLogDescriptor.VERSION_20), expected);
runExpecting(() -> testRecovery(new CommitLogDescriptor(4, null, EncryptionContextGenerator.createDisabledContext()), logData), expected);
runExpecting(() -> testRecovery(new CommitLogDescriptor(4, commitLogCompression, encryptionContext), logData), expected);
}
protected void testRecovery(byte[] logData) throws Exception
@ -489,7 +508,6 @@ public class CommitLogTest
boolean originalState = DatabaseDescriptor.isAutoSnapshot();
try
{
CommitLog.instance.resetUnsafe(true);
boolean prev = DatabaseDescriptor.isAutoSnapshot();
DatabaseDescriptor.setAutoSnapshot(false);
ColumnFamilyStore cfs1 = Keyspace.open(KEYSPACE1).getColumnFamilyStore(STANDARD1);
@ -523,7 +541,6 @@ public class CommitLogTest
@Test
public void testTruncateWithoutSnapshotNonDurable() throws IOException
{
CommitLog.instance.resetUnsafe(true);
boolean originalState = DatabaseDescriptor.getAutoSnapshot();
try
{
@ -551,88 +568,7 @@ public class CommitLogTest
}
@Test
public void replay_StandardMmapped() throws IOException
{
ParameterizedClass originalCompression = DatabaseDescriptor.getCommitLogCompression();
EncryptionContext originalEncryptionContext = DatabaseDescriptor.getEncryptionContext();
try
{
DatabaseDescriptor.setCommitLogCompression(null);
DatabaseDescriptor.setEncryptionContext(EncryptionContextGenerator.createDisabledContext());
CommitLog.instance.resetUnsafe(true);
replaySimple(CommitLog.instance);
replayWithDiscard(CommitLog.instance);
}
finally
{
DatabaseDescriptor.setCommitLogCompression(originalCompression);
DatabaseDescriptor.setEncryptionContext(originalEncryptionContext);
CommitLog.instance.resetUnsafe(true);
}
}
@Test
public void replay_Compressed_LZ4() throws IOException
{
replay_Compressed(new ParameterizedClass(LZ4Compressor.class.getName(), Collections.<String, String>emptyMap()));
}
@Test
public void replay_Compressed_Snappy() throws IOException
{
replay_Compressed(new ParameterizedClass(SnappyCompressor.class.getName(), Collections.<String, String>emptyMap()));
}
@Test
public void replay_Compressed_Deflate() throws IOException
{
replay_Compressed(new ParameterizedClass(DeflateCompressor.class.getName(), Collections.<String, String>emptyMap()));
}
private void replay_Compressed(ParameterizedClass parameterizedClass) throws IOException
{
ParameterizedClass originalCompression = DatabaseDescriptor.getCommitLogCompression();
EncryptionContext originalEncryptionContext = DatabaseDescriptor.getEncryptionContext();
try
{
DatabaseDescriptor.setCommitLogCompression(parameterizedClass);
DatabaseDescriptor.setEncryptionContext(EncryptionContextGenerator.createDisabledContext());
CommitLog.instance.resetUnsafe(true);
replaySimple(CommitLog.instance);
replayWithDiscard(CommitLog.instance);
}
finally
{
DatabaseDescriptor.setCommitLogCompression(originalCompression);
DatabaseDescriptor.setEncryptionContext(originalEncryptionContext);
CommitLog.instance.resetUnsafe(true);
}
}
@Test
public void replay_Encrypted() throws IOException
{
ParameterizedClass originalCompression = DatabaseDescriptor.getCommitLogCompression();
EncryptionContext originalEncryptionContext = DatabaseDescriptor.getEncryptionContext();
try
{
DatabaseDescriptor.setCommitLogCompression(null);
DatabaseDescriptor.setEncryptionContext(EncryptionContextGenerator.createContext(true));
CommitLog.instance.resetUnsafe(true);
replaySimple(CommitLog.instance);
replayWithDiscard(CommitLog.instance);
}
finally
{
DatabaseDescriptor.setCommitLogCompression(originalCompression);
DatabaseDescriptor.setEncryptionContext(originalEncryptionContext);
CommitLog.instance.resetUnsafe(true);
}
}
private void replaySimple(CommitLog commitLog) throws IOException
public void replaySimple() throws IOException
{
int cellCount = 0;
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(STANDARD1);
@ -641,28 +577,29 @@ public class CommitLogTest
.add("val", bytes("this is a string"))
.build();
cellCount += 1;
commitLog.add(rm1);
CommitLog.instance.add(rm1);
final Mutation rm2 = new RowUpdateBuilder(cfs.metadata, 0, "k2")
.clustering("bytes")
.add("val", bytes("this is a string"))
.build();
cellCount += 1;
commitLog.add(rm2);
CommitLog.instance.add(rm2);
commitLog.sync(true);
CommitLog.instance.sync(true);
Replayer replayer = new Replayer(commitLog, ReplayPosition.NONE);
List<String> activeSegments = commitLog.getActiveSegmentNames();
Replayer replayer = new Replayer(CommitLog.instance, ReplayPosition.NONE);
List<String> activeSegments = CommitLog.instance.getActiveSegmentNames();
Assert.assertFalse(activeSegments.isEmpty());
File[] files = new File(commitLog.location).listFiles((file, name) -> activeSegments.contains(name));
File[] files = new File(CommitLog.instance.location).listFiles((file, name) -> activeSegments.contains(name));
replayer.recover(files);
assertEquals(cellCount, replayer.cells);
}
private void replayWithDiscard(CommitLog commitLog) throws IOException
@Test
public void replayWithDiscard() throws IOException
{
int cellCount = 0;
int max = 1024;
@ -676,7 +613,7 @@ public class CommitLogTest
.clustering("bytes")
.add("val", bytes("this is a string"))
.build();
ReplayPosition position = commitLog.add(rm1);
ReplayPosition position = CommitLog.instance.add(rm1);
if (i == discardPosition)
replayPosition = position;
@ -686,13 +623,13 @@ public class CommitLogTest
}
}
commitLog.sync(true);
CommitLog.instance.sync(true);
Replayer replayer = new Replayer(commitLog, replayPosition);
List<String> activeSegments = commitLog.getActiveSegmentNames();
Replayer replayer = new Replayer(CommitLog.instance, replayPosition);
List<String> activeSegments = CommitLog.instance.getActiveSegmentNames();
Assert.assertFalse(activeSegments.isEmpty());
File[] files = new File(commitLog.location).listFiles((file, name) -> activeSegments.contains(name));
File[] files = new File(CommitLog.instance.location).listFiles((file, name) -> activeSegments.contains(name));
replayer.recover(files);
assertEquals(cellCount, replayer.cells);

4
test/unit/org/apache/cassandra/db/commitlog/CommitLogUpgradeTestMaker.java
View File

@ -102,8 +102,8 @@ public class CommitLogUpgradeTestMaker
CommitLog commitLog = CommitLog.instance;
System.out.format("\nUsing commit log size: %dmb, compressor: %s, encryption: %s, sync: %s, %s\n",
mb(DatabaseDescriptor.getCommitLogSegmentSize()),
commitLog.compressor != null ? commitLog.compressor.getClass().getSimpleName() : "none",
commitLog.encryptionContext.isEnabled() ? "enabled" : "none",
commitLog.configuration.getCompressorName(),
commitLog.configuration.useEncryption(),
commitLog.executor.getClass().getSimpleName(),
randomSize ? "random size" : "");
final List<CommitlogExecutor> threads = new ArrayList<>();