Merge branch 'cassandra-5.0' into trunk

CHANGES.txt

@@ -5,6 +5,7 @@
 Merged from 5.0:
  * Backport Automated Repair Inside Cassandra for CEP-37 (CASSANDRA-21138)
 Merged from 4.1:
+ * Harden data resurrection startup check with atomic heartbeat file write with fallback (CASSANDRA-21290)
 Merged from 4.0:
  * Rate limit password changes (CASSANDRA-21202)
 

src/java/org/apache/cassandra/service/DataResurrectionCheck.java

@@ -19,6 +19,7 @@
 package org.apache.cassandra.service;
 
 import java.io.IOException;
+import java.nio.file.Files;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -46,6 +47,7 @@ import org.apache.cassandra.io.util.File;
 import org.apache.cassandra.schema.KeyspaceMetadata;
 import org.apache.cassandra.schema.SchemaKeyspace;
 import org.apache.cassandra.utils.Clock;
+import org.apache.cassandra.utils.Hex;
 import org.apache.cassandra.utils.JsonUtils;
 import org.apache.cassandra.utils.Pair;
 
@@ -54,7 +56,6 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static java.util.stream.Collectors.joining;
 import static java.util.stream.Collectors.toList;
 import static java.util.stream.Collectors.toSet;
-import static org.apache.cassandra.exceptions.StartupException.ERR_WRONG_DISK_STATE;
 import static org.apache.cassandra.exceptions.StartupException.ERR_WRONG_MACHINE_STATE;
 import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
 
@@ -87,12 +88,24 @@ public class DataResurrectionCheck implements StartupCheck
 
         public void serializeToJsonFile(File outputFile) throws IOException
         {
-            JsonUtils.serializeToJsonFile(this, outputFile);
+            JsonUtils.serializeToJsonFileAtomic(this, outputFile);
         }
 
         public static Heartbeat deserializeFromJsonFile(File file) throws IOException
         {
-            return JsonUtils.deserializeFromJsonFile(Heartbeat.class, file);
+            byte[] bytes = Files.readAllBytes(file.toPath());
+            try
+            {
+                return JsonUtils.deserializeFromJsonBytes(Heartbeat.class, bytes);
+            }
+            catch (IOException ex)
+            {
+                int maxLogBytes = Math.min(bytes.length, 1024);
+                String hexContent = bytes.length > 0 ? Hex.bytesToHex(bytes, 0, maxLogBytes) : "(empty)";
+                LOGGER.error("Failed to deserialize heartbeat file {} (length: {} bytes, first {} bytes hex: {})",
+                             file, bytes.length, maxLogBytes, hexContent, ex);
+                throw ex;
+            }
         }
 
         @Override
@@ -186,7 +199,10 @@ public class DataResurrectionCheck implements StartupCheck
         }
         catch (IOException ex)
         {
-            throw new StartupException(ERR_WRONG_DISK_STATE, "Failed to deserialize heartbeat file " + heartbeatFile);
+            LOGGER.warn("Failed to deserialize heartbeat file {}. Falling back to file last modified time.",
+                        heartbeatFile, ex);
+            Instant lastModified = Instant.ofEpochMilli(heartbeatFile.lastModified());
+            heartbeat = new Heartbeat(lastModified);
         }
 
         if (heartbeat.lastHeartbeat == null)
@@ -310,7 +326,7 @@ public class DataResurrectionCheck implements StartupCheck
     List<TableGCPeriod> getTablesGcPeriods(String userKeyspace)
     {
         Optional<KeyspaceMetadata> keyspaceMetadata = SchemaKeyspace.fetchNonSystemKeyspaces().get(userKeyspace);
-        if (!keyspaceMetadata.isPresent())
+        if (keyspaceMetadata.isEmpty())
             return Collections.emptyList();
 
         KeyspaceMetadata ksmd = keyspaceMetadata.get();

src/java/org/apache/cassandra/utils/JsonUtils.java

@@ -194,6 +194,38 @@ public final class JsonUtils
         }
     }
 
+    public static void serializeToJsonFileAtomic(Object object, File outputFile) throws IOException
+    {
+        // Try to write then perform atomic move so that file can't be corrupted
+        // by process crash in the middle of the writing.
+        File tempFile = new File(outputFile.path() + ".tmp");
+        try
+        {
+            // Serialize to bytes first so we can flush and fsync before close.
+            // Jackson's writeValue(OutputStream, ...) auto-closes the stream,
+            // which would prevent us from calling sync() afterwards.
+            byte[] data = JSON_OBJECT_PRETTY_WRITER.writeValueAsBytes(object);
+            try (FileOutputStreamPlus out = tempFile.newOutputStream(OVERWRITE))
+            {
+                out.write(data);
+                // Force data to disk before rename to ensure durability.
+                // Without this, a crash after rename but before OS flushes to disk
+                // can leave the file with zero-filled or corrupted blocks.
+                out.sync();
+            }
+            tempFile.move(outputFile);
+            // Fsync the parent directory to ensure the rename is durable.
+            // Without this, a crash after rename can revert to the old directory entry.
+            // See: https://transactional.blog/how-to-learn/disk-io
+            SyncUtil.trySyncDir(outputFile.parent());
+        }
+        catch (IOException ex)
+        {
+            tempFile.deleteIfExists();
+            throw ex;
+        }
+    }
+
     public static <T> T deserializeFromJsonFile(Class<T> tClass, File file) throws IOException
     {
         try (FileInputStreamPlus in = file.newInputStream())
@@ -202,6 +234,11 @@ public final class JsonUtils
         }
     }
 
