Merge branch 'cassandra-5.0' into trunk

CHANGES.txt

@@ -253,6 +253,7 @@ Merged from 4.1:
  * Enforce CQL message size limit on multiframe messages (CASSANDRA-20052)
  * Fix race condition in DecayingEstimatedHistogramReservoir during rescale (CASSANDRA-19365)
 Merged from 4.0:
+ * Handle sstable metadata stats file getting a new mtime after compaction has finished (CASSANDRA-18119)
  * Honor MAX_PARALLEL_TRANSFERS correctly (CASSANDRA-20532)
  * Updating a column with a new TTL but same expiration time is non-deterministic and causes repair mismatches. (CASSANDRA-20561)
  * Avoid computing prepared statement size for unprepared batches (CASSANDRA-20556)

src/java/org/apache/cassandra/db/lifecycle/LogFile.java

@@ -179,6 +179,10 @@ final class LogFile implements AutoCloseable
         this.id = id;
     }
 
+    /**
+     * Check a variety of the internals of the LogRecord as well as the state of the LogRecord vs. the files found on disk
+     * to ensure they remain correct and nothing was changed external to the process.
+     */
     boolean verify()
     {
         records.clear();
@@ -245,6 +249,9 @@ final class LogFile implements AutoCloseable
         return record;
     }
 
+    /**
+     * Sets the {@link LogRecord.Status#error} if something wrong is found with the record.
+     */
     static void verifyRecord(LogRecord record, List<File> existingFiles)
     {
         if (record.checksum != record.computeChecksum())
@@ -256,6 +263,7 @@ final class LogFile implements AutoCloseable
             return;
         }
 
+        // If it's not a removal we don't check it since we're not going to take action on it
         if (record.type != Type.REMOVE)
             return;
 
@@ -269,6 +277,16 @@ final class LogFile implements AutoCloseable
         // we can have transaction files with mismatching updateTime resolutions due to switching between jdk8 and jdk11, truncate both to be consistent:
         if (truncateMillis(record.updateTime) != truncateMillis(record.status.onDiskRecord.updateTime) && record.status.onDiskRecord.updateTime > 0)
         {
+            // handle the case where we have existing broken transaction file on disk, where the update time is
+            // based on the stats file. This is just for the first upgrade, patched versions never base the update
+            // time on the stats file.
+            LogRecord statsIncluded = LogRecord.make(record.type, existingFiles, existingFiles.size(), record.absolutePath(), true);
+            if (truncateMillis(statsIncluded.updateTime) == truncateMillis(record.updateTime))
+            {
+                logger.warn("Found a legacy log record {} with updateTime based on the stats file, ignoring to allow startup to continue", record);
+                return;
+            }
+
             record.setError(String.format("Unexpected files detected for sstable [%s]: " +
                                           "last update time [%tc] (%d) should have been [%tc] (%d)",
                                           record.fileName(),
@@ -276,7 +294,6 @@ final class LogFile implements AutoCloseable
                                           record.status.onDiskRecord.updateTime,
                                           record.updateTime,
                                           record.updateTime));
-
         }
     }
 

src/java/org/apache/cassandra/db/lifecycle/LogRecord.java

@@ -20,7 +20,6 @@
  */
 package org.apache.cassandra.db.lifecycle;
 
-
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -38,14 +37,21 @@ import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import java.util.zip.CRC32;
 
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import org.apache.cassandra.io.sstable.Component;
+import org.apache.cassandra.io.sstable.Descriptor;
 import org.apache.cassandra.io.sstable.SSTable;
+import org.apache.cassandra.io.sstable.format.SSTableFormat;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.io.util.File;
 import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.io.util.PathUtils;
 import org.apache.cassandra.utils.FBUtilities;
 
