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

Add non-blocking mode for CDC writes 

patch by Yifan Cai; reviewed by Josh McKenzie for CASSANDRA-17001
This commit is contained in:
Yifan Cai 2021-12-03 12:18:31 -08:00
parent 7871a28542
commit d9460a04da
12 changed files with 327 additions and 156 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
4.1
* Add non-blocking mode for CDC writes (CASSANDRA-17001)
* Add guardrails framework (CASSANDRA-17147)
* Harden resource management on SSTable components to prevent future leaks (CASSANDRA-17174)
* Make nodes more resilient to local unrelated files during startup (CASSANDRA-17082)

4
NEWS.txt
View File

@ -38,6 +38,10 @@ using the provided 'sstableupgrade' tool.
New features
------------
- CDC data flushing now can be configured to be non-blocking with the configuration cdc_block_writes. Setting to true,
any writes to the CDC-enabled tables will be blocked when reaching to the limit for CDC data on disk, which is the
existing and the default behavior. Setting to false, the writes to the CDC-enabled tables will be accepted and
the oldest CDC data on disk will be deleted to ensure the size constraint.
- New native functions to convert unix time values into C* native types: toDate(bigint), toTimestamp(bigint),
mintimeuuid(bigint) and maxtimeuuid(bigint)
- Support for multiple permission in a single GRANT/REVOKE/LIST statement has been added. It allows to

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

@ -281,6 +281,9 @@ public class Config
// Change-data-capture logs
public boolean cdc_enabled = false;
// When true, new CDC mutations are rejected/blocked when reaching max CDC storage.
// When false, new CDC mutations can always be added. But it will remove the oldest CDC commit log segment on full.
public volatile boolean cdc_block_writes = true;
public String cdc_raw_directory;
public int cdc_total_space_in_mb = 0;
public int cdc_free_space_check_interval_ms = 250;

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

@ -2095,6 +2095,13 @@ public class DatabaseDescriptor
return (int) ByteUnit.MEBI_BYTES.toBytes(conf.commitlog_segment_size_in_mb);
}
/**
* Update commitlog_segment_size_in_mb in the tests.
* {@link CommitLogSegmentManagerCDC} uses the CommitLogSegmentSize to estimate the file size on allocation.
* It is important to keep the value unchanged for the estimation to be correct.
* @param sizeMegabytes
*/
@VisibleForTesting /* Only for testing */
public static void setCommitLogSegmentSize(int sizeMegabytes)
{
conf.commitlog_segment_size_in_mb = sizeMegabytes;
@ -3157,6 +3164,16 @@ public class DatabaseDescriptor
conf.cdc_enabled = cdc_enabled;
}
public static boolean getCDCBlockWrites()
{
return conf.cdc_block_writes;
}
public static void setCDCBlockWrites(boolean val)
{
conf.cdc_block_writes = val;
}
public static String getCDCLogLocation()
{
return conf.cdc_raw_directory;

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

@ -28,6 +28,7 @@ import java.util.zip.CRC32;
import com.google.common.annotations.VisibleForTesting;
import org.apache.cassandra.io.util.File;
import com.google.common.base.Preconditions;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -420,6 +421,29 @@ public class CommitLog implements CommitLogMBean
return segmentRatios;
}
@Override
public boolean getCDCBlockWrites()
{
return DatabaseDescriptor.getCDCBlockWrites();
}
@Override
public void setCDCBlockWrites(boolean val)
{
Preconditions.checkState(DatabaseDescriptor.isCDCEnabled(),
"Unable to set block_writes (%s): CDC is not enabled.", val);
Preconditions.checkState(segmentManager instanceof CommitLogSegmentManagerCDC,
"CDC is enabled but we have the wrong CommitLogSegmentManager type: %s. " +
"Please report this as bug.", segmentManager.getClass().getName());
boolean oldVal = DatabaseDescriptor.getCDCBlockWrites();
CommitLogSegment currentSegment = segmentManager.allocatingFrom();
// Update the current segment CDC state to PERMITTED if block_writes is disabled now, and it was in FORBIDDEN state
if (!val && currentSegment.getCDCState() == CommitLogSegment.CDCState.FORBIDDEN)
currentSegment.setCDCState(CommitLogSegment.CDCState.PERMITTED);
DatabaseDescriptor.setCDCBlockWrites(val);
logger.info("Updated CDC block_writes from {} to {}", oldVal, val);
}
/**
* Shuts down the threads used by the commit log, blocking until completion.
* TODO this should accept a timeout, and throw TimeoutException

18
src/java/org/apache/cassandra/db/commitlog/CommitLogDescriptor.java
View File

@ -35,6 +35,7 @@ import java.util.zip.CRC32;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Objects;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.ParameterizedClass;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.FSReadError;
@ -51,6 +52,7 @@ public class CommitLogDescriptor
private static final String SEPARATOR = "-";
private static final String FILENAME_PREFIX = "CommitLog" + SEPARATOR;
private static final String FILENAME_EXTENSION = ".log";
private static final String INDEX_FILENAME_SUFFIX = "_cdc.idx";
// match both legacy and new version of commitlogs Ex: CommitLog-12345.log and CommitLog-4-12345.log.
private static final Pattern COMMIT_LOG_FILE_PATTERN = Pattern.compile(FILENAME_PREFIX + "((\\d+)(" + SEPARATOR + "\\d+)?)" + FILENAME_EXTENSION);
@ -220,7 +222,21 @@ public class CommitLogDescriptor
public String cdcIndexFileName()
{
return FILENAME_PREFIX + version + SEPARATOR + id + "_cdc.idx";
return FILENAME_PREFIX + version + SEPARATOR + id + INDEX_FILENAME_SUFFIX;
}
/**
* Infer the corresponding cdc index file using its cdc commitlog file
* @param cdcCommitLogSegment
* @return cdc index file or null if the cdc index file cannot be inferred.
*/
public static File inferCdcIndexFile(File cdcCommitLogSegment)
{
if (!isValid(cdcCommitLogSegment.name()))
return null;
String cdcFileName = cdcCommitLogSegment.name();
String indexFileName = cdcFileName.substring(0, cdcFileName.length() - FILENAME_EXTENSION.length()) + INDEX_FILENAME_SUFFIX;
return new File(DatabaseDescriptor.getCDCLogLocation(), indexFileName);
}
/**

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

@ -84,4 +84,8 @@ public interface CommitLogMBean
* @return A map between active log segments and the compression ratio achieved for each.
*/
public Map<String, Double> getActiveSegmentCompressionRatios();
public boolean getCDCBlockWrites();
public void setCDCBlockWrites(boolean val);
}

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

