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.11' into trunk

This commit is contained in:
Jason Brown 2017-12-05 05:10:08 -08:00
parent d274c6ac02 c3a1a4fa80
commit 2402acd47e
19 changed files with 287 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -175,6 +175,7 @@
* Update jackson JSON jars (CASSANDRA-13949)
* Avoid locks when checking LCS fanout and if we should defrag (CASSANDRA-13930)
Merged from 3.0:
* More frequent commitlog chained markers (CASSANDRA-13987)
* Fix serialized size of DataLimits (CASSANDRA-14057)
* Add flag to allow dropping oversized read repair mutations (CASSANDRA-13975)
* Fix SSTableLoader logger message (CASSANDRA-14003)

10
conf/cassandra.yaml
View File

@ -375,10 +375,18 @@ counter_cache_save_period: 7200
#
# the default option is "periodic" where writes may be acked immediately
# and the CommitLog is simply synced every commitlog_sync_period_in_ms
# milliseconds.
# milliseconds.
commitlog_sync: periodic
commitlog_sync_period_in_ms: 10000
# Time interval in millis at which we should update the chained markers in the commitlog.
# This allows more of the commitlog to be replayed from the mmapped file
# if the cassandra process crashes; this does not help in durability for surviving a host fail.
# This value only makes sense if it is significantly less that commitlog_sync_period_in_ms,
# and only applies to periodic mode when not using commitlog compression or encryption.
# commitlog_marker_period_in_ms: 100
# The size of the individual commitlog file segments. A commitlog
# segment may be archived, deleted, or recycled once all the data
# in it (potentially from each columnfamily in the system) has been

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

@ -192,6 +192,7 @@ public class Config
public double commitlog_sync_batch_window_in_ms = Double.NaN;
public double commitlog_sync_group_window_in_ms = Double.NaN;
public int commitlog_sync_period_in_ms;
public int commitlog_marker_period_in_ms = 0;
public int commitlog_segment_size_in_mb = 32;
public ParameterizedClass commitlog_compression;
public int commitlog_max_compression_buffers_in_pool = 3;

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

@ -1812,6 +1812,16 @@ public class DatabaseDescriptor
conf.commitlog_sync_period_in_ms = periodMillis;
}
public static void setCommitLogMarkerPeriod(int markerPeriod)
{
conf.commitlog_marker_period_in_ms = markerPeriod;
}
public static int getCommitLogMarkerPeriod()
{
return conf.commitlog_marker_period_in_ms;
}
public static Config.CommitLogSync getCommitLogSync()
{
return conf.commitlog_sync;

8
src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogSegmentManager.java
View File

@ -512,9 +512,11 @@ public abstract class AbstractCommitLogSegmentManager
}
/**
* Forces a disk flush on the commit log files that need it. Blocking.
* Requests commit log files sync themselves, if needed. This may or may not involve flushing to disk.
*
* @param flush Request that the sync operation flush the file to disk.
*/
public void sync() throws IOException
public void sync(boolean flush) throws IOException
{
CommitLogSegment current = allocatingFrom;
for (CommitLogSegment segment : getActiveSegments())
@ -522,7 +524,7 @@ public abstract class AbstractCommitLogSegmentManager
// Do not sync segments that became active after sync started.
if (segment.id > current.id)
return;
segment.sync();
segment.sync(flush);
}
}

84
src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java
View File

