do flush in the LogSegment class

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

@@ -33,7 +33,6 @@
 import java.nio.file.StandardOpenOption;
 import java.util.Objects;
 import java.util.Optional;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
 /**
@@ -51,7 +50,6 @@ public class FileRecords extends AbstractRecords implements Closeable {
     private final AtomicInteger size;
     private final FileChannel channel;
     private volatile File file;
-    private final AtomicBoolean needsFlushParentDir;
 
     /**
      * The {@code FileRecords.open} methods should be used instead of this constructor whenever possible.
@@ -61,15 +59,13 @@ public class FileRecords extends AbstractRecords implements Closeable {
                 FileChannel channel,
                 int start,
                 int end,
-                boolean isSlice,
-                boolean needsFlushParentDir) throws IOException {
+                boolean isSlice) throws IOException {
         this.file = file;
         this.channel = channel;
         this.start = start;
         this.end = end;
         this.isSlice = isSlice;
         this.size = new AtomicInteger();
-        this.needsFlushParentDir = new AtomicBoolean(needsFlushParentDir);
 
         if (isSlice) {
             // don't check the file size if this is just a slice view
@@ -140,7 +136,7 @@ public void readInto(ByteBuffer buffer, int position) throws IOException {
     public FileRecords slice(int position, int size) throws IOException {
         int availableBytes = availableBytes(position, size);
         int startPosition = this.start + position;
-        return new FileRecords(file, channel, startPosition, startPosition + availableBytes, true, false);
+        return new FileRecords(file, channel, startPosition, startPosition + availableBytes, true);
     }
 
     /**
@@ -199,16 +195,12 @@ public int append(MemoryRecords records) throws IOException {
      */
     public void flush() throws IOException {
         channel.force(true);
-        if (needsFlushParentDir.getAndSet(false)) {
-            Utils.flushParentDir(file.toPath());
-        }
     }
 
     /**
      * 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 {
-        needsFlushParentDir.set(false);
         Utils.flushParentDir(file.toPath());
     }
 
@@ -258,9 +250,9 @@ public void updateParentDir(File parentDir) {
      * Rename the file that backs this message set
      * @throws IOException if rename fails.
      */