@ -69,7 +69,7 @@ public abstract class CommitLogSegment
FORBIDDEN,
CONTAINS
}
Object cdcStateLock = new Object();
final Object cdcStateLock = new Object();
private final static AtomicInteger nextId = new AtomicInteger(1);
private static long replayLimitId;

121
src/java/org/apache/cassandra/db/commitlog/CommitLogSegmentManagerCDC.java
View File

@ -19,13 +19,16 @@
package org.apache.cassandra.db.commitlog;
import java.io.IOException;
import java.nio.file.FileVisitResult;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.attribute.BasicFileAttributes;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.RateLimiter;
import org.apache.cassandra.io.util.File;
import org.slf4j.Logger;
@ -67,22 +70,54 @@ public class CommitLogSegmentManagerCDC extends AbstractCommitLogSegmentManager
cdcSizeTracker.processDiscardedSegment(segment);
if (delete)
FileUtils.deleteWithConfirm(segment.logFile);
segment.logFile.delete();
if (segment.getCDCState() != CDCState.CONTAINS)
{
// Always delete hard-link from cdc folder if this segment didn't contain CDC data. Note: File may not exist
// if processing discard during startup.
File cdcLink = segment.getCDCFile();
if (cdcLink.exists())
FileUtils.deleteWithConfirm(cdcLink);
File cdcIndexFile = segment.getCDCIndexFile();
if (cdcIndexFile.exists())
FileUtils.deleteWithConfirm(cdcIndexFile);
deleteCDCFiles(cdcLink, cdcIndexFile);
}
}
/**
* Delete the oldest hard-linked CDC commit log segment to free up space.
* @return total deleted file size in bytes
*/
public long deleteOldestLinkedCDCCommitLogSegment()
{
File cdcDir = new File(DatabaseDescriptor.getCDCLogLocation());
Preconditions.checkState(cdcDir.isDirectory(), "The CDC directory does not exist.");
File[] files = cdcDir.tryList(f -> CommitLogDescriptor.isValid(f.name()));
Preconditions.checkState(files != null && files.length > 0,
"There should be at least 1 CDC commit log segment.");
List<File> sorted = Arrays.stream(files)
.sorted(Comparator.comparingLong(File::lastModified))
.collect(Collectors.toList());
File oldestCdcFile = sorted.get(0);
File cdcIndexFile = CommitLogDescriptor.inferCdcIndexFile(oldestCdcFile);
return deleteCDCFiles(oldestCdcFile, cdcIndexFile);
}
private long deleteCDCFiles(File cdcLink, File cdcIndexFile)
{
long total = 0;
if (cdcLink != null && cdcLink.exists())
{
total += cdcLink.length();
cdcLink.delete();
}
if (cdcIndexFile != null && cdcIndexFile.exists())
{
total += cdcIndexFile.length();
cdcIndexFile.delete();
}
return total;
}
/**
* Initiates the shutdown process for the management thread. Also stops the cdc on-disk size calculator executor.
*/
@ -170,7 +205,7 @@ public class CommitLogSegmentManagerCDC extends AbstractCommitLogSegmentManager
if (cdcFile.exists() && !cdcIndexFile.exists())
{
logger.trace("(Unopened) CDC segment {} is no longer needed and will be deleted now", cdcFile);
FileUtils.deleteWithConfirm(cdcFile);
cdcFile.delete();
}
}
@ -193,15 +228,15 @@ public class CommitLogSegmentManagerCDC extends AbstractCommitLogSegmentManager
{
private final RateLimiter rateLimiter = RateLimiter.create(1000.0 / DatabaseDescriptor.getCDCDiskCheckInterval());
private ExecutorService cdcSizeCalculationExecutor;
private CommitLogSegmentManagerCDC segmentManager;
// Used instead of size during walk to remove chance of over-allocation
private volatile long sizeInProgress = 0;
private final CommitLogSegmentManagerCDC segmentManager;
// track the total size between two dictionary size calculations
private final AtomicLong sizeInProgress;
CDCSizeTracker(CommitLogSegmentManagerCDC segmentManager, File path)
{
super(path);
this.segmentManager = segmentManager;
this.sizeInProgress = new AtomicLong(0);
}
/**
@ -209,7 +244,7 @@ public class CommitLogSegmentManagerCDC extends AbstractCommitLogSegmentManager
*/
public void start()
{
size = 0;
sizeInProgress.getAndSet(0);
cdcSizeCalculationExecutor = executorFactory().configureSequential("CDCSizeCalculationExecutor")
.withRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy())
.withQueueLimit(0)
@ -221,7 +256,7 @@ public class CommitLogSegmentManagerCDC extends AbstractCommitLogSegmentManager
* Synchronous size recalculation on each segment creation/deletion call could lead to very long delays in new
* segment allocation, thus long delays in thread signaling to wake waiting allocation / writer threads.
*
* This can be reached either from the segment management thread in ABstractCommitLogSegmentManager or from the
* This can be reached either from the segment management thread in AbstractCommitLogSegmentManager or from the
* size recalculation executor, so we synchronize on this object to reduce the race overlap window available for
* size to get off.
*
@ -232,11 +267,26 @@ public class CommitLogSegmentManagerCDC extends AbstractCommitLogSegmentManager
// See synchronization in CommitLogSegment.setCDCState
synchronized(segment.cdcStateLock)
{
segment.setCDCState(defaultSegmentSize() + totalCDCSizeOnDisk() > allowableCDCBytes()
int segmentSize = defaultSegmentSize();
long allowance = allowableCDCBytes();
boolean blocking = DatabaseDescriptor.getCDCBlockWrites();
segment.setCDCState(blocking && segmentSize + sizeInProgress.get() > allowance
? CDCState.FORBIDDEN
: CDCState.PERMITTED);
// Remove the oldest cdc segment file when exceeding the CDC storage allowance
while (!blocking && segmentSize + sizeInProgress.get() > allowance)
{
long releasedSize = segmentManager.deleteOldestLinkedCDCCommitLogSegment();
sizeInProgress.getAndAdd(-releasedSize);
logger.debug("Freed up {} bytes after deleting the oldest CDC commit log segment in non-blocking mode. " +
"Total on-disk CDC size: {}; allowed CDC size: {}",
releasedSize, sizeInProgress.get() + segmentSize, allowance);
}
// Aggresively count in the (estimated) size of new segments.
if (segment.getCDCState() == CDCState.PERMITTED)
size += defaultSegmentSize();
sizeInProgress.getAndAdd(segmentSize);
}
// Take this opportunity to kick off a recalc to pick up any consumer file deletion.
@ -250,9 +300,13 @@ public class CommitLogSegmentManagerCDC extends AbstractCommitLogSegmentManager
{
// Add to flushed size before decrementing unflushed so we don't have a window of false generosity
if (segment.getCDCState() == CDCState.CONTAINS)
size += segment.onDiskSize();
sizeInProgress.getAndAdd(segment.onDiskSize());
// Subtract the (estimated) size of the segment from processNewSegment.
// For the segement that CONTAINS, we update with adding the actual onDiskSize and removing the estimated size.
// For the segment that remains in PERMITTED, the file is to be deleted and the estimate should be returned.
if (segment.getCDCState() != CDCState.FORBIDDEN)
size -= defaultSegmentSize();
sizeInProgress.getAndAdd(-defaultSegmentSize());
}
// Take this opportunity to kick off a recalc to pick up any consumer file deletion.
@ -268,7 +322,7 @@ public class CommitLogSegmentManagerCDC extends AbstractCommitLogSegmentManager
{
try
{
cdcSizeCalculationExecutor.submit(() -> recalculateOverflowSize());
cdcSizeCalculationExecutor.submit(this::recalculateOverflowSize);
}
catch (RejectedExecutionException e)
{
@ -287,6 +341,8 @@ public class CommitLogSegmentManagerCDC extends AbstractCommitLogSegmentManager
private int defaultSegmentSize()
{
// CommitLogSegmentSize is only loaded from yaml.
// There is a setter but is used only for testing.
return DatabaseDescriptor.getCommitLogSegmentSize();
}
@ -294,25 +350,17 @@ public class CommitLogSegmentManagerCDC extends AbstractCommitLogSegmentManager
{
try
{
resetSize();
// The Arrays.stream approach is considerably slower on Windows than linux
sizeInProgress = 0;
Files.walkFileTree(path.toPath(), this);
size = sizeInProgress;
sizeInProgress.getAndSet(getAllocatedSize());
}
catch (IOException ie)
{
CommitLog.instance.handleCommitError("Failed CDC Size Calculation", ie);
CommitLog.handleCommitError("Failed CDC Size Calculation", ie);
}
}
@Override
public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException
{
sizeInProgress += attrs.size();
return FileVisitResult.CONTINUE;
}
public void shutdown()
{
if (cdcSizeCalculationExecutor != null && !cdcSizeCalculationExecutor.isShutdown())
@ -323,12 +371,7 @@ public class CommitLogSegmentManagerCDC extends AbstractCommitLogSegmentManager
private void addSize(long toAdd)
{
size += toAdd;
}
private long totalCDCSizeOnDisk()
{
return size;
sizeInProgress.getAndAdd(toAdd);
}
}
@ -347,6 +390,6 @@ public class CommitLogSegmentManagerCDC extends AbstractCommitLogSegmentManager
}
catch (InterruptedException e) {}
return cdcSizeTracker.totalCDCSizeOnDisk();
return cdcSizeTracker.getAllocatedSize();
}
}

10
src/java/org/apache/cassandra/utils/DirectorySizeCalculator.java
View File

@ -31,7 +31,7 @@ import org.apache.cassandra.io.util.File;
public class DirectorySizeCalculator extends SimpleFileVisitor<Path>
{
protected volatile long size = 0;
private volatile long size = 0;
protected final File path;
public DirectorySizeCalculator(File path)
@ -63,4 +63,12 @@ public class DirectorySizeCalculator extends SimpleFileVisitor<Path>
{
return size;
}
/**
* Reset the size to 0 in case that the size calculator is used multiple times
*/
protected void resetSize()
{
size = 0;
}
}

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

@ -29,10 +29,12 @@ import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.ParameterizedClass;
import org.apache.cassandra.config.TransparentDataEncryptionOptions;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.compress.LZ4Compressor;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.io.util.FileDataInput;
import org.apache.cassandra.io.util.FileSegmentInputStream;
import org.apache.cassandra.net.MessagingService;
@ -309,4 +311,19 @@ public class CommitLogDescriptorTest
CommitLogDescriptor desc2 = new CommitLogDescriptor(CommitLogDescriptor.current_version, 1, compression, enabledEncryption);
Assert.assertEquals(desc1, desc2);
}
@Test
public void testInferCDCIndexFile()
{
DatabaseDescriptor.daemonInitialization();
String fileNameSuffix = "CommitLog-2-1340512736956320000";
File validCdcLink = new File(fileNameSuffix + ".log");
File inferredIndexFile = CommitLogDescriptor.inferCdcIndexFile(validCdcLink);
Assert.assertNotNull(inferredIndexFile);
Assert.assertEquals(fileNameSuffix + "_cdc.idx", inferredIndexFile.name());
File invalidCdcLink = new File(fileNameSuffix + ".invalidlog");
inferredIndexFile = CommitLogDescriptor.inferCdcIndexFile(invalidCdcLink);
Assert.assertNull(inferredIndexFile);
}
}

