CASSSIDECAR-484: Fix the spurious oldest segment age in CdcRawDirectorySpaceCleaner

CHANGES.txt

@@ -1,5 +1,6 @@
 0.5.0
 -----
+ * Fix the spurious oldest segment age in CdcRawDirectorySpaceCleaner (CASSSIDECAR-484)
  * Wire CDC configs in configs table to SidecarCdcOptions/SidecarStatePersister (CASSSIDECAR-483)
  * Implement durable operational job tracker (CASSSIDECAR-374)
  * Remove filesystem path from Http response (CASSSIDECAR-477)

server/src/main/java/org/apache/cassandra/sidecar/tasks/CdcRawDirectorySpaceCleaner.java

@@ -22,6 +22,7 @@
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.nio.file.attribute.BasicFileAttributes;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
@@ -232,8 +233,13 @@ protected void cleanUpCdcRawDirectory(File cdcRawDirectory)
         Collections.sort(segmentFiles);
         long nowInMillis = timeProvider.currentTimeMillis();
 
-        // track the age of the oldest commit log segment to give indication of the time-window buffer available
-        cdcMetrics.oldestSegmentAge.metric.setValue((int) MILLISECONDS.toSeconds(nowInMillis - segmentFiles.get(0).lastModified()));
+        // track the age of the oldest commit log segment to give indication of the time-window buffer available.
+        // Skip emission if lastModified is 0 (file was deleted before we could snapshot its lastModified time).
+        long oldestLastModified = segmentFiles.get(0).lastModified();
+        if (oldestLastModified > 0)
+        {
+            cdcMetrics.oldestSegmentAge.metric.setValue((int) MILLISECONDS.toSeconds(nowInMillis - oldestLastModified));
+        }
 
         LOGGER.debug("Cdc data cleaner directorySizeBytes={} maxedUsageBytes={} upperLimitBytes={}",
                      directorySizeBytes, maxUsageBytes, upperLimitBytes);
