KAFKA-3968: fsync the parent directory of a segment file when the file is created

clients/src/main/java/org/apache/kafka/common/record/FileRecords.java

@@ -197,6 +197,13 @@ public void flush() throws IOException {
         channel.force(true);
     }
 
+    /**
+     * Flush the parent directory of a file to the physical disk, which makes sure the file is accessible after crashing.
+     */
+    public void flushParentDir() throws IOException {
+        Utils.flushParentDir(file.toPath());
+    }
+
     /**
      * Close this record set
      */
@@ -245,7 +252,7 @@ public void updateParentDir(File parentDir) {
      */
     public void renameTo(File f) throws IOException {
         try {
-            Utils.atomicMoveWithFallback(file.toPath(), f.toPath());
+            Utils.atomicMoveWithFallback(file.toPath(), f.toPath(), false);
         } finally {
             this.file = f;
         }
@@ -433,7 +440,8 @@ public static FileRecords open(File file,
     public static FileRecords open(File file,
                                    boolean fileAlreadyExists,
                                    int initFileSize,
-                                   boolean preallocate) throws IOException {
+                                   boolean preallocate,
+                                   boolean needsRecovery) throws IOException {
         return open(file, true, fileAlreadyExists, initFileSize, preallocate);
     }
 

clients/src/main/java/org/apache/kafka/common/utils/Utils.java

@@ -17,6 +17,7 @@
 package org.apache.kafka.common.utils;
 
 import java.nio.BufferUnderflowException;
+import java.nio.file.StandardOpenOption;
 import java.util.AbstractMap;
 import java.util.EnumSet;
 import java.util.SortedSet;
@@ -893,10 +894,23 @@ public static ClassLoader getContextOrKafkaClassLoader() {
 
     /**
      * Attempts to move source to target atomically and falls back to a non-atomic move if it fails.
+     * This function also flushes the parent directory to guarantee crash consistency.
      *
      * @throws IOException if both atomic and non-atomic moves fail
      */
     public static void atomicMoveWithFallback(Path source, Path target) throws IOException {
+        atomicMoveWithFallback(source, target, true);
+    }
+
+    /**
+     * Attempts to move source to target atomically and falls back to a non-atomic move if it fails.
+     * This function allows callers to decide whether to flush the parent directory. This is needed
+     * when a sequence of atomicMoveWithFallback is called for the same directory and we don't want
+     * to repeatedly flush the same parent directory.
+     *
+     * @throws IOException if both atomic and non-atomic moves fail
+     */
+    public static void atomicMoveWithFallback(Path source, Path target, boolean needFlushParentDir) throws IOException {
         try {
             Files.move(source, target, StandardCopyOption.ATOMIC_MOVE);
         } catch (IOException outer) {
@@ -908,6 +922,29 @@ public static void atomicMoveWithFallback(Path source, Path target) throws IOExc
                 inner.addSuppressed(outer);
                 throw inner;
             }
+        } finally {
+            if (needFlushParentDir) {
+                flushParentDir(target);
+            }
+        }
+    }
+
+    /**
+     * Flushes the parent directory to guarantee crash consistency.
+     *
+     * @throws IOException if flushing the parent directory fails.
+     */
+    public static void flushParentDir(Path path) throws IOException {
+        FileChannel dir = null;
+        try {
+            Path parent = path.toAbsolutePath().getParent();
+            if (parent != null) {
+                dir = FileChannel.open(parent, StandardOpenOption.READ);
+                dir.force(true);
+            }
+        } finally {
+            if (dir != null)
+                dir.close();
         }
     }
 

clients/src/test/java/org/apache/kafka/common/record/FileRecordsTest.java

@@ -366,7 +366,7 @@ public void testTruncateIfSizeIsDifferentToTargetSize() throws IOException {
     @Test
     public void testPreallocateTrue() throws IOException {
         File temp = tempFile();
-        FileRecords fileRecords = FileRecords.open(temp, false, 1024 * 1024, true);
+        FileRecords fileRecords = FileRecords.open(temp, false, 1024 * 1024, true, false);
         long position = fileRecords.channel().position();
         int size = fileRecords.sizeInBytes();
         assertEquals(0, position);
@@ -380,7 +380,7 @@ public void testPreallocateTrue() throws IOException {
     @Test
     public void testPreallocateFalse() throws IOException {
         File temp = tempFile();
-        FileRecords set = FileRecords.open(temp, false, 1024 * 1024, false);
+        FileRecords set = FileRecords.open(temp, false, 1024 * 1024, false, false);
         long position = set.channel().position();
         int size = set.sizeInBytes();
         assertEquals(0, position);
@@ -394,7 +394,7 @@ public void testPreallocateFalse() throws IOException {
     @Test
     public void testPreallocateClearShutdown() throws IOException {
         File temp = tempFile();
-        FileRecords fileRecords = FileRecords.open(temp, false, 1024 * 1024, true);
+        FileRecords fileRecords = FileRecords.open(temp, false, 1024 * 1024, true, false);
         append(fileRecords, values);
 
         int oldPosition = (int) fileRecords.channel().position();
@@ -404,7 +404,7 @@ public void testPreallocateClearShutdown() throws IOException {
         fileRecords.close();
 
         File tempReopen = new File(temp.getAbsolutePath());
-        FileRecords setReopen = FileRecords.open(tempReopen, true, 1024 * 1024, true);
+        FileRecords setReopen = FileRecords.open(tempReopen, true, 1024 * 1024, true, false);
         int position = (int) setReopen.channel().position();
         int size = setReopen.sizeInBytes();
 
@@ -439,7 +439,7 @@ public void testSearchForTimestamp() throws IOException {
 
     private void testSearchForTimestamp(RecordVersion version) throws IOException {
         File temp = tempFile();
-        FileRecords fileRecords = FileRecords.open(temp, false, 1024 * 1024, true);
+        FileRecords fileRecords = FileRecords.open(temp, false, 1024 * 1024, true, false);
         appendWithOffsetAndTimestamp(fileRecords, version, 10L, 5, 0);
         appendWithOffsetAndTimestamp(fileRecords, version, 11L, 6, 1);
 

clients/src/test/java/org/apache/kafka/common/record/LazyDownConversionRecordsTest.java

@@ -153,7 +153,7 @@ private static MemoryRecords convertRecords(MemoryRecords recordsToConvert, byte
             try (TransferableChannel channel = toTransferableChannel(FileChannel.open(outputFile.toPath(), StandardOpenOption.READ, StandardOpenOption.WRITE))) {
                 int written = 0;
                 while (written < bytesToConvert) written += lazySend.writeTo(channel, written, bytesToConvert - written);
-                try (FileRecords convertedRecords = FileRecords.open(outputFile, true, written, false)) {
+                try (FileRecords convertedRecords = FileRecords.open(outputFile, true, written, false, false)) {
                     convertedRecordsBuffer = ByteBuffer.allocate(convertedRecords.sizeInBytes());
                     convertedRecords.readInto(convertedRecordsBuffer, 0);
                 }

core/src/main/scala/kafka/log/AbstractIndex.scala

@@ -207,7 +207,7 @@ abstract class AbstractIndex(@volatile private var _file: File, val baseOffset:
    * @throws IOException if rename fails
    */
   def renameTo(f: File): Unit = {
-    try Utils.atomicMoveWithFallback(file.toPath, f.toPath)
+    try Utils.atomicMoveWithFallback(file.toPath, f.toPath, false)
     finally _file = f
   }
 

core/src/main/scala/kafka/log/LazyIndex.scala

@@ -131,7 +131,7 @@ object LazyIndex {
     def updateParentDir(parentDir: File): Unit = _file = new File(parentDir, file.getName)
 
     def renameTo(f: File): Unit = {
-      try Utils.atomicMoveWithFallback(file.toPath, f.toPath)
+      try Utils.atomicMoveWithFallback(file.toPath, f.toPath, false)
       catch {
         case _: NoSuchFileException if !file.exists => ()
       }

core/src/main/scala/kafka/log/Log.scala

@@ -708,7 +708,8 @@ class Log(@volatile private var _dir: File,
           baseOffset = baseOffset,
           config,
           time = time,
-          fileAlreadyExists = true)
+          fileAlreadyExists = true,
+          needsRecovery = !hadCleanShutdown)
 
         try segment.sanityCheck(timeIndexFileNewlyCreated)
         catch {
@@ -2325,7 +2326,7 @@ class Log(@volatile private var _dir: File,
    * @throws IOException if the file can't be renamed and still exists
    */
   private def deleteSegmentFiles(segments: Iterable[LogSegment], asyncDelete: Boolean, deleteProducerStateSnapshots: Boolean = true): Unit = {
-    segments.foreach(_.changeFileSuffixes("", Log.DeletedFileSuffix))
+    segments.foreach(_.changeFileSuffixes("", Log.DeletedFileSuffix, false))
 
     def deleteSegments(): Unit = {
       info(s"Deleting segment files ${segments.mkString(",")}")
@@ -2388,7 +2389,7 @@ class Log(@volatile private var _dir: File,
       // need to do this in two phases to be crash safe AND do the delete asynchronously
       // if we crash in the middle of this we complete the swap in loadSegments()
       if (!isRecoveredSwapFile)
-        sortedNewSegments.reverse.foreach(_.changeFileSuffixes(Log.CleanedFileSuffix, Log.SwapFileSuffix))
+        sortedNewSegments.reverse.foreach(_.changeFileSuffixes(Log.CleanedFileSuffix, Log.SwapFileSuffix, false))
       sortedNewSegments.reverse.foreach(addSegment(_))
       val newSegmentBaseOffsets = sortedNewSegments.map(_.baseOffset).toSet
 

core/src/main/scala/kafka/log/LogManager.scala

@@ -28,7 +28,7 @@ import kafka.server.metadata.ConfigRepository
 import kafka.server._
 import kafka.utils._
 import org.apache.kafka.common.{KafkaException, TopicPartition, Uuid}
-import org.apache.kafka.common.utils.Time
+import org.apache.kafka.common.utils.{Time, Utils}
 import org.apache.kafka.common.errors.{InconsistentTopicIdException, KafkaStorageException, LogDirNotFoundException}
 
 import scala.jdk.CollectionConverters._
@@ -150,6 +150,7 @@ class LogManager(logDirs: Seq[File],
           val created = dir.mkdirs()
           if (!created)
             throw new IOException(s"Failed to create data directory ${dir.getAbsolutePath}")
+          Utils.flushParentDir(dir.toPath)
         }
         if (!dir.isDirectory || !dir.canRead)
           throw new IOException(s"${dir.getAbsolutePath} is not a readable log directory.")
@@ -866,6 +867,7 @@ class LogManager(logDirs: Seq[File],
       val dir = new File(logDirPath, logDirName)
       try {
         Files.createDirectories(dir.toPath)
+        Utils.flushParentDir(dir.toPath)
         Success(dir)
       } catch {
         case e: IOException =>

core/src/main/scala/kafka/log/LogSegment.scala

@@ -20,6 +20,7 @@ import java.io.{File, IOException}
 import java.nio.file.{Files, NoSuchFileException}
 import java.nio.file.attribute.FileTime
 import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicBoolean
 import kafka.common.LogSegmentOffsetOverflowException
 import kafka.metrics.{KafkaMetricsGroup, KafkaTimer}
 import kafka.server.epoch.LeaderEpochFileCache
@@ -59,7 +60,8 @@ class LogSegment private[log] (val log: FileRecords,
                                val baseOffset: Long,
                                val indexIntervalBytes: Int,
                                val rollJitterMs: Long,
-                               val time: Time) extends Logging {
+                               val time: Time,
+                               val needsFlushParentDir: Boolean = false) extends Logging {
 
   def offsetIndex: OffsetIndex = lazyOffsetIndex.get
 
@@ -95,6 +97,9 @@ class LogSegment private[log] (val log: FileRecords,
   /* the number of bytes since we last added an entry in the offset index */
   private var bytesSinceLastIndexEntry = 0
 
+  /* whether or not we need to flush the parent dir during flush */
+  private val atomicNeedsFlushParentDir = new AtomicBoolean(needsFlushParentDir)
+
   // The timestamp we used for time based log rolling and for ensuring max compaction delay
   // volatile for LogCleaner to see the update
   @volatile private var rollingBasedTimestamp: Option[Long] = None
@@ -472,6 +477,9 @@ class LogSegment private[log] (val log: FileRecords,
       offsetIndex.flush()
       timeIndex.flush()
       txnIndex.flush()
+      // We only need to flush the parent of the log file because all other files share the same parent
+      if (atomicNeedsFlushParentDir.getAndSet(false))
+        log.flushParentDir()
     }
   }
 
@@ -490,11 +498,14 @@ class LogSegment private[log] (val log: FileRecords,
    * Change the suffix for the index and log files for this log segment
    * IOException from this method should be handled by the caller
    */
-  def changeFileSuffixes(oldSuffix: String, newSuffix: String): Unit = {
+  def changeFileSuffixes(oldSuffix: String, newSuffix: String, needsFlushParentDir: Boolean = true): Unit = {
     log.renameTo(new File(CoreUtils.replaceSuffix(log.file.getPath, oldSuffix, newSuffix)))
     lazyOffsetIndex.renameTo(new File(CoreUtils.replaceSuffix(lazyOffsetIndex.file.getPath, oldSuffix, newSuffix)))
     lazyTimeIndex.renameTo(new File(CoreUtils.replaceSuffix(lazyTimeIndex.file.getPath, oldSuffix, newSuffix)))
     txnIndex.renameTo(new File(CoreUtils.replaceSuffix(txnIndex.file.getPath, oldSuffix, newSuffix)))
+    // We only need to flush the parent of the log file because all other files share the same parent
+    if (needsFlushParentDir)
+      log.flushParentDir()
   }
 
   /**
@@ -657,17 +668,19 @@ class LogSegment private[log] (val log: FileRecords,
 object LogSegment {
 
   def open(dir: File, baseOffset: Long, config: LogConfig, time: Time, fileAlreadyExists: Boolean = false,
-           initFileSize: Int = 0, preallocate: Boolean = false, fileSuffix: String = ""): LogSegment = {
+           initFileSize: Int = 0, preallocate: Boolean = false, fileSuffix: String = "",
+           needsRecovery: Boolean = true): LogSegment = {
     val maxIndexSize = config.maxIndexSize
     new LogSegment(
-      FileRecords.open(Log.logFile(dir, baseOffset, fileSuffix), fileAlreadyExists, initFileSize, preallocate),
+      FileRecords.open(Log.logFile(dir, baseOffset, fileSuffix), fileAlreadyExists, initFileSize, preallocate, needsRecovery),
       LazyIndex.forOffset(Log.offsetIndexFile(dir, baseOffset, fileSuffix), baseOffset = baseOffset, maxIndexSize = maxIndexSize),
       LazyIndex.forTime(Log.timeIndexFile(dir, baseOffset, fileSuffix), baseOffset = baseOffset, maxIndexSize = maxIndexSize),
       new TransactionIndex(baseOffset, Log.transactionIndexFile(dir, baseOffset, fileSuffix)),
       baseOffset,
       indexIntervalBytes = config.indexInterval,
       rollJitterMs = config.randomSegmentJitter,
-      time)
+      time,
+      needsRecovery || !fileAlreadyExists)
   }
 
   def deleteIfExists(dir: File, baseOffset: Long, fileSuffix: String = ""): Unit = {

core/src/main/scala/kafka/log/TransactionIndex.scala

@@ -109,7 +109,7 @@ class TransactionIndex(val startOffset: Long, @volatile private var _file: File)
   def renameTo(f: File): Unit = {
     try {
       if (file.exists)
-        Utils.atomicMoveWithFallback(file.toPath, f.toPath)
+        Utils.atomicMoveWithFallback(file.toPath, f.toPath, false)
     } finally _file = f
   }
 

core/src/main/scala/kafka/raft/KafkaMetadataLog.scala

@@ -300,7 +300,7 @@ final class KafkaMetadataLog private (
       val path = Snapshots.snapshotPath(log.dir.toPath, snapshotId)
       val destination = Snapshots.deleteRename(path, snapshotId)
       try {
-        Utils.atomicMoveWithFallback(path, destination)
+        Utils.atomicMoveWithFallback(path, destination, false)
       } catch {
         case e: IOException =>
           error(s"Error renaming snapshot file: $path to $destination", e)

core/src/test/scala/unit/kafka/log/LogCleanerTest.scala

@@ -1370,7 +1370,7 @@ class LogCleanerTest {
     //    On recovery, clean operation is aborted. All messages should be present in the log
     log.logSegments.head.changeFileSuffixes("", Log.CleanedFileSuffix)
     for (file <- dir.listFiles if file.getName.endsWith(Log.DeletedFileSuffix)) {
-      Utils.atomicMoveWithFallback(file.toPath, Paths.get(CoreUtils.replaceSuffix(file.getPath, Log.DeletedFileSuffix, "")))
+      Utils.atomicMoveWithFallback(file.toPath, Paths.get(CoreUtils.replaceSuffix(file.getPath, Log.DeletedFileSuffix, "")), false)
     }
     log = recoverAndCheck(config, allKeys)
 
@@ -1386,7 +1386,7 @@ class LogCleanerTest {
     //    renamed to .deleted. Clean operation is resumed during recovery.
     log.logSegments.head.changeFileSuffixes("", Log.SwapFileSuffix)
     for (file <- dir.listFiles if file.getName.endsWith(Log.DeletedFileSuffix)) {
-      Utils.atomicMoveWithFallback(file.toPath, Paths.get(CoreUtils.replaceSuffix(file.getPath, Log.DeletedFileSuffix, "")))
+      Utils.atomicMoveWithFallback(file.toPath, Paths.get(CoreUtils.replaceSuffix(file.getPath, Log.DeletedFileSuffix, "")), false)
     }
     log = recoverAndCheck(config, cleanedKeys)
 

core/src/test/scala/unit/kafka/log/LogTest.scala

@@ -3431,7 +3431,7 @@ class LogTest {
       segment.truncateTo(0)
     })
     for (file <- logDir.listFiles if file.getName.endsWith(Log.DeletedFileSuffix))
-      Utils.atomicMoveWithFallback(file.toPath, Paths.get(CoreUtils.replaceSuffix(file.getPath, Log.DeletedFileSuffix, "")))
+      Utils.atomicMoveWithFallback(file.toPath, Paths.get(CoreUtils.replaceSuffix(file.getPath, Log.DeletedFileSuffix, "")), false)
 
     val recoveredLog = recoverAndCheck(logConfig, expectedKeys)
     assertEquals(expectedKeys, LogTest.keysInLog(recoveredLog))
@@ -3459,7 +3459,7 @@ class LogTest {
       segment.truncateTo(0)
     }
     for (file <- logDir.listFiles if file.getName.endsWith(Log.DeletedFileSuffix))
-      Utils.atomicMoveWithFallback(file.toPath, Paths.get(CoreUtils.replaceSuffix(file.getPath, Log.DeletedFileSuffix, "")))
+      Utils.atomicMoveWithFallback(file.toPath, Paths.get(CoreUtils.replaceSuffix(file.getPath, Log.DeletedFileSuffix, "")), false)
 
     val recoveredLog = recoverAndCheck(logConfig, expectedKeys)
     assertEquals(expectedKeys, LogTest.keysInLog(recoveredLog))
@@ -3483,7 +3483,7 @@ class LogTest {
         segment.changeFileSuffixes("", Log.SwapFileSuffix)
     })
     for (file <- logDir.listFiles if file.getName.endsWith(Log.DeletedFileSuffix))
-      Utils.atomicMoveWithFallback(file.toPath, Paths.get(CoreUtils.replaceSuffix(file.getPath, Log.DeletedFileSuffix, "")))
+      Utils.atomicMoveWithFallback(file.toPath, Paths.get(CoreUtils.replaceSuffix(file.getPath, Log.DeletedFileSuffix, "")), false)
 
     // Truncate the old segment
     segmentWithOverflow.truncateTo(0)

raft/src/test/java/org/apache/kafka/snapshot/SnapshotsTest.java

@@ -116,7 +116,7 @@ public void testDeleteSnapshot(boolean renameBeforeDeleting) throws IOException
 
             if (renameBeforeDeleting)
                 // rename snapshot before deleting
-                Utils.atomicMoveWithFallback(snapshotPath, Snapshots.deleteRename(snapshotPath, snapshotId));
+                Utils.atomicMoveWithFallback(snapshotPath, Snapshots.deleteRename(snapshotPath, snapshotId), false);
 
             assertTrue(Snapshots.deleteSnapshotIfExists(logDirPath, snapshot.snapshotId()));
             assertFalse(Files.exists(snapshotPath));