@ -27,6 +27,8 @@ import org.slf4j.LoggerFactory;
import com.codahale.metrics.Timer.Context;
import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.commitlog.CommitLogSegment.Allocation;
import org.apache.cassandra.utils.NoSpamLogger;
import org.apache.cassandra.utils.concurrent.WaitQueue;
@ -48,7 +50,24 @@ public abstract class AbstractCommitLogService
final CommitLog commitLog;
private final String name;
private final long pollIntervalNanos;
/**
* The duration between syncs to disk.
*/
private final long syncIntervalNanos;
/**
* The duration between updating the chained markers in the the commit log file. This value should be
* 0 < {@link #markerIntervalNanos} <= {@link #syncIntervalNanos}.
*/
private final long markerIntervalNanos;
/**
* A flag that callers outside of the sync thread can use to signal they want the commitlog segments
* to be flushed to disk. Note: this flag is primarily to support commit log's batch mode, which requires
* an immediate flush to disk on every mutation; see {@link BatchCommitLogService#maybeWaitForSync(Allocation)}.
*/
private volatile boolean syncRequested;
private static final Logger logger = LoggerFactory.getLogger(AbstractCommitLogService.class);
@ -58,19 +77,45 @@ public abstract class AbstractCommitLogService
*
* Subclasses may be notified when a sync finishes by using the syncComplete WaitQueue.
*/
AbstractCommitLogService(final CommitLog commitLog, final String name, final long pollIntervalMillis)
AbstractCommitLogService(final CommitLog commitLog, final String name, final long syncIntervalMillis)
{
this(commitLog, name, syncIntervalMillis, syncIntervalMillis);
}
/**
* CommitLogService provides a fsync service for Allocations, fulfilling either the
* Batch or Periodic contract.
*
* Subclasses may be notified when a sync finishes by using the syncComplete WaitQueue.
*/
AbstractCommitLogService(final CommitLog commitLog, final String name, final long syncIntervalMillis, long markerIntervalMillis)
{
this.commitLog = commitLog;
this.name = name;
this.pollIntervalNanos = TimeUnit.NANOSECONDS.convert(pollIntervalMillis, TimeUnit.MILLISECONDS);
this.syncIntervalNanos = TimeUnit.NANOSECONDS.convert(syncIntervalMillis, TimeUnit.MILLISECONDS);
// if we are not using periodic mode, or we using compression/encryption, we shouldn't update the chained markers
// faster than the sync interval
if (DatabaseDescriptor.getCommitLogSync() != Config.CommitLogSync.periodic || commitLog.configuration.useCompression() || commitLog.configuration.useEncryption())
markerIntervalMillis = syncIntervalMillis;
// apply basic bounds checking on the marker interval
if (markerIntervalMillis <= 0 || markerIntervalMillis > syncIntervalMillis)
{
logger.debug("commit log marker interval {} is less than zero or above the sync interval {}; setting value to sync interval",
markerIntervalMillis, syncIntervalMillis);
markerIntervalMillis = syncIntervalMillis;
}
this.markerIntervalNanos = TimeUnit.NANOSECONDS.convert(markerIntervalMillis, TimeUnit.MILLISECONDS);
}
// Separated into individual method to ensure relevant objects are constructed before this is started.
void start()
{
if (pollIntervalNanos < 1)
if (syncIntervalNanos < 1)
throw new IllegalArgumentException(String.format("Commit log flush interval must be positive: %fms",
pollIntervalNanos * 1e-6));
syncIntervalNanos * 1e-6));
Runnable runnable = new Runnable()
{
@ -90,16 +135,24 @@ public abstract class AbstractCommitLogService
try
{
// sync and signal
long syncStarted = System.nanoTime();
// This is a target for Byteman in CommitLogSegmentManagerTest
commitLog.sync();
lastSyncedAt = syncStarted;
syncComplete.signalAll();
long pollStarted = System.nanoTime();
if (lastSyncedAt + syncIntervalNanos <= pollStarted || shutdownRequested || syncRequested)
{
// in this branch, we want to flush the commit log to disk
commitLog.sync(true);
syncRequested = false;
lastSyncedAt = pollStarted;
syncComplete.signalAll();
}
else
{
// in this branch, just update the commit log sync headers
commitLog.sync(false);
}
// sleep any time we have left before the next one is due
long now = System.nanoTime();
long wakeUpAt = syncStarted + pollIntervalNanos;
long wakeUpAt = pollStarted + markerIntervalNanos;
if (wakeUpAt < now)
{
// if we have lagged noticeably, update our lag counter
@ -112,7 +165,7 @@ public abstract class AbstractCommitLogService
lagCount++;
}
syncCount++;
totalSyncDuration += now - syncStarted;
totalSyncDuration += now - pollStarted;
if (firstLagAt > 0)
{
@ -143,7 +196,7 @@ public abstract class AbstractCommitLogService
break;
// sleep for full poll-interval after an error, so we don't spam the log file
LockSupport.parkNanos(pollIntervalNanos);
LockSupport.parkNanos(markerIntervalNanos);
}
}
}
@ -168,8 +221,9 @@ public abstract class AbstractCommitLogService
/**
* Request an additional sync cycle without blocking.
*/
public void requestExtraSync()
void requestExtraSync()
{
syncRequested = true;
LockSupport.unpark(thread);
}

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