+import static org.apache.cassandra.io.sstable.Descriptor.TMP_EXT;
 import static org.apache.cassandra.utils.LocalizeString.toUpperCaseLocalized;
 
 /**
@@ -55,6 +61,10 @@ import static org.apache.cassandra.utils.LocalizeString.toUpperCaseLocalized;
  */
 final class LogRecord
 {
+    private static final Logger logger = LoggerFactory.getLogger(LogRecord.class);
+    @VisibleForTesting
+    static boolean INCLUDE_STATS_FOR_TESTS = false;
+
     public enum Type
     {
         UNKNOWN, // a record that cannot be parsed
@@ -78,7 +88,10 @@ final class LogRecord
             return this == record.type;
         }
 
-        public boolean isFinal() { return this == Type.COMMIT || this == Type.ABORT; }
+        public boolean isFinal()
+        {
+            return this == Type.COMMIT || this == Type.ABORT;
+        }
     }
 
     /**
@@ -194,17 +207,65 @@ final class LogRecord
 
     public LogRecord withExistingFiles(List<File> existingFiles)
     {
+        if (!absolutePath.isPresent())
+            throw new IllegalStateException(String.format("Cannot create record from existing files for type %s - file is not present", type));
+
         return make(type, existingFiles, 0, absolutePath.get());
     }
 
+    /**
+     * We create a LogRecord based on the files on disk; there's some subtlety around how we handle stats files as the
+     * timestamp can be mutated by the async completion of compaction if things race with node shutdown. To work around this,
+     * we don't take the stats file timestamp into account when calculating nor using the timestamps for all the components
+     * as we build the LogRecord.
+     */
     public static LogRecord make(Type type, List<File> files, int minFiles, String absolutePath)
     {
+        return make(type, files, minFiles, absolutePath, INCLUDE_STATS_FOR_TESTS);
+    }
+
+    /**
+     * In most cases we skip including the stats file timestamp entirely as it can be mutated during anticompaction
+     * and thus "invalidate" the LogRecord. There is an edge case where we have a LogRecord that was written w/the wrong
+     * timestamp (i.e. included a mutated stats file) and we need the node to come up, so we need to expose the selective
+     * ability to either include the stats file timestamp or not.
+     *
+     * See {@link LogFile#verifyRecord}
+     */
+    static LogRecord make(Type type, List<File> files, int minFiles, String absolutePath, boolean includeStatsFile)
+    {
+        List<File> toVerify;
+        File statsFile = null;
+        if (!includeStatsFile && !files.isEmpty())
+        {
+            toVerify = new ArrayList<>(files.size() - 1);
+            for (File f : files)
+            {
+                if (!f.name().endsWith(TMP_EXT))
+                {
+                    if (Descriptor.componentFromFile(f) == SSTableFormat.Components.STATS)
+                        statsFile = f;
+                    else
+                        toVerify.add(f);
+                }
+            }
+        }
+        else
+        {
+            toVerify = files;
+        }
         // CASSANDRA-11889: File.lastModified() returns a positive value only if the file exists, therefore
         // we filter by positive values to only consider the files that still exists right now, in case things
         // changed on disk since getExistingFiles() was called
-        List<Long> positiveModifiedTimes = files.stream().map(File::lastModified).filter(lm -> lm > 0).collect(Collectors.toList());
+        List<Long> positiveModifiedTimes = toVerify.stream().map(File::lastModified).filter(lm -> lm > 0).collect(Collectors.toList());
         long lastModified = positiveModifiedTimes.stream().reduce(0L, Long::max);
-        return new LogRecord(type, absolutePath, lastModified, Math.max(minFiles, positiveModifiedTimes.size()));
+
+        // We need to preserve the file count for the number of existing files found on disk even though we ignored the
+        // stats file during our timestamp calculation. If the stats file still exists, we add in the count of it as
+        // a separate validation assumption that it's one of the files considered valid in this LogRecord.
+        boolean addStatTS = statsFile != null && statsFile.exists();
+        int positiveTSCount = addStatTS ? positiveModifiedTimes.size() + 1 : positiveModifiedTimes.size();
+        return new LogRecord(type, absolutePath, lastModified, Math.max(minFiles, positiveTSCount));
     }
 
     private LogRecord(Type type, long updateTime)

src/java/org/apache/cassandra/db/lifecycle/LogTransaction.java