+    public static <T> T deserializeFromJsonBytes(Class<T> tClass, byte[] bytes) throws IOException
+    {
+        return JSON_OBJECT_MAPPER.readValue(bytes, tClass);
+    }
+
     /**
      * Handles unquoting and case-insensitivity in map keys.
      */

test/unit/org/apache/cassandra/service/StartupChecksTest.java

@@ -18,11 +18,13 @@
 package org.apache.cassandra.service;
 
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.FileStore;
 import java.nio.file.FileSystem;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.nio.file.attribute.FileTime;
 import java.nio.file.spi.FileSystemProvider;
 import java.time.Instant;
 import java.util.HashMap;
@@ -30,13 +32,13 @@ import java.util.List;
 import java.util.Map;
 import java.util.ServiceLoader;
 import java.util.Set;
+import java.util.UUID;
 import java.util.concurrent.Callable;
 import java.util.stream.Collectors;
 
 import com.vdurmont.semver4j.Semver;
 
 import org.junit.After;
-import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.Assume;
 import org.junit.Before;
@@ -95,14 +97,13 @@ public class StartupChecksTest
 
     StartupChecks startupChecks;
     Path sstableDir;
-    static File heartbeatFile;
+    File heartbeatFile;
 
     StartupChecksConfiguration options = new StartupChecksConfiguration(new StartupChecks().withDefaultTests(), new HashMap<>());
 
     @BeforeClass
     public static void setupServer()
     {
-        heartbeatFile = createTempFile("cassandra-heartbeat-", "");
         SchemaLoader.prepareServer();
     }
 
@@ -118,6 +119,8 @@ public class StartupChecksTest
         sstableDir = Paths.get(dataDir.absolutePath(), "Keyspace1", "Standard1");
         Files.createDirectories(sstableDir);
 
+        heartbeatFile = createTempFile("cassandra-heartbeat-" + UUID.randomUUID(), "");
+
         options.enable("check_data_resurrection");
         options.getConfig("check_data_resurrection")
                .put(HEARTBEAT_FILE_CONFIG_PROPERTY, heartbeatFile.absolutePath());
@@ -129,11 +132,6 @@ public class StartupChecksTest
     public void tearDown() throws IOException
     {
         new File(sstableDir).deleteRecursive();
-    }
-
-    @AfterClass
-    public static void tearDownClass()
-    {
         heartbeatFile.delete();
     }
 
@@ -250,6 +248,53 @@ public class StartupChecksTest
         verifyFailure(startupChecks, "Invalid tables: abc.def");
     }
 
+    @Test
+    public void testDataResurrectionCheckLastModifiedFallback() throws Exception
+    {
+        DataResurrectionCheck check = new DataResurrectionCheck() {
+            @Override
+            List<String> getKeyspaces()
+            {
+                return singletonList("test_ks");
+            }
+
+            @Override
+            List<TableGCPeriod> getTablesGcPeriods(String userKeyspace)
+            {
+                return singletonList(new TableGCPeriod("test_table", 10));
+            }
+        };
+
+        int originalHintWindow = DatabaseDescriptor.getMaxHintWindow();
+        try
+        {
+            DatabaseDescriptor.setMaxHintWindow(5 * 1000);
+
+            // Empty file
+            Files.write(heartbeatFile.toPath(), "".getBytes(StandardCharsets.UTF_8));
+            Instant recentTimestamp = Instant.ofEpochMilli(Clock.Global.currentTimeMillis());
+            Files.setLastModifiedTime(heartbeatFile.toPath(), FileTime.from(recentTimestamp));
+
+            startupChecks.withTest(check);
+            verifySuccess(startupChecks);
+        }
+        finally
+        {
+            DatabaseDescriptor.setMaxHintWindow(originalHintWindow);
+        }
+    }
+
+    private void verifySuccess(StartupChecks tests) {
+        try
+        {
+            tests.verify(options);
+        }
+        catch (StartupException e)
+        {
+            fail("Failed startup check with error: " + e.getMessage());
+        }
+    }
+
     @Test
     public void testKernelBug1057843Check() throws Exception
     {