diff --git a/CHANGES.txt b/CHANGES.txt index fcccccdb72..7823f87a70 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -23,6 +23,7 @@ * (cqlsh) Improve waiting for a trace to complete (CASSANDRA-7626) * Fix tracing of concurrent range slices and 2ary index queries (CASSANDRA-7626) Merged from 2.0: + * Always flush on truncate (CASSANDRA-7511) * Fix ReversedType(DateType) mapping to native protocol (CASSANDRA-7576) * Always merge ranges owned by a single node (CASSANDRA-6930) * Track max/min timestamps for range tombstones (CASSANDRA-7647) diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index d89967d8e1..8d47b49516 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -1223,6 +1223,11 @@ public class DatabaseDescriptor return conf.auto_snapshot; } + @VisibleForTesting + public static void setAutoSnapshot(boolean autoSnapshot) { + conf.auto_snapshot = autoSnapshot; + } + public static boolean isAutoBootstrap() { return conf.auto_bootstrap; diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 8e7bdeb4f0..a1220dfe55 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -2420,25 +2420,12 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean // position in the System keyspace. logger.debug("truncating {}", name); - if (DatabaseDescriptor.isAutoSnapshot()) - { - // flush the CF being truncated before forcing the new segment - forceBlockingFlush(); + // flush the CF being truncated before forcing the new segment + forceBlockingFlush(); - // sleep a little to make sure that our truncatedAt comes after any sstable - // that was part of the flushed we forced; otherwise on a tie, it won't get deleted. - Uninterruptibles.sleepUninterruptibly(1, TimeUnit.MILLISECONDS); - } - else - { - // just nuke the memtable data w/o writing to disk first - synchronized (data) - { - final Flush flush = new Flush(true); - flushExecutor.execute(flush); - postFlushExecutor.submit(flush.postFlush); - } - } + // sleep a little to make sure that our truncatedAt comes after any sstable + // that was part of the flushed we forced; otherwise on a tie, it won't get deleted. + Uninterruptibles.sleepUninterruptibly(1, TimeUnit.MILLISECONDS); Runnable truncateRunnable = new Runnable() { diff --git a/test/conf/cassandra.yaml b/test/conf/cassandra.yaml index 43995e6e75..ec988e299f 100644 --- a/test/conf/cassandra.yaml +++ b/test/conf/cassandra.yaml @@ -6,6 +6,7 @@ cluster_name: Test Cluster memtable_allocation_type: offheap_objects commitlog_sync: batch commitlog_sync_batch_window_in_ms: 1.0 +commitlog_segment_size_in_mb: 5 partitioner: org.apache.cassandra.dht.ByteOrderedPartitioner listen_address: 127.0.0.1 storage_port: 7010 diff --git a/test/unit/org/apache/cassandra/db/CommitLogTest.java b/test/unit/org/apache/cassandra/db/CommitLogTest.java index 7046536948..a58549a12c 100644 --- a/test/unit/org/apache/cassandra/db/CommitLogTest.java +++ b/test/unit/org/apache/cassandra/db/CommitLogTest.java @@ -22,6 +22,7 @@ package org.apache.cassandra.db; import java.io.*; import java.nio.ByteBuffer; import java.util.UUID; +import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.zip.CRC32; import java.util.zip.Checksum; @@ -36,10 +37,12 @@ import org.apache.cassandra.config.Config; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.commitlog.CommitLog; import org.apache.cassandra.db.commitlog.CommitLogDescriptor; +import org.apache.cassandra.db.commitlog.ReplayPosition; import org.apache.cassandra.db.commitlog.CommitLogSegment; import org.apache.cassandra.db.composites.CellName; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.utils.FBUtilities; import static org.apache.cassandra.utils.ByteBufferUtil.bytes; @@ -295,4 +298,33 @@ public class CommitLogTest extends SchemaLoader } } + @Test + public void testTruncateWithoutSnapshot() throws ExecutionException, InterruptedException + { + CommitLog.instance.resetUnsafe(); + boolean prev = DatabaseDescriptor.isAutoSnapshot(); + DatabaseDescriptor.setAutoSnapshot(false); + ColumnFamilyStore cfs1 = Keyspace.open("Keyspace1").getColumnFamilyStore("Standard1"); + ColumnFamilyStore cfs2 = Keyspace.open("Keyspace1").getColumnFamilyStore("Standard2"); + + final Mutation rm1 = new Mutation("Keyspace1", bytes("k")); + rm1.add("Standard1", Util.cellname("c1"), ByteBuffer.allocate(100), 0); + rm1.apply(); + cfs1.truncateBlocking(); + DatabaseDescriptor.setAutoSnapshot(prev); + final Mutation rm2 = new Mutation("Keyspace1", bytes("k")); + rm2.add("Standard2", Util.cellname("c1"), ByteBuffer.allocate(DatabaseDescriptor.getCommitLogSegmentSize() / 4), 0); + + for (int i = 0 ; i < 5 ; i++) + CommitLog.instance.add(rm2); + + Assert.assertEquals(2, CommitLog.instance.activeSegments()); + ReplayPosition position = CommitLog.instance.getContext(); + for (Keyspace ks : Keyspace.system()) + for (ColumnFamilyStore syscfs : ks.getColumnFamilyStores()) + CommitLog.instance.discardCompletedSegments(syscfs.metadata.cfId, position); + CommitLog.instance.discardCompletedSegments(cfs2.metadata.cfId, position); + Assert.assertEquals(1, CommitLog.instance.activeSegments()); + } + }