@@ -549,6 +549,8 @@ class LogTransaction extends Transactional.AbstractTransactional implements Tran
             try(LogFile txn = LogFile.make(entry.getKey(), entry.getValue()))
             {
                 logger.info("Verifying logfile transaction {}", txn);
+                // We don't check / include the stats file timestamp on LogRecord creation / verification as that might
+                // be modified by a race in compaction notification and then needlessly fail subsequent node starts.
                 if (txn.verify())
                 {
                     Throwable failure = txn.removeUnfinishedLeftovers(null);

test/unit/org/apache/cassandra/db/lifecycle/LogTransactionTest.java

@@ -17,7 +17,6 @@
  */
 package org.apache.cassandra.db.lifecycle;
 
-
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.nio.file.Files;
@@ -38,6 +37,7 @@ import java.util.stream.Stream;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Sets;
+import org.apache.cassandra.db.streaming.ComponentContext;
 import org.junit.Test;
 
 import org.apache.cassandra.Util;
@@ -79,6 +79,7 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
@@ -1260,12 +1261,173 @@ public class LogTransactionTest extends AbstractTransactionalTest
         logs.finish();
     }
 
+    @Test
+    public void testStatsTSMatchOnStart() throws Throwable
+    {
+        ColumnFamilyStore cfs = MockSchema.newCFS(KEYSPACE);
+        File dataFolder = new Directories(cfs.metadata()).getDirectoryForNewSSTables();
+        SSTableReader sstable = sstable(dataFolder, cfs, 0, 128);
+
+        try(LogTransaction log = new LogTransaction(OperationType.COMPACTION))
+        {
+            assertNotNull(log);
+            log.trackNew(sstable);
+
+            // Confirm we can remove leftovers when they match
+            LogTransaction.removeUnfinishedLeftovers(cfs.metadata());
+        }
+
+        File sFile = sstable.descriptor.fileFor(SSTableFormat.Components.STATS);
+        assertFalse("Found STATS file but expected it to be cleaned up.", Files.exists(sFile.toPath()));
+        sstable.selfRef().release();
+    }
+
+    @Test
+    public void testStatsTSMatchDuringList() throws Throwable
+    {
+        ColumnFamilyStore cfs = MockSchema.newCFS(KEYSPACE);
+        File dataFolder = new Directories(cfs.metadata()).getDirectoryForNewSSTables();
+        SSTableReader sstable = sstable(dataFolder, cfs, 0, 128);
+
+        try(LogTransaction log = new LogTransaction(OperationType.COMPACTION))
+        {
+            assertNotNull(log);
+            log.trackNew(sstable);
+
+            // Confirm we can successfully classify files when they match - this triggers the LogAwareFileLister verify
+            listFiles(dataFolder, Directories.OnTxnErr.THROW, Directories.FileType.FINAL);
+        }
+        sstable.selfRef().release();
+    }
+
+    @Test
+    public void testStatsTSMismatchDuringStart() throws Throwable
+    {
+        ColumnFamilyStore cfs = MockSchema.newCFS(KEYSPACE);
+        File dataFolder = new Directories(cfs.metadata()).getDirectoryForNewSSTables();
+        SSTableReader sstable = sstable(dataFolder, cfs, 0, 128);
+
+        File sFile = sstable.descriptor.fileFor(SSTableFormat.Components.STATS);
+        assertTrue("STATS file not created successfully in test setup", Files.exists(sFile.toPath()));
+
+        // Confirm we can remove leftovers even if the STATS file doesn't match
+        try(LogTransaction log = new LogTransaction(OperationType.COMPACTION))
+        {
+            assertNotNull(log);
+
+            // Need to flag the transaction as having a REMOVE entry so it'll trigger the path to calculate stats on list
+            log.obsoleted(sstable);
+
+            // Need to sleep for long enough to bypass the millisecond truncation logic due to jdk8 and jdk11 change
+            Thread.sleep(2000);
+            assertTrue("Failed to set mtime for STATS file to currentTimeMillis()", sFile.trySetLastModified(System.currentTimeMillis()));
+
+            // Confirm we have an mtime mismatch
+            File dFile = sstable.descriptor.fileFor(SSTableFormat.Components.DATA);
+            assertNotEquals(sFile.lastModified(), dFile.lastModified());
+
+            // We need to add another LogRecord as we allow partial or incorrect entries as the last record...
+            log.trackNew(sstable(dataFolder, cfs, 2, 128));
+
+            assertTrue("STATS file gone before removeUnfinished...", Files.exists(sFile.toPath()));
+            // Confirm we can remove leftovers when the STATS file mismatches
+            log.prepareToCommit(); // commit so that obsolete sstable components will be removed.
+            log.commit();
+            ComponentContext.create(sstable);
+            assertTrue(LogTransaction.removeUnfinishedLeftovers(cfs.metadata()));
+        }
+
+        sstable.selfRef().release();
+    }
+
+    @Test
+    public void testWrongTimestampInTxnFile() throws IOException, InterruptedException
+    {
+        ColumnFamilyStore cfs = MockSchema.newCFS(KEYSPACE);
+        File dataFolder = new Directories(cfs.metadata()).getDirectoryForNewSSTables();
+        SSTableReader sstable = sstable(dataFolder, cfs, 0, 128);
+
+        File sFile = sstable.descriptor.fileFor(SSTableFormat.Components.STATS);
+        assertTrue("STATS file not created successfully in test setup", Files.exists(sFile.toPath()));
+
+        LogRecord.INCLUDE_STATS_FOR_TESTS = true;
+
+        Thread.sleep(2000);
+        assertTrue("Failed to set mtime for STATS file to currentTimeMillis()", sFile.trySetLastModified(System.currentTimeMillis()));
+
+        // Confirm we can remove leftovers even if the STATS file doesn't match
+        try(LogTransaction log = new LogTransaction(OperationType.COMPACTION))
+        {
+            assertNotNull(log);
+            // Need to flag the transaction as having a REMOVE entry so it'll trigger the path to calculate stats on list
+            log.obsoleted(sstable);
+            // Need to sleep for long enough to bypass the millisecond truncation logic due to jdk8 and jdk11 change
+            // Confirm we have an mtime mismatch
+            File dFile = sstable.descriptor.fileFor(SSTableFormat.Components.DATA);
+            assertNotEquals(sFile.lastModified(), dFile.lastModified());
+
+            // We need to add another LogRecord as we allow partial or incorrect entries as the last record...
+            log.trackNew(sstable(dataFolder, cfs, 2, 128));
+
+            assertTrue("STATS file gone before removeUnfinished...", Files.exists(sFile.toPath()));
+            // Confirm we can remove leftovers when the STATS file mismatches
+            LogRecord.INCLUDE_STATS_FOR_TESTS = false;
+            assertTrue(LogTransaction.removeUnfinishedLeftovers(cfs.metadata()));
+        }
+
+        sstable.selfRef().release();
+    }
+
+    /**
+     * We do not consider the stats file's ts for any cases at this point
+     */
+    @Test
+    public void testStatsTSMismatchDuringList() throws Throwable
+    {
+        SSTableReader sstable = null;
+        try
+        {
+            ColumnFamilyStore cfs = MockSchema.newCFS(KEYSPACE);
+            File dataFolder = new Directories(cfs.metadata()).getDirectoryForNewSSTables();
+            sstable = sstable(dataFolder, cfs, 0, 128);
+            File sFile = sstable.descriptor.fileFor(SSTableFormat.Components.STATS);
+            assertTrue("STATS file not created successfully in test setup", Files.exists(sFile.toPath()));
+
+            try(LogTransaction log = new LogTransaction(OperationType.COMPACTION))
+            {
+                assertNotNull(log);
+
+                // Need to flag the transaction as having a REMOVE entry so it'll trigger the path to calculate stats on list
+                log.obsoleted(sstable);
+
+                // Need to sleep for long enough to bypass the millisecond truncation logic due to jdk8 and jdk11 change
+                Thread.sleep(2000);
+                assertTrue("Failed to set mtime for STATS file to currentTimeMillis()", sFile.trySetLastModified(System.currentTimeMillis()));
+
+                // Confirm we have an mtime mismatch
+                File dFile = sstable.descriptor.fileFor(SSTableFormat.Components.DATA);
+                assertNotEquals(sFile.lastModified(), dFile.lastModified());
+
+                // We need to add another LogRecord as we allow partial or incorrect entries as the last record...
+                log.trackNew(sstable(dataFolder, cfs, 2, 128));
+
+                // Confirm we don't get a mismatch LogRecord error when the STATS file is different even on listFiles case
+                listFiles(dataFolder, Directories.OnTxnErr.THROW, Directories.FileType.FINAL);
+            }
+        }
+        finally
+        {
+            if (sstable != null)
+                sstable.selfRef().release();
+        }
+    }
+
     private static SSTableReader sstable(File dataFolder, ColumnFamilyStore cfs, int generation, int size) throws IOException
     {
         Descriptor descriptor = new Descriptor(dataFolder, cfs.getKeyspaceName(), cfs.getTableName(), new SequenceBasedSSTableId(generation), DatabaseDescriptor.getSelectedSSTableFormat());
         if (BigFormat.isSelected())
         {
-            Set<Component> components = ImmutableSet.of(Components.DATA, Components.PRIMARY_INDEX, Components.FILTER, Components.TOC);
+            Set<Component> components = ImmutableSet.of(Components.DATA, Components.PRIMARY_INDEX, Components.FILTER, Components.TOC, Components.STATS);
             for (Component component : components)
             {
                 File file = descriptor.fileFor(component);
@@ -1399,12 +1561,12 @@ public class LogTransactionTest extends AbstractTransactionalTest
 
     static Set<File> getTemporaryFiles(File folder)
     {
-        return listFiles(folder, Directories.FileType.TEMPORARY);
+        return listFiles(folder, Directories.OnTxnErr.IGNORE, Directories.FileType.TEMPORARY);
     }
 
     static Set<File> getFinalFiles(File folder)
     {
-        return listFiles(folder, Directories.FileType.FINAL);
+        return listFiles(folder, Directories.OnTxnErr.IGNORE, Directories.FileType.FINAL);
     }
 
     // Used by listFiles - this test is deliberately racing with files being
@@ -1429,12 +1591,12 @@ public class LogTransactionTest extends AbstractTransactionalTest
         }
     }
 
-    static Set<File> listFiles(File folder, Directories.FileType... types)
+    static Set<File> listFiles(File folder, Directories.OnTxnErr err, Directories.FileType... types)
     {
         Collection<Directories.FileType> match = Arrays.asList(types);
         return new LogAwareFileLister(folder.toPath(),
                                       (file, type) -> match.contains(type),
-                                      Directories.OnTxnErr.IGNORE).list()
+                                      err).list()
                        .stream()
                        .flatMap(LogTransactionTest::toCanonicalIgnoringNotFound)
                        .collect(Collectors.toSet());