@@ -254,31 +260,47 @@ protected void cleanUpCdcRawDirectory(File cdcRawDirectory)
             while (i < segmentFiles.size() - 1 && directorySizeBytes > upperLimitBytes)
             {
                 CdcRawSegmentFile segment = segmentFiles.get(i);
-                long ageMillis = nowInMillis - segment.lastModified();
-
-                if (ageMillis < criticalMillis)
-                {
-                    LOGGER.error("Insufficient Cdc buffer size to maintain {}-minute window segment={} maxSize={} ageMinutes={}",
-                                 MILLISECONDS.toMinutes(criticalMillis), segment, upperLimitBytes,
-                                 MILLISECONDS.toMinutes(ageMillis));
-                    cdcMetrics.criticalCdcRawSpace.metric.update(1);
-                }
-                else if (ageMillis < lowMillis)
-                {
-                    LOGGER.warn("Insufficient Cdc buffer size to maintain {}-minute window segment={} maxSize={} ageMinutes={}",
-                                MILLISECONDS.toMinutes(lowMillis), segment, upperLimitBytes,
-                                MILLISECONDS.toMinutes(ageMillis));
-                    cdcMetrics.lowCdcRawSpace.metric.update(1);
-                }
+                long segmentLastModified = segment.lastModified();
                 long length = 0;
-                try
+                // When lastModified is 0 the segment was already reclaimed between the directory
+                // Skip the buffer-window alerts and the metrics update but still discount the cached size
+                // from the local budget so the outer loop stops at the right point instead of over-cleaning 
+                // subsequent live segments.
+                if (segmentLastModified > 0)
                 {
-                    length = deleteSegment(segment);
-                    cdcMetrics.deletedSegment.metric.update(length);
+                    long ageMillis = nowInMillis - segmentLastModified;
+
+                    if (ageMillis < criticalMillis)
+                    {
+                        LOGGER.error("Insufficient Cdc buffer size to maintain {}-minute window segment={} maxSize={} ageMinutes={}",
+                                     MILLISECONDS.toMinutes(criticalMillis), segment, upperLimitBytes,
+                                     MILLISECONDS.toMinutes(ageMillis));
+                        cdcMetrics.criticalCdcRawSpace.metric.update(1);
+                    }
+                    else if (ageMillis < lowMillis)
+                    {
+                        LOGGER.warn("Insufficient Cdc buffer size to maintain {}-minute window segment={} maxSize={} ageMinutes={}",
+                                    MILLISECONDS.toMinutes(lowMillis), segment, upperLimitBytes,
+                                    MILLISECONDS.toMinutes(ageMillis));
+                        cdcMetrics.lowCdcRawSpace.metric.update(1);
+                    }
+
+                    try
+                    {
+                        length = deleteSegment(segment);
+                        cdcMetrics.deletedSegment.metric.update(length);
+                    }
+                    catch (IOException e)
+                    {
+                        LOGGER.warn("Failed to delete cdc segment", e);
+                    }
                 }
-                catch (IOException e)
+                else
                 {
-                    LOGGER.warn("Failed to delete cdc segment", e);
+                    LOGGER.warn("Skipping delete for already-reclaimed cdc segment {}; a concurrent reclaim "
+                                + "(e.g. Cassandra reclaiming cdc_raw) occurred between the directory scan and "
+                                + "the deletion pass", segment);
+                    length = segment.length();
                 }
                 directorySizeBytes -= length;
                 i++;
@@ -398,14 +420,33 @@ protected static class CdcRawSegmentFile implements Comparable<CdcRawSegmentFile
         private final File indexFile;
         private final long segmentId;
         private final long len;
+        // Snapshot size and lastModified from a single readAttributes call so a concurrent
+        // change between separate stat fields cannot produce an inconsistent pair (e.g. len > 0
+        // with lastModified == 0). A later delete before the cleanup loop is still handled by
+        // the lastModified > 0 guards.
+        private final long lastModified;
 
         CdcRawSegmentFile(File logFile)
         {
             this.file = logFile;
             final String name = logFile.getName();
             this.segmentId = parseSegmentId(name);
-            this.len = logFile.length();
             this.indexFile = CdcUtil.getIdxFile(logFile);
+            long size = 0L;
+            long mtime = 0L;
+            try
+            {
+                BasicFileAttributes attrs = Files.readAttributes(logFile.toPath(), BasicFileAttributes.class);
+                size = attrs.size();
+                mtime = attrs.lastModifiedTime().toMillis();
+            }
+            catch (IOException e)
+            {
+                // File removed or unreadable between listFiles and construction; len/mtime
+                // stay 0 and the lastModified > 0 guards skip metric emission and deletion.
+            }
+            this.len = size;
+            this.lastModified = mtime;
         }
 
         public boolean exists()
@@ -430,7 +471,7 @@ public long indexLength()
 
         public long lastModified()
         {
-            return file.lastModified();
+            return lastModified;
         }
 
         public Path path()

server/src/test/java/org/apache/cassandra/sidecar/tasks/CdcRawDirectorySpaceCleanerTest.java

@@ -54,6 +54,7 @@
 import org.mockito.stubbing.Answer;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
@@ -140,6 +141,127 @@ void testCdcRawDirectorySpaceCleaner(@TempDir Path tempDir) throws IOException
         cleaner.routineCleanUp(); // it should run fine.
     }
 