-    public void renameTo(File f, boolean needsFlushParentDir) throws IOException {
+    public void renameTo(File f) throws IOException {
         try {
-            Utils.atomicMoveWithFallback(file.toPath(), f.toPath(), needsFlushParentDir);
+            Utils.atomicMoveWithFallback(file.toPath(), f.toPath(), false);
         } finally {
             this.file = f;
         }
@@ -439,23 +431,22 @@ public static FileRecords open(File file,
                                    boolean mutable,
                                    boolean fileAlreadyExists,
                                    int initFileSize,
-                                   boolean preallocate,
-                                   boolean needsRecovery) throws IOException {
+                                   boolean preallocate) throws IOException {
         FileChannel channel = openChannel(file, mutable, fileAlreadyExists, initFileSize, preallocate);
         int end = (!fileAlreadyExists && preallocate) ? 0 : Integer.MAX_VALUE;
-        return new FileRecords(file, channel, 0, end, false, mutable && needsRecovery);
+        return new FileRecords(file, channel, 0, end, false);
     }
 
     public static FileRecords open(File file,
                                    boolean fileAlreadyExists,
                                    int initFileSize,
                                    boolean preallocate,
                                    boolean needsRecovery) throws IOException {
-        return open(file, true, fileAlreadyExists, initFileSize, preallocate, needsRecovery);
+        return open(file, true, fileAlreadyExists, initFileSize, preallocate);
     }
 
     public static FileRecords open(File file, boolean mutable) throws IOException {
-        return open(file, mutable, false, 0, false, false);
+        return open(file, mutable, false, 0, false);
     }
 
     public static FileRecords open(File file) throws IOException {

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

@@ -90,7 +90,7 @@ public void testAppendProtectsFromOverflow() throws Exception {
         FileChannel fileChannelMock = mock(FileChannel.class);
         when(fileChannelMock.size()).thenReturn((long) Integer.MAX_VALUE);
 
-        FileRecords records = new FileRecords(fileMock, fileChannelMock, 0, Integer.MAX_VALUE, false, false);
+        FileRecords records = new FileRecords(fileMock, fileChannelMock, 0, Integer.MAX_VALUE, false);
         assertThrows(IllegalArgumentException.class, () -> append(records, values));
     }
 
@@ -100,7 +100,7 @@ public void testOpenOversizeFile() throws Exception {
         FileChannel fileChannelMock = mock(FileChannel.class);
         when(fileChannelMock.size()).thenReturn(Integer.MAX_VALUE + 5L);
 
-        assertThrows(KafkaException.class, () -> new FileRecords(fileMock, fileChannelMock, 0, Integer.MAX_VALUE, false, false));
+        assertThrows(KafkaException.class, () -> new FileRecords(fileMock, fileChannelMock, 0, Integer.MAX_VALUE, false));
     }
 
     @Test
@@ -314,7 +314,7 @@ public void testTruncateNotCalledIfSizeIsSameAsTargetSize() throws IOException {
         when(channelMock.size()).thenReturn(42L);
         when(channelMock.position(42L)).thenReturn(null);
 
-        FileRecords fileRecords = new FileRecords(tempFile(), channelMock, 0, Integer.MAX_VALUE, false, false);
+        FileRecords fileRecords = new FileRecords(tempFile(), channelMock, 0, Integer.MAX_VALUE, false);
         fileRecords.truncateTo(42);
 
         verify(channelMock, atLeastOnce()).size();
@@ -331,7 +331,7 @@ public void testTruncateNotCalledIfSizeIsBiggerThanTargetSize() throws IOExcepti
 
         when(channelMock.size()).thenReturn(42L);
 
-        FileRecords fileRecords = new FileRecords(tempFile(), channelMock, 0, Integer.MAX_VALUE, false, false);
+        FileRecords fileRecords = new FileRecords(tempFile(), channelMock, 0, Integer.MAX_VALUE, false);
 
         try {
             fileRecords.truncateTo(43);
@@ -353,7 +353,7 @@ public void testTruncateIfSizeIsDifferentToTargetSize() throws IOException {
         when(channelMock.size()).thenReturn(42L);
         when(channelMock.truncate(anyLong())).thenReturn(channelMock);
 
-        FileRecords fileRecords = new FileRecords(tempFile(), channelMock, 0, Integer.MAX_VALUE, false, false);
+        FileRecords fileRecords = new FileRecords(tempFile(), channelMock, 0, Integer.MAX_VALUE, false);
         fileRecords.truncateTo(23);
 
         verify(channelMock, atLeastOnce()).size();

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

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

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

@@ -74,9 +74,9 @@ class LazyIndex[T <: AbstractIndex] private (@volatile private var indexWrapper:
     }
   }
 
-  def renameTo(f: File, needFlushParentDir: Boolean): Unit = {
+  def renameTo(f: File): Unit = {
     inLock(lock) {
-      indexWrapper.renameTo(f, needFlushParentDir)
+      indexWrapper.renameTo(f)
     }
   }
 
@@ -114,7 +114,7 @@ object LazyIndex {
 
     def updateParentDir(f: File): Unit
 
-    def renameTo(f: File, needFlushParentDir: Boolean): Unit
+    def renameTo(f: File): Unit
 
     def deleteIfExists(): Boolean
 
@@ -130,8 +130,8 @@ object LazyIndex {
 
     def updateParentDir(parentDir: File): Unit = _file = new File(parentDir, file.getName)
 
-    def renameTo(f: File, needFlushParentDir: Boolean): Unit = {
-      try Utils.atomicMoveWithFallback(file.toPath, f.toPath, needFlushParentDir)
+    def renameTo(f: File): Unit = {
+      try Utils.atomicMoveWithFallback(file.toPath, f.toPath, false)
       catch {
         case _: NoSuchFileException if !file.exists => ()
       }
@@ -152,7 +152,7 @@ object LazyIndex {
 
     def updateParentDir(parentDir: File): Unit = index.updateParentDir(parentDir)
 
-    def renameTo(f: File, needFlushParentDir: Boolean): Unit = index.renameTo(f, needFlushParentDir)
+    def renameTo(f: File): Unit = index.renameTo(f)
 
     def deleteIfExists(): Boolean = index.deleteIfExists()
 

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,13 +498,13 @@ 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, needFlushParentDir: Boolean = true): Unit = {
-    log.renameTo(new File(CoreUtils.replaceSuffix(log.file.getPath, oldSuffix, newSuffix)), false)
-    lazyOffsetIndex.renameTo(new File(CoreUtils.replaceSuffix(lazyOffsetIndex.file.getPath, oldSuffix, newSuffix)), false)
-    lazyTimeIndex.renameTo(new File(CoreUtils.replaceSuffix(lazyTimeIndex.file.getPath, oldSuffix, newSuffix)), false)
-    txnIndex.renameTo(new File(CoreUtils.replaceSuffix(txnIndex.file.getPath, oldSuffix, newSuffix)), false)
+  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 (needFlushParentDir)
+    if (needsFlushParentDir)
       log.flushParentDir()
   }
 
@@ -671,7 +679,8 @@ object LogSegment {
       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

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

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

@@ -3022,7 +3022,7 @@ class LogTest {
     append(10)
     // Kind of a hack, but renaming the index to a directory ensures that the append
     // to the index will fail.
-    log.activeSegment.txnIndex.renameTo(log.dir, false)
+    log.activeSegment.txnIndex.renameTo(log.dir)
     assertThrows(classOf[KafkaStorageException], () => appendEndTxnMarkerAsLeader(log, pid, epoch, ControlRecordType.ABORT, coordinatorEpoch = 1))
     assertThrows(classOf[KafkaStorageException], () => log.appendAsLeader(TestUtils.singletonRecords(value = null), leaderEpoch = 0))
     assertThrows(classOf[KafkaStorageException], () => readLog(log, 0, 4096).records.records.iterator.next().offset)
@@ -4587,7 +4587,7 @@ class LogTest {
 
     // Kind of a hack, but renaming the index to a directory ensures that the append
     // to the index will fail.
-    log.activeSegment.txnIndex.renameTo(log.dir, false)
+    log.activeSegment.txnIndex.renameTo(log.dir)
 
     // The append will be written to the log successfully, but the write to the index will fail
     assertThrows(classOf[KafkaStorageException],

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

@@ -159,7 +159,7 @@ class TransactionIndexTest {
     val renamed = TestUtils.tempFile()
     index.append(new AbortedTxn(producerId = 0L, firstOffset = 0, lastOffset = 10, lastStableOffset = 2))
 
-    index.renameTo(renamed, false)
+    index.renameTo(renamed)
     index.append(new AbortedTxn(producerId = 1L, firstOffset = 5, lastOffset = 15, lastStableOffset = 16))
 
     val abortedTxns = index.collectAbortedTxns(0L, 100L).abortedTransactions

raft/src/main/java/org/apache/kafka/snapshot/FileRawSnapshotReader.java

@@ -73,8 +73,7 @@ public static FileRawSnapshotReader open(Path logDir, OffsetAndEpoch snapshotId)
             false, // mutable
             true, // fileAlreadyExists
             0, // initFileSize
-            false, // preallocate
-            false // needsRecovery
+            false // preallocate
         );
 
         return new FileRawSnapshotReader(fileRecords, snapshotId);