Merge branch 'cassandra-5.0' into trunk

Author: krummas

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 @@ static boolean isLogFile(File file)
         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 @@ LogRecord setErrorInReplicas(LogRecord record)
         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 @@ static void verifyRecord(LogRecord record, List<File> existingFiles)
             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,14 +277,23 @@ static void verifyRecord(LogRecord record, List<File> existingFiles)
         // 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(),
                                           record.status.onDiskRecord.updateTime,
                                           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.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 @@
  */
 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 @@ public boolean matches(LogRecord record)
             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 @@ private static String absolutePath(File baseFile)
 
     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 @@ static boolean removeUnfinishedLeftovers(Map.Entry<String, List<File>> entry)
             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);