+    @Test
+    void testOldestSegmentAgeNotPoisonedWhenTimestampUnavailable(@TempDir Path tempDir) throws IOException
+    {
+        // Large cdc_total_space so the cleaner does not delete anything; we only exercise the age metric.
+        CdcSystemViewsDatabaseAccessor systemViewsDatabaseAccessor = mock(CdcSystemViewsDatabaseAccessor.class);
+        when(systemViewsDatabaseAccessor.cdcTotalSpaceBytesSetting()).thenReturn(1L << 30); // 1 GiB
+        CdcConfiguration cdcConfiguration = new CdcConfigurationImpl();
+        ServiceConfiguration serviceConfiguration = mock(ServiceConfiguration.class);
+        when(serviceConfiguration.cdcConfiguration()).thenReturn(cdcConfiguration);
+
+        // Two small, completed (index-backed) segments so the cleaner proceeds past the < 2 files guard.
+        InstanceMetadata instanceMetadata = mock(InstanceMetadata.class);
+        File cdcDir = Files.createDirectory(tempDir.resolve(CdcRawDirectorySpaceCleaner.CDC_DIR_NAME)).toFile();
+        writeCdcSegment(cdcDir, TEST_SEGMENT_FILE_NAME_1, 1024, true);
+        writeCdcSegment(cdcDir, TEST_SEGMENT_FILE_NAME_2, 1024, true);
+        when(instanceMetadata.dataDirs()).thenReturn(List.of(cdcDir.getParent() + "/data"));
+        when(instanceMetadata.cdcDir()).thenReturn(cdcDir.getParent() + "/cdc_raw");
+        InstancesMetadata instancesMetadata = new InstancesMetadataImpl(instanceMetadata, DnsResolvers.DEFAULT);
+
+        // Pin "now" so the emitted age is deterministic: file mtime is exactly 120s before the cleaner's clock.
+        FakeTimeProvider timeProvider = new FakeTimeProvider();
+        long nowMillis = (System.currentTimeMillis() / 1000) * 1000; // second-aligned to dodge fs mtime rounding
+        timeProvider.advance(nowMillis, TimeUnit.MILLISECONDS);
+
+        // FILE_1 has the smallest segmentId, so it is the oldest and drives the gauge. Set it to 120s ago.
+        File oldest = new File(cdcDir, TEST_SEGMENT_FILE_NAME_1);
+        long twoMinutesAgoMillis = nowMillis - TimeUnit.SECONDS.toMillis(120);
+        assumeTrue(oldest.setLastModified(twoMinutesAgoMillis), "Filesystem does not support setting mtime");
+
+        CdcRawDirectorySpaceCleaner cleaner = new CdcRawDirectorySpaceCleaner(
+        timeProvider,
+        systemViewsDatabaseAccessor,
+        serviceConfiguration,
+        instancesMetadata,
+        mockSidecarMetrics
+        );
+
+        // First run: the oldest segment was last modified 120s before "now", so the gauge must report ~120s.
+        // Allow +/-1s to tolerate filesystems that store mtime at second granularity.
+        cleaner.routineCleanUp();
+        int ageAfterValid = cdcMetrics.oldestSegmentAge.metric.getValue();
+        assertThat(ageAfterValid).isBetween(119, 121);
+
+        // Simulate the oldest segment's timestamp being unavailable (concurrent delete/rotation makes
+        // File.lastModified() return 0). Skip on filesystems that do not support a 0 mtime.
+        assumeTrue(oldest.setLastModified(0L) && oldest.lastModified() == 0L,
+                   "Filesystem does not support a 0 mtime; skipping guard assertion");
+
+        // Second run: the guard must skip emitting, leaving the previous value intact instead of
+        // publishing the ~1.8e9s (~57 year) bogus value.
+        cleaner.routineCleanUp();
+        int ageAfterUnavailable = cdcMetrics.oldestSegmentAge.metric.getValue();
+        assertThat(ageAfterUnavailable)
+        .as("oldestSegmentAge must not be poisoned to a decades-large value when the timestamp is unavailable")
+        .isEqualTo(ageAfterValid)
+        .isLessThan(1_000_000);
+    }
+
+    @Test
+    void testDeletionLoopSkipsSegmentWhenTimestampUnavailable(@TempDir Path tempDir) throws IOException
+    {
+        // Small cdc_total_space so the deletion loop enters and must decide per-segment whether
+        // to call deleteSegment(). Three ~4 KiB segments against a 4 KiB budget → the loop
+        // processes seg 1 and seg 2 (seg 3 is retained as the last active segment).
+        CdcSystemViewsDatabaseAccessor systemViewsDatabaseAccessor = mock(CdcSystemViewsDatabaseAccessor.class);
+        when(systemViewsDatabaseAccessor.cdcTotalSpaceBytesSetting()).thenReturn(4L * 1024L);
+        CdcConfiguration cdcConfiguration = new CdcConfigurationImpl();
+        ServiceConfiguration serviceConfiguration = mock(ServiceConfiguration.class);
+        when(serviceConfiguration.cdcConfiguration()).thenReturn(cdcConfiguration);
+
+        InstanceMetadata instanceMetadata = mock(InstanceMetadata.class);
+        File cdcDir = Files.createDirectory(tempDir.resolve(CdcRawDirectorySpaceCleaner.CDC_DIR_NAME)).toFile();
+        writeCdcSegment(cdcDir, TEST_SEGMENT_FILE_NAME_1, 4096, true);
+        writeCdcSegment(cdcDir, TEST_SEGMENT_FILE_NAME_2, 4096, true);
+        writeCdcSegment(cdcDir, TEST_SEGMENT_FILE_NAME_3, 4096, true);
+        when(instanceMetadata.dataDirs()).thenReturn(List.of(cdcDir.getParent() + "/data"));
+        when(instanceMetadata.cdcDir()).thenReturn(cdcDir.getParent() + "/cdc_raw");
+        InstancesMetadata instancesMetadata = new InstancesMetadataImpl(instanceMetadata, DnsResolvers.DEFAULT);
+
+        FakeTimeProvider timeProvider = new FakeTimeProvider();
+        long nowMillis = (System.currentTimeMillis() / 1000) * 1000;
+        timeProvider.advance(nowMillis, TimeUnit.MILLISECONDS);
+
+        // Set the oldest segment's mtime to 0 to simulate a concurrent reclamation/rotation whose
+        // timestamp is unavailable by the time the sidecar reads it in CdcRawSegmentFile's constructor.
+        File oldest = new File(cdcDir, TEST_SEGMENT_FILE_NAME_1);
+        File oldestIdx = new File(cdcDir, CdcUtil.getIdxFileName(TEST_SEGMENT_FILE_NAME_1));
+        assumeTrue(oldest.setLastModified(0L) && oldest.lastModified() == 0L,
+                   "Filesystem does not support a 0 mtime; skipping guard assertion");
+
+        CdcRawDirectorySpaceCleaner cleaner = new CdcRawDirectorySpaceCleaner(
+        timeProvider,
+        systemViewsDatabaseAccessor,
+        serviceConfiguration,
+        instancesMetadata,
+        mockSidecarMetrics
+        );
+
+        // Drain any residual accumulation the DeltaGauges may carry from a prior test that
+        // shared the static MetricRegistry — getValue() atomically reads and resets to 0.
+        cdcMetrics.deletedSegment.metric.getValue();
+
+        cleaner.routineCleanUp();
+
+        // Guard must have caused the deletion loop to skip calling deleteSegment() on the phantom.
+        // Pre-fix, (now - 0) collapsed to ~10^12 ms which is neither < critical nor < low, so no
+        // alert fired, but deleteSegment() still ran and Files.deleteIfExists removed both files.
+        assertThat(oldest)
+        .as("phantom segment's log file must not be deleted by the cleaner when its timestamp is unavailable")
+        .exists();
+        assertThat(oldestIdx)
+        .as("phantom segment's idx file must not be deleted by the cleaner when its timestamp is unavailable")
+        .exists();
+
+        // deletedSegment must count only bytes the cleaner actually deleted (seg 2). Pre-fix it
+        // would have included seg 1's bytes as well, roughly doubling the value.
+        assertThat(cdcMetrics.deletedSegment.metric.getValue())
+        .as("deletedSegment metric must count only actually-deleted segments, not the phantom")
+        .isLessThan(2L * 4096L);
+    }
+
     @Test
     void testMaxUsageBytes()
     {