262
test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentManagerCDCTest.java
View File

@ -65,31 +65,8 @@ public class CommitLogSegmentManagerCDCTest extends CQLTester
@Test
public void testCDCWriteFailure() throws Throwable
{
createTable("CREATE TABLE %s (idx int, data text, primary key(idx)) WITH cdc=true;");
CommitLogSegmentManagerCDC cdcMgr = (CommitLogSegmentManagerCDC)CommitLog.instance.segmentManager;
TableMetadata cfm = currentTableMetadata();
// Confirm that logic to check for whether or not we can allocate new CDC segments works
Integer originalCDCSize = DatabaseDescriptor.getCDCSpaceInMB();
try
{
DatabaseDescriptor.setCDCSpaceInMB(32);
// Spin until we hit CDC capacity and make sure we get a CDCWriteException
try
{
// Should trigger on anything < 20:1 compression ratio during compressed test
for (int i = 0; i < 100; i++)
{
new RowUpdateBuilder(cfm, 0, i)
.add("data", randomizeBuffer(DatabaseDescriptor.getCommitLogSegmentSize() / 3))
.build().apply();
}
Assert.fail("Expected CDCWriteException from full CDC but did not receive it.");
}
catch (CDCWriteException e)
{
// expected, do nothing
}
testWithCDCSpaceInMb(32, () -> {
createTableAndBulkWrite();
expectCurrentCDCState(CDCState.FORBIDDEN);
// Confirm we can create a non-cdc table and write to it even while at cdc capacity
@ -97,6 +74,7 @@ public class CommitLogSegmentManagerCDCTest extends CQLTester
execute("INSERT INTO %s (idx, data) VALUES (1, '1');");
// Confirm that, on flush+recyle, we see files show up in cdc_raw
CommitLogSegmentManagerCDC cdcMgr = (CommitLogSegmentManagerCDC)CommitLog.instance.segmentManager;
Keyspace.open(keyspace()).getColumnFamilyStore(currentTable()).forceBlockingFlush();
CommitLog.instance.forceRecycleAllSegments();
cdcMgr.awaitManagementTasksCompletion();
@ -109,57 +87,55 @@ public class CommitLogSegmentManagerCDCTest extends CQLTester
// Update size tracker to reflect deleted files. Should flip flag on current allocatingFrom to allow.
cdcMgr.updateCDCTotalSize();
expectCurrentCDCState(CDCState.PERMITTED);
}
finally
{
DatabaseDescriptor.setCDCSpaceInMB(originalCDCSize);
}
});
}
@Test
public void testSegmentFlaggingOnCreation() throws Throwable
{
CommitLogSegmentManagerCDC cdcMgr = (CommitLogSegmentManagerCDC)CommitLog.instance.segmentManager;
String ct = createTable("CREATE TABLE %s (idx int, data text, primary key(idx)) WITH cdc=true;");
testSegmentFlaggingOnCreation0();
}
int origSize = DatabaseDescriptor.getCDCSpaceInMB();
try
{
DatabaseDescriptor.setCDCSpaceInMB(16);
TableMetadata ccfm = Keyspace.open(keyspace()).getColumnFamilyStore(ct).metadata();
// Spin until we hit CDC capacity and make sure we get a CDCWriteException
try
{
for (int i = 0; i < 1000; i++)
{
new RowUpdateBuilder(ccfm, 0, i)
.add("data", randomizeBuffer(DatabaseDescriptor.getCommitLogSegmentSize() / 3))
.build().apply();
}
Assert.fail("Expected CDCWriteException from full CDC but did not receive it.");
}
catch (CDCWriteException e) { }
@Test
public void testSegmentFlaggingWithNonblockingOnCreation() throws Throwable
{
testWithNonblockingMode(this::testSegmentFlaggingOnCreation0);
}
expectCurrentCDCState(CDCState.FORBIDDEN);
CommitLog.instance.forceRecycleAllSegments();
cdcMgr.awaitManagementTasksCompletion();
// Delete all files in cdc_raw
for (File f : new File(DatabaseDescriptor.getCDCLogLocation()).tryList())
f.tryDelete();
cdcMgr.updateCDCTotalSize();
// Confirm cdc update process changes flag on active segment
expectCurrentCDCState(CDCState.PERMITTED);
// Clear out archived CDC files
for (File f : new File(DatabaseDescriptor.getCDCLogLocation()).tryList()) {
FileUtils.deleteWithConfirm(f);
}
}
finally
{
DatabaseDescriptor.setCDCSpaceInMB(origSize);
@Test
public void testNonblockingShouldMaintainSteadyDiskUsage() throws Throwable
{
final int commitlogSize = DatabaseDescriptor.getCommitLogSegmentSize() / 1024 / 1024;
final int cdcSizeLimit = commitlogSize + 1;
// Clear out all CDC files
for (File f : new File(DatabaseDescriptor.getCDCLogLocation()).tryList()) {
FileUtils.deleteWithConfirm(f);
}
testWithNonblockingMode(() -> testWithCDCSpaceInMb(cdcSizeLimit, () -> {
CommitLogSegmentManagerCDC cdcMgr = (CommitLogSegmentManagerCDC)CommitLog.instance.segmentManager;
Assert.assertEquals(0, cdcMgr.updateCDCTotalSize());
createTableAndBulkWrite();
// Only the current commit log will be kept.
// The older ones are deleted immediately on creating a new segment due to exceeding size limit.
long actualSize = cdcMgr.updateCDCTotalSize();
Assert.assertTrue(actualSize <= cdcSizeLimit * 1024 * 1024);
Assert.assertTrue(actualSize >= DatabaseDescriptor.getCommitLogSegmentSize());
}));
}
@Test // switch from blocking to nonblocking, then back to blocking
public void testSwitchingCDCWriteModes() throws Throwable
{
String tableName = createTableAndBulkWrite();
expectCurrentCDCState(CDCState.FORBIDDEN);
testWithNonblockingMode(() -> {
bulkWrite(tableName);
expectCurrentCDCState(CDCState.CONTAINS);
});
bulkWrite(tableName);
expectCurrentCDCState(CDCState.FORBIDDEN);
}
@Test
@ -187,30 +163,12 @@ public class CommitLogSegmentManagerCDCTest extends CQLTester
}
@Test
public void testCompletedFlag() throws IOException
public void testCompletedFlag() throws Throwable
{
createTable("CREATE TABLE %s (idx int, data text, primary key(idx)) WITH cdc=true;");
String tableName = createTable("CREATE TABLE %s (idx int, data text, primary key(idx)) WITH cdc=true;");
CommitLogSegment initialSegment = CommitLog.instance.segmentManager.allocatingFrom();
Integer originalCDCSize = DatabaseDescriptor.getCDCSpaceInMB();
DatabaseDescriptor.setCDCSpaceInMB(8);
try
{
for (int i = 0; i < 1000; i++)
{
new RowUpdateBuilder(currentTableMetadata(), 0, 1)
.add("data", randomizeBuffer(DatabaseDescriptor.getCommitLogSegmentSize() / 3))
.build().apply();
}
}
catch (CDCWriteException ce)
{
// pass. Expected since we'll have a file or two linked on restart of CommitLog due to replay
}
finally
{
DatabaseDescriptor.setCDCSpaceInMB(originalCDCSize);
}
testWithCDCSpaceInMb(8, () -> bulkWrite(tableName));
CommitLog.instance.forceRecycleAllSegments();
@ -280,32 +238,10 @@ public class CommitLogSegmentManagerCDCTest extends CQLTester
}
@Test
public void testReplayLogic() throws IOException
public void testReplayLogic() throws Throwable
{
// Assert.assertEquals(0, new File(DatabaseDescriptor.getCDCLogLocation()).listFiles().length);
String table_name = createTable("CREATE TABLE %s (idx int, data text, primary key(idx)) WITH cdc=true;");
Integer originalCDCSize = DatabaseDescriptor.getCDCSpaceInMB();
DatabaseDescriptor.setCDCSpaceInMB(8);
TableMetadata ccfm = Keyspace.open(keyspace()).getColumnFamilyStore(table_name).metadata();
try
{
for (int i = 0; i < 1000; i++)
{
new RowUpdateBuilder(ccfm, 0, i)
.add("data", randomizeBuffer(DatabaseDescriptor.getCommitLogSegmentSize() / 3))
.build().apply();
}
Assert.fail("Expected CDCWriteException from full CDC but did not receive it.");
}
catch (CDCWriteException e)
{
// pass
}
finally
{
DatabaseDescriptor.setCDCSpaceInMB(originalCDCSize);
}
// Assert.assertEquals(0, new File(DatabaseDescriptor.getCDCLogLocation()).tryList().length);
testWithCDCSpaceInMb(8, this::createTableAndBulkWrite);
CommitLog.instance.sync(true);
CommitLog.instance.stopUnsafe(false);
@ -449,4 +385,102 @@ public class CommitLogSegmentManagerCDCTest extends CQLTester
expectedState, currentState));
}
}
private void testWithNonblockingMode(Testable test) throws Throwable
{
boolean original = DatabaseDescriptor.getCDCBlockWrites();
CommitLog.instance.setCDCBlockWrites(false);
try
{
test.run();
}
catch (Throwable e)
{
e.printStackTrace();
}
finally
{
CommitLog.instance.setCDCBlockWrites(original);
}
}
private void testWithCDCSpaceInMb(int size, Testable test) throws Throwable
{
int origSize = DatabaseDescriptor.getCDCSpaceInMB();
DatabaseDescriptor.setCDCSpaceInMB(size);
try
{
test.run();
}
finally
{
DatabaseDescriptor.setCDCSpaceInMB(origSize);
}
}
private String createTableAndBulkWrite() throws Throwable
{
String tableName = createTable("CREATE TABLE %s (idx int, data text, primary key(idx)) WITH cdc=true;");
bulkWrite(tableName);
return tableName;
}
private void bulkWrite(String tableName) throws Throwable
{
TableMetadata ccfm = Keyspace.open(keyspace()).getColumnFamilyStore(tableName).metadata();
boolean blockWrites = DatabaseDescriptor.getCDCBlockWrites();
// Spin to make sure we hit CDC capacity
try
{
for (int i = 0; i < 1000; i++)
{
new RowUpdateBuilder(ccfm, 0, i)
.add("data", randomizeBuffer(DatabaseDescriptor.getCommitLogSegmentSize() / 3))
.build().applyFuture().get();
}
if (blockWrites)
Assert.fail("Expected CDCWriteException from full CDC but did not receive it.");
}
catch (CDCWriteException e)
{
if (!blockWrites)
Assert.fail("Excepted no CDCWriteException when not blocking writes but received it.");
}
}
private void testSegmentFlaggingOnCreation0() throws Throwable
{
testWithCDCSpaceInMb(16, () -> {
boolean blockWrites = DatabaseDescriptor.getCDCBlockWrites();
createTableAndBulkWrite();
CommitLogSegmentManagerCDC cdcMgr = (CommitLogSegmentManagerCDC)CommitLog.instance.segmentManager;
expectCurrentCDCState(blockWrites? CDCState.FORBIDDEN : CDCState.CONTAINS);
// When block writes, releasing CDC commit logs should update the CDC state to PERMITTED
if (blockWrites)
{
CommitLog.instance.forceRecycleAllSegments();
cdcMgr.awaitManagementTasksCompletion();
// Delete all files in cdc_raw
for (File f : new File(DatabaseDescriptor.getCDCLogLocation()).tryList())
f.delete();
cdcMgr.updateCDCTotalSize();
// Confirm cdc update process changes flag on active segment
expectCurrentCDCState(CDCState.PERMITTED);
}
// Clear out archived CDC files
for (File f : new File(DatabaseDescriptor.getCDCLogLocation()).tryList()) {
FileUtils.deleteWithConfirm(f);
}
});
}
private interface Testable
{
void run() throws Throwable;
}
}