You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by ad...@apache.org on 2021/06/01 15:31:25 UTC
[cassandra] 01/01: Merge branch 'cassandra-3.0' into cassandra-3.11
This is an automated email from the ASF dual-hosted git repository.
adelapena pushed a commit to branch cassandra-3.11
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit a20913910574e290458d1875e86ce37bcf67c86f
Merge: 803a23b 844c8b0
Author: Andrés de la Peña <a....@gmail.com>
AuthorDate: Tue Jun 1 16:25:24 2021 +0100
Merge branch 'cassandra-3.0' into cassandra-3.11
.../TimeWindowCompactionStrategyTest.java | 93 ++++++++++++----------
1 file changed, 51 insertions(+), 42 deletions(-)
diff --cc test/unit/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategyTest.java
index b67bf16,051e7c0..d427852
--- a/test/unit/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategyTest.java
+++ b/test/unit/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategyTest.java
@@@ -187,25 -179,21 +190,25 @@@ public class TimeWindowCompactionStrate
List<SSTableReader> sstrs = new ArrayList<>(cfs.getLiveSSTables());
// We'll put 3 sstables into the newest bucket
- for (int i = 0 ; i < 3; i++)
+ for (int i = 0; i < 3; i++)
{
- Pair<Long,Long> bounds = getWindowBoundsInMillis(TimeUnit.HOURS, 1, tstamp );
- Pair<Long, Long> bounds = getWindowBoundsInMillis(TimeUnit.HOURS, 1, tstamp );
++ Pair<Long, Long> bounds = getWindowBoundsInMillis(TimeUnit.HOURS, 1, tstamp);
buckets.put(bounds.left, sstrs.get(i));
}
- List<SSTableReader> newBucket = newestBucket(buckets, 4, 32, TimeUnit.HOURS, 1, new SizeTieredCompactionStrategyOptions(), getWindowBoundsInMillis(TimeUnit.HOURS, 1, System.currentTimeMillis()).left );
- assertTrue("incoming bucket should not be accepted when it has below the min threshold SSTables", newBucket.isEmpty());
- TimeWindowCompactionStrategy.NewestBucket newBucket = newestBucket(buckets, 4, 32, new SizeTieredCompactionStrategyOptions(), getWindowBoundsInMillis(TimeUnit.HOURS, 1, System.currentTimeMillis()).left );
- newBucket = newestBucket(buckets, 2, 32, TimeUnit.HOURS, 1, new SizeTieredCompactionStrategyOptions(), getWindowBoundsInMillis(TimeUnit.HOURS, 1, System.currentTimeMillis()).left);
- assertFalse("incoming bucket should be accepted when it is larger than the min threshold SSTables", newBucket.isEmpty());
++ TimeWindowCompactionStrategy.NewestBucket newBucket = newestBucket(buckets, 4, 32, new SizeTieredCompactionStrategyOptions(), getWindowBoundsInMillis(TimeUnit.HOURS, 1, System.currentTimeMillis()).left);
+ assertTrue("incoming bucket should not be accepted when it has below the min threshold SSTables", newBucket.sstables.isEmpty());
+ assertEquals("there should be no estimated remaining tasks when bucket is below min threshold SSTables", 0, newBucket.estimatedRemainingTasks);
+
+
+ newBucket = newestBucket(buckets, 2, 32, new SizeTieredCompactionStrategyOptions(), getWindowBoundsInMillis(TimeUnit.HOURS, 1, System.currentTimeMillis()).left);
- assertTrue("incoming bucket should be accepted when it is larger than the min threshold SSTables", !newBucket.sstables.isEmpty());
++ assertFalse("incoming bucket should be accepted when it is larger than the min threshold SSTables", newBucket.sstables.isEmpty());
+ assertEquals("there should be one estimated remaining task when bucket is larger than the min threshold SSTables", 1, newBucket.estimatedRemainingTasks);
// And 2 into the second bucket (1 hour back)
- for (int i = 3 ; i < 5; i++)
+ for (int i = 3; i < 5; i++)
{
- Pair<Long,Long> bounds = getWindowBoundsInMillis(TimeUnit.HOURS, 1, tstamp2 );
- Pair<Long, Long> bounds = getWindowBoundsInMillis(TimeUnit.HOURS, 1, tstamp2 );
++ Pair<Long, Long> bounds = getWindowBoundsInMillis(TimeUnit.HOURS, 1, tstamp2);
buckets.put(bounds.left, sstrs.get(i));
}
@@@ -235,11 -223,8 +238,11 @@@
buckets.put(bounds.left, sstrs.get(i));
}
- newBucket = newestBucket(buckets, 4, 32, TimeUnit.DAYS, 1, new SizeTieredCompactionStrategyOptions(), getWindowBoundsInMillis(TimeUnit.HOURS, 1, System.currentTimeMillis()).left);
- assertEquals("new bucket should be trimmed to max threshold of 32", newBucket.size(), 32);
+ newBucket = newestBucket(buckets, 4, 32, new SizeTieredCompactionStrategyOptions(), getWindowBoundsInMillis(TimeUnit.HOURS, 1, System.currentTimeMillis()).left);
- assertEquals("new bucket should be trimmed to max threshold of 32", newBucket.sstables.size(), 32);
++ assertEquals("new bucket should be trimmed to max threshold of 32", newBucket.sstables.size(), 32);
+
+ // one per bucket because they are all eligible and one more for the sstables that were trimmed
+ assertEquals("there should be one estimated remaining task per eligible bucket", buckets.keySet().size() + 1, newBucket.estimatedRemainingTasks);
}
@@@ -291,64 -279,4 +297,67 @@@
t.transaction.abort();
}
+ @Test
+ public void testDropOverlappingExpiredSSTables() throws InterruptedException
+ {
+ Keyspace keyspace = Keyspace.open(KEYSPACE1);
+ ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF_STANDARD1);
+ cfs.truncateBlocking();
+ cfs.disableAutoCompaction();
+
++ long timestamp = System.currentTimeMillis();
+ ByteBuffer value = ByteBuffer.wrap(new byte[100]);
+
- // create 2 sstables
- DecoratedKey key = Util.dk(String.valueOf("expired"));
- new RowUpdateBuilder(cfs.metadata, System.currentTimeMillis(), 1, key.getKey())
++ // Create a expiring sstable with a TTL
++ DecoratedKey key = Util.dk("expired");
++ new RowUpdateBuilder(cfs.metadata, timestamp, TTL_SECONDS, key.getKey())
+ .clustering("column")
+ .add("val", value).build().applyUnsafe();
+
+ cfs.forceBlockingFlush();
+ SSTableReader expiredSSTable = cfs.getLiveSSTables().iterator().next();
+ Thread.sleep(10);
+
- new RowUpdateBuilder(cfs.metadata, System.currentTimeMillis() - 1000, key.getKey())
++ // Create a second sstable without TTL and with a row superceded by the expiring row
++ new RowUpdateBuilder(cfs.metadata, timestamp - 1000, key.getKey())
+ .clustering("column")
+ .add("val", value).build().applyUnsafe();
- key = Util.dk(String.valueOf("nonexpired"));
- new RowUpdateBuilder(cfs.metadata, System.currentTimeMillis(), key.getKey())
++ key = Util.dk("nonexpired");
++ new RowUpdateBuilder(cfs.metadata, timestamp, key.getKey())
+ .clustering("column")
+ .add("val", value).build().applyUnsafe();
+
+ cfs.forceBlockingFlush();
+ assertEquals(cfs.getLiveSSTables().size(), 2);
+
+ Map<String, String> options = new HashMap<>();
-
+ options.put(TimeWindowCompactionStrategyOptions.COMPACTION_WINDOW_SIZE_KEY, "30");
+ options.put(TimeWindowCompactionStrategyOptions.COMPACTION_WINDOW_UNIT_KEY, "SECONDS");
+ options.put(TimeWindowCompactionStrategyOptions.TIMESTAMP_RESOLUTION_KEY, "MILLISECONDS");
+ options.put(TimeWindowCompactionStrategyOptions.EXPIRED_SSTABLE_CHECK_FREQUENCY_SECONDS_KEY, "0");
+ TimeWindowCompactionStrategy twcs = new TimeWindowCompactionStrategy(cfs, options);
+ for (SSTableReader sstable : cfs.getLiveSSTables())
+ twcs.addSSTable(sstable);
+
+ twcs.startup();
- assertNull(twcs.getNextBackgroundTask((int) (System.currentTimeMillis() / 1000)));
- Thread.sleep(2000);
- assertNull(twcs.getNextBackgroundTask((int) (System.currentTimeMillis()/1000)));
++ assertNull(twcs.getNextBackgroundTask(nowInSeconds()));
++
++ // Wait for the expiration of the first sstable
++ Thread.sleep(TimeUnit.SECONDS.toMillis(TTL_SECONDS + 1));
++ assertNull(twcs.getNextBackgroundTask(nowInSeconds()));
+
+ options.put(TimeWindowCompactionStrategyOptions.UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION_KEY, "true");
+ twcs = new TimeWindowCompactionStrategy(cfs, options);
+ for (SSTableReader sstable : cfs.getLiveSSTables())
+ twcs.addSSTable(sstable);
+
+ twcs.startup();
- AbstractCompactionTask t = twcs.getNextBackgroundTask((int) (System.currentTimeMillis()/1000));
++ AbstractCompactionTask t = twcs.getNextBackgroundTask(nowInSeconds());
+ assertNotNull(t);
+ assertEquals(1, Iterables.size(t.transaction.originals()));
+ SSTableReader sstable = t.transaction.originals().iterator().next();
+ assertEquals(sstable, expiredSSTable);
+ twcs.shutdown();
+ t.transaction.abort();
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org