@ -233,9 +233,9 @@ public class CommitLog implements CommitLogMBean
/**
* Forces a disk flush on the commit log files that need it. Blocking.
*/
public void sync() throws IOException
public void sync(boolean flush) throws IOException
{
segmentManager.sync();
segmentManager.sync(flush);
}
/**

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

@ -97,6 +97,12 @@ public abstract class CommitLogSegment
@VisibleForTesting
volatile int lastSyncedOffset;
/**
* Everything before this offset has it's markers written into the {@link #buffer}, but has not necessarily
* been flushed to disk. This value should be greater than or equal to {@link #lastSyncedOffset}.
*/
private volatile int lastMarkerOffset;
// The end position of the buffer. Initially set to its capacity and updated to point to the last written position
// as the segment is being closed.
// No need to be volatile as writes are protected by appendOrder barrier.
@ -184,7 +190,8 @@ public abstract class CommitLogSegment
{
CommitLogDescriptor.writeHeader(buffer, descriptor, additionalHeaderParameters());
endOfBuffer = buffer.capacity();
lastSyncedOffset = buffer.position();
lastSyncedOffset = lastMarkerOffset = buffer.position();
allocatePosition.set(lastSyncedOffset + SYNC_MARKER_SIZE);
headerWritten = true;
}
@ -261,7 +268,7 @@ public abstract class CommitLogSegment
// ensures no more of this segment is writeable, by allocating any unused section at the end and marking it discarded
void discardUnusedTail()
{
// We guard this with the OpOrdering instead of synchronised due to potential dead-lock with CLSM.advanceAllocatingFrom()
// We guard this with the OpOrdering instead of synchronised due to potential dead-lock with ACLSM.advanceAllocatingFrom()
// Ensures endOfBuffer update is reflected in the buffer end position picked up by sync().
// This actually isn't strictly necessary, as currently all calls to discardUnusedTail are executed either by the thread
// running sync or within a mutation already protected by this OpOrdering, but to prevent future potential mistakes,
@ -300,15 +307,20 @@ public abstract class CommitLogSegment
}
/**
* Forces a disk flush for this segment file.
* Update the chained markers in the commit log buffer and possibly force a disk flush for this segment file.
*
* @param flush true if the segment should flush to disk; else, false for just updating the chained markers.
*/
synchronized void sync()
synchronized void sync(boolean flush)
{
if (!headerWritten)
throw new IllegalStateException("commit log header has not been written");
boolean close = false;
assert lastMarkerOffset >= lastSyncedOffset : String.format("commit log segment positions are incorrect: last marked = %d, last synced = %d",
lastMarkerOffset, lastSyncedOffset);
// check we have more work to do
if (allocatePosition.get() <= lastSyncedOffset + SYNC_MARKER_SIZE)
final boolean needToMarkData = allocatePosition.get() > lastMarkerOffset + SYNC_MARKER_SIZE;
final boolean hasDataToFlush = lastSyncedOffset != lastMarkerOffset;
if (!(needToMarkData || hasDataToFlush))
return;
// Note: Even if the very first allocation of this sync section failed, we still want to enter this
// to ensure the segment is closed. As allocatePosition is set to 1 beyond the capacity of the buffer,
@ -316,34 +328,50 @@ public abstract class CommitLogSegment
// succeeded in the previous sync.
assert buffer != null; // Only close once.
int startMarker = lastSyncedOffset;
// Allocate a new sync marker; this is both necessary in itself, but also serves to demarcate
// the point at which we can safely consider records to have been completely written to.
int nextMarker = allocate(SYNC_MARKER_SIZE);
if (nextMarker < 0)
boolean close = false;
int startMarker = lastMarkerOffset;
int nextMarker, sectionEnd;
if (needToMarkData)
{
// Ensure no more of this CLS is writeable, and mark ourselves for closing.
discardUnusedTail();
close = true;
// Allocate a new sync marker; this is both necessary in itself, but also serves to demarcate
// the point at which we can safely consider records to have been completely written to.
nextMarker = allocate(SYNC_MARKER_SIZE);
if (nextMarker < 0)
{
// Ensure no more of this CLS is writeable, and mark ourselves for closing.
discardUnusedTail();
close = true;
// We use the buffer size as the synced position after a close instead of the end of the actual data
// to make sure we only close the buffer once.
// The endOfBuffer position may be incorrect at this point (to be written by another stalled thread).
nextMarker = buffer.capacity();
// We use the buffer size as the synced position after a close instead of the end of the actual data
// to make sure we only close the buffer once.
// The endOfBuffer position may be incorrect at this point (to be written by another stalled thread).
nextMarker = buffer.capacity();
}
// Wait for mutations to complete as well as endOfBuffer to have been written.
waitForModifications();
sectionEnd = close ? endOfBuffer : nextMarker;
// Possibly perform compression or encryption and update the chained markers
write(startMarker, sectionEnd);
lastMarkerOffset = sectionEnd;
}
else
{
// note: we don't need to waitForModifications() as, once we get to this block, we are only doing the flush
// and any mutations have already been fully written into the segment (as we wait for it in the previous block).
nextMarker = lastMarkerOffset;
sectionEnd = nextMarker;
}
// Wait for mutations to complete as well as endOfBuffer to have been written.
waitForModifications();
int sectionEnd = close ? endOfBuffer : nextMarker;
// Possibly perform compression or encryption, writing to file and flush.
write(startMarker, sectionEnd);
if (flush || close)
{
flush(startMarker, sectionEnd);
if (cdcState == CDCState.CONTAINS)
writeCDCIndexFile(descriptor, sectionEnd, close);
lastSyncedOffset = lastMarkerOffset = nextMarker;
}
if (cdcState == CDCState.CONTAINS)
writeCDCIndexFile(descriptor, sectionEnd, close);
// Signal the sync as complete.
lastSyncedOffset = nextMarker;
if (close)
internalClose();
syncComplete.signalAll();
@ -394,6 +422,8 @@ public abstract class CommitLogSegment
abstract void write(int lastSyncedOffset, int nextMarker);
abstract void flush(int startMarker, int nextMarker);
public boolean isStillAllocating()
{
return allocatePosition.get() < endOfBuffer;
@ -488,7 +518,7 @@ public abstract class CommitLogSegment
synchronized void close()
{
discardUnusedTail();
sync();
sync(true);
assert buffer == null;
}

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

@ -21,7 +21,6 @@ import java.nio.ByteBuffer;
import org.apache.cassandra.io.FSWriteError;
import org.apache.cassandra.io.compress.ICompressor;
import org.apache.cassandra.utils.SyncUtil;
/**
* Compressed commit log segment. Provides an in-memory buffer for the mutation threads. On sync compresses the written
@ -30,7 +29,7 @@ import org.apache.cassandra.utils.SyncUtil;
* The format of the compressed commit log is as follows:
* - standard commit log header (as written by {@link CommitLogDescriptor#writeHeader(ByteBuffer, CommitLogDescriptor)})
* - a series of 'sync segments' that are written every time the commit log is sync()'ed
* -- a sync section header, see {@link CommitLogSegment#writeSyncMarker(ByteBuffer, int, int, int)}
* -- a sync section header, see {@link CommitLogSegment#writeSyncMarker(long, ByteBuffer, int, int, int)}
* -- total plain text length for this section
* -- a block of compressed data
*/
@ -82,7 +81,6 @@ public class CompressedSegment extends FileDirectSegment
channel.write(compressedBuffer);
assert channel.position() - lastWrittenPos == compressedBuffer.limit();
lastWrittenPos = channel.position();
SyncUtil.force(channel, true);
}
catch (Exception e)
{

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

@ -32,7 +32,6 @@ import org.apache.cassandra.io.compress.ICompressor;
import org.apache.cassandra.security.EncryptionUtils;
import org.apache.cassandra.security.EncryptionContext;
import org.apache.cassandra.utils.Hex;
import org.apache.cassandra.utils.SyncUtil;
import static org.apache.cassandra.security.EncryptionUtils.ENCRYPTED_BLOCK_HEADER_SIZE;
@ -143,8 +142,6 @@ public class EncryptedSegment extends FileDirectSegment
channel.position(syncMarkerPosition);
channel.write(buffer);
SyncUtil.force(channel, true);
}
catch (Exception e)
{

14
src/java/org/apache/cassandra/db/commitlog/FileDirectSegment.java
View File

@ -21,6 +21,7 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import org.apache.cassandra.io.FSWriteError;
import org.apache.cassandra.utils.SyncUtil;
/**
* Writes to the backing commit log file only on sync, allowing transformations of the mutations,
@ -63,4 +64,17 @@ public abstract class FileDirectSegment extends CommitLogSegment
manager.notifyBufferFreed();
}
}
@Override
protected void flush(int startMarker, int nextMarker)
{
try
{
SyncUtil.force(channel, true);
}
catch (Exception e)
{
throw new FSWriteError(e, getPath());
}
}
}

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

@ -77,7 +77,11 @@ public class MemoryMappedSegment extends CommitLogSegment
// write previous sync marker to point to next sync marker
// we don't chain the crcs here to ensure this method is idempotent if it fails
writeSyncMarker(id, buffer, startMarker, startMarker, nextMarker);
}
@Override
protected void flush(int startMarker, int nextMarker)
{
try
{
SyncUtil.force((MappedByteBuffer) buffer);

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

@ -25,7 +25,7 @@ class PeriodicCommitLogService extends AbstractCommitLogService
public PeriodicCommitLogService(final CommitLog commitLog)
{
super(commitLog, "PERIODIC-COMMIT-LOG-SYNCER", DatabaseDescriptor.getCommitLogSyncPeriod());
super(commitLog, "PERIODIC-COMMIT-LOG-SYNCER", DatabaseDescriptor.getCommitLogSyncPeriod(), DatabaseDescriptor.getCommitLogMarkerPeriod());
}
protected void maybeWaitForSync(CommitLogSegment.Allocation alloc)

2
test/unit/org/apache/cassandra/db/commitlog/CDCTestReplayer.java
View File

@ -39,7 +39,7 @@ public class CDCTestReplayer extends CommitLogReplayer
public CDCTestReplayer() throws IOException
{
super(CommitLog.instance, CommitLogPosition.NONE, null, ReplayFilter.create());
CommitLog.instance.sync();
CommitLog.instance.sync(true);
commitLogReader = new CommitLogTestReader();
}

98
test/unit/org/apache/cassandra/db/commitlog/CommitLogChainedMarkersTest.java Normal file
View File

@ -0,0 +1,98 @@
/*
* 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.db.commitlog;
import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Random;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
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.compaction.CompactionManager;
import org.apache.cassandra.db.marshal.AsciiType;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.schema.KeyspaceParams;
import org.jboss.byteman.contrib.bmunit.BMRule;
import org.jboss.byteman.contrib.bmunit.BMUnitRunner;
/**
* Tests the commitlog to make sure we can replay it - explicitly for the case where we update the chained markers
* in the commit log segment but do not flush the file to disk.
*/
@RunWith(BMUnitRunner.class)
public class CommitLogChainedMarkersTest
{
private static final String KEYSPACE1 = "CommitLogTest";
private static final String STANDARD1 = "CommitLogChainedMarkersTest";
@Test
@BMRule(name = "force all calls to sync() to not flush to disk",
targetClass = "CommitLogSegment",
targetMethod = "sync(boolean)",
action = "$flush = false")
public void replayCommitLogWithoutFlushing() throws IOException
{
// this method is blend of CommitLogSegmentBackpressureTest & CommitLogReaderTest methods
DatabaseDescriptor.daemonInitialization();
DatabaseDescriptor.setCommitLogSegmentSize(5);
DatabaseDescriptor.setCommitLogSync(Config.CommitLogSync.periodic);
DatabaseDescriptor.setCommitLogSyncPeriod(10000 * 1000);
DatabaseDescriptor.setCommitLogMarkerPeriod(1);
SchemaLoader.prepareServer();
SchemaLoader.createKeyspace(KEYSPACE1,
KeyspaceParams.simple(1),
SchemaLoader.standardCFMD(KEYSPACE1, STANDARD1, 0, AsciiType.instance, BytesType.instance));
CompactionManager.instance.disableAutoCompaction();
ColumnFamilyStore cfs1 = Keyspace.open(KEYSPACE1).getColumnFamilyStore(STANDARD1);
byte[] entropy = new byte[1024];
new Random().nextBytes(entropy);
final Mutation m = new RowUpdateBuilder(cfs1.metadata.get(), 0, "k")
.clustering("bytes")
.add("val", ByteBuffer.wrap(entropy))
.build();
int samples = 10000;
for (int i = 0; i < samples; i++)
CommitLog.instance.add(m);
CommitLog.instance.sync(false);
ArrayList<File> toCheck = CommitLogReaderTest.getCommitLogs();
CommitLogReader reader = new CommitLogReader();
CommitLogReaderTest.TestCLRHandler testHandler = new CommitLogReaderTest.TestCLRHandler(cfs1.metadata.get());
for (File f : toCheck)
reader.readCommitLogSegment(testHandler, f, CommitLogReader.ALL_MUTATIONS, false);
Assert.assertEquals(samples, testHandler.seenMutationCount());
}
}

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

@ -68,12 +68,12 @@ public class CommitLogSegmentBackpressureTest
@BMRules(rules = {@BMRule(name = "Acquire Semaphore before sync",
targetClass = "AbstractCommitLogService$1",
targetMethod = "run",
targetLocation = "AT INVOKE org.apache.cassandra.db.commitlog.CommitLog.sync",
targetLocation = "AT INVOKE org.apache.cassandra.db.commitlog.CommitLog.sync(boolean)",
action = "org.apache.cassandra.db.commitlog.CommitLogSegmentBackpressureTest.allowSync.acquire()"),
@BMRule(name = "Release Semaphore after sync",
targetClass = "AbstractCommitLogService$1",
targetMethod = "run",
targetLocation = "AFTER INVOKE org.apache.cassandra.db.commitlog.CommitLog.sync",
targetLocation = "AFTER INVOKE org.apache.cassandra.db.commitlog.CommitLog.sync(boolean)",
action = "org.apache.cassandra.db.commitlog.CommitLogSegmentBackpressureTest.allowSync.release()")})
public void testCompressedCommitLogBackpressure() throws Throwable
{

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

@ -167,7 +167,7 @@ public class CommitLogSegmentManagerCDCTest extends CQLTester
.add("data", randomizeBuffer(DatabaseDescriptor.getCommitLogSegmentSize() / 3))
.build().apply();
CommitLog.instance.sync();
CommitLog.instance.sync(true);
CommitLogSegment currentSegment = CommitLog.instance.segmentManager.allocatingFrom();
int syncOffset = currentSegment.lastSyncedOffset;
@ -233,7 +233,7 @@ public class CommitLogSegmentManagerCDCTest extends CQLTester
Assert.assertFalse("Expected index file to not be created but found: " + cdcIndexFile, cdcIndexFile.exists());
// Sync and confirm no index written as index is written on flush
CommitLog.instance.sync();
CommitLog.instance.sync(true);
Assert.assertTrue("File does not exist: " + linked, Files.exists(linked));
Assert.assertFalse("Expected index file to not be created but found: " + cdcIndexFile, cdcIndexFile.exists());
@ -260,7 +260,7 @@ public class CommitLogSegmentManagerCDCTest extends CQLTester
Assert.assertTrue("File does not exist: " + linked, Files.exists(linked));
// Sync and confirm index written as index is written on flush
CommitLog.instance.sync();
CommitLog.instance.sync(true);
Assert.assertTrue("File does not exist: " + linked, Files.exists(linked));
Assert.assertTrue("Expected cdc index file after flush but found none: " + cdcIndexFile, cdcIndexFile.exists());
@ -293,7 +293,7 @@ public class CommitLogSegmentManagerCDCTest extends CQLTester
// pass
}
CommitLog.instance.sync();
CommitLog.instance.sync(true);
CommitLog.instance.stopUnsafe(false);
// Build up a list of expected index files after replay and then clear out cdc_raw

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

@ -341,7 +341,7 @@ public abstract class CommitLogTest
// "Flush": this won't delete anything
TableId id1 = rm.getTableIds().iterator().next();
CommitLog.instance.sync();
CommitLog.instance.sync(true);
CommitLog.instance.discardCompletedSegments(id1, CommitLogPosition.NONE, CommitLog.instance.getCurrentPosition());
assertEquals(1, CommitLog.instance.segmentManager.getActiveSegments().size());
@ -676,7 +676,7 @@ public abstract class CommitLogTest
cellCount += 1;
CommitLog.instance.add(rm2);
CommitLog.instance.sync();
CommitLog.instance.sync(true);
SimpleCountingReplayer replayer = new SimpleCountingReplayer(CommitLog.instance, CommitLogPosition.NONE, cfs.metadata());
List<String> activeSegments = CommitLog.instance.getActiveSegmentNames();
@ -713,7 +713,7 @@ public abstract class CommitLogTest
}
}
CommitLog.instance.sync();
CommitLog.instance.sync(true);
SimpleCountingReplayer replayer = new SimpleCountingReplayer(CommitLog.instance, commitLogPosition, cfs.metadata());
List<String> activeSegments = CommitLog.instance.getActiveSegmentNames();
@ -806,7 +806,7 @@ public abstract class CommitLogTest
DatabaseDescriptor.setDiskFailurePolicy(oldPolicy);
}
CommitLog.instance.sync();
CommitLog.instance.sync(true);
System.setProperty("cassandra.replayList", KEYSPACE1 + "." + STANDARD1);
// Currently we don't attempt to re-flush a memtable that failed, thus make sure data is replayed by commitlog.
// If retries work subsequent flushes should clear up error and this should change to expect 0.
@ -839,7 +839,7 @@ public abstract class CommitLogTest
for (SSTableReader reader : cfs.getLiveSSTables())
reader.reloadSSTableMetadata();
CommitLog.instance.sync();
CommitLog.instance.sync(true);
System.setProperty("cassandra.replayList", KEYSPACE1 + "." + STANDARD1);
// In the absence of error, this should be 0 because forceBlockingFlush/forceRecycleAllSegments would have
// persisted all data in the commit log. Because we know there was an error, there must be something left to

2
test/unit/org/apache/cassandra/db/commitlog/CommitLogTestReplayer.java
View File

@ -40,7 +40,7 @@ public class CommitLogTestReplayer extends CommitLogReplayer
public CommitLogTestReplayer(Predicate<Mutation> processor) throws IOException
{
super(CommitLog.instance, CommitLogPosition.NONE, null, ReplayFilter.create());
CommitLog.instance.sync();
CommitLog.instance.sync(true);
this.processor = processor;
commitLogReader = new CommitLogTestReader();