You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2026/09/28 09:03:03 UTC
(kafka) branch trunk updated: KAFKA-21075 Consolidate flushDir/flushDirIfExists/flushFileIfExists into flushPath/flushPathIfExists (#23441)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 18bc66cac90 KAFKA-21075 Consolidate flushDir/flushDirIfExists/flushFileIfExists into flushPath/flushPathIfExists (#23441)
18bc66cac90 is described below
commit 18bc66cac9081b3916fa3dd76f69355835421c24
Author: Murali Basani <mu...@aiven.io>
AuthorDate: Mon Sep 28 11:02:55 2026 +0200
KAFKA-21075 Consolidate flushDir/flushDirIfExists/flushFileIfExists into flushPath/flushPathIfExists (#23441)
Ref: https://issues.apache.org/jira/browse/KAFKA-21075 Raised in review
of #23316 (#23316 (comment)).
`flushFileIfExists` did not skip fsync on Windows and z/OS while
`flushDir` did. Rather than adding the same guard twice, this PR merges
the three flush helpers into two:
- `flushPath`: fsyncs a file or directory. Handles the null check and
the Windows/z/OS skip.
- `flushPathIfExists`: delegates to `flushPath` and swallows
`NoSuchFileException` at DEBUG level, since callers expect the path may
already be gone.
`flushDir`, `flushDirIfExists` and `flushFileIfExists` are removed and
all callers are updated.
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
.../java/org/apache/kafka/common/utils/Utils.java | 39 +++++++++-------------
.../org/apache/kafka/common/utils/UtilsTest.java | 36 ++++++++++++++++++++
.../kafka/storage/internals/log/AbstractIndex.java | 2 +-
.../kafka/storage/internals/log/LocalLog.java | 4 +--
.../kafka/storage/internals/log/LogManager.java | 2 +-
.../kafka/storage/internals/log/UnifiedLog.java | 2 +-
6 files changed, 57 insertions(+), 28 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java b/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
index 5fa1b1b83b5..f02f8141a33 100644
--- a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
+++ b/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
@@ -952,47 +952,40 @@ public final class Utils {
}
} finally {
if (needFlushParentDir) {
- flushDir(target.toAbsolutePath().normalize().getParent());
+ flushPath(target.toAbsolutePath().normalize().getParent());
}
}
}
/**
- * Flushes dirty directories to guarantee crash consistency.
+ * Flushes the dirty file or directory at the given path to guarantee crash consistency.
* <p>
- * Note: We don't fsync directories on Windows OS because otherwise it'll throw AccessDeniedException (KAFKA-13391)
+ * This is a no-op if the path is {@code null}.
+ * <p>
+ * Note: We don't fsync on Windows OS and z/OS because otherwise it'll throw AccessDeniedException (KAFKA-13391)
*
- * @throws IOException if flushing the directory fails.
+ * @throws NoSuchFileException if the path does not exist.
+ * @throws IOException if flushing the path fails.
*/
- public static void flushDir(Path path) throws IOException {
+ public static void flushPath(Path path) throws IOException {
if (path != null && !OperatingSystem.IS_WINDOWS && !OperatingSystem.IS_ZOS) {
- try (FileChannel dir = FileChannel.open(path, StandardOpenOption.READ)) {
- dir.force(true);
+ try (FileChannel channel = FileChannel.open(path, StandardOpenOption.READ)) {
+ channel.force(true);
}
}
}
/**
- * Flushes dirty directories to guarantee crash consistency with swallowing {@link NoSuchFileException}
+ * Flushes the dirty file or directory at the given path to guarantee crash consistency,
+ * swallowing {@link NoSuchFileException} if the path does not exist.
*
- * @throws IOException if flushing the directory fails.
+ * @throws IOException if flushing the path fails.
*/
- public static void flushDirIfExists(Path path) throws IOException {
+ public static void flushPathIfExists(Path path) throws IOException {
try {
- flushDir(path);
- } catch (NoSuchFileException e) {
- log.warn("Failed to flush directory {}", path);
- }
- }
-
- /**
- * Flushes dirty file with swallowing {@link NoSuchFileException}
- */
- public static void flushFileIfExists(Path path) throws IOException {
- try (FileChannel fileChannel = FileChannel.open(path, StandardOpenOption.READ)) {
- fileChannel.force(true);
+ flushPath(path);
} catch (NoSuchFileException e) {
- log.warn("Failed to flush file {}", path, e);
+ log.debug("Skipped flushing {} since it no longer exists", path);
}
}
diff --git a/clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java b/clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java
index 70d9c220058..5ec218100f9 100755
--- a/clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java
@@ -17,6 +17,7 @@
package org.apache.kafka.common.utils;
import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.utils.internals.OperatingSystem;
import org.apache.kafka.common.utils.internals.SingleByteBufferOutputStream;
import org.apache.kafka.test.TestUtils;
@@ -39,6 +40,7 @@ import java.nio.charset.StandardCharsets;
import java.nio.file.DirectoryStream;
import java.nio.file.FileSystem;
import java.nio.file.Files;
+import java.nio.file.NoSuchFileException;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.nio.file.attribute.BasicFileAttributes;
@@ -93,6 +95,7 @@ import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
+import static org.junit.jupiter.api.Assumptions.assumeFalse;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.atLeastOnce;
@@ -608,6 +611,39 @@ public class UtilsTest {
}
}
+ @Test
+ public void testFlushPathWithExistingFile() throws IOException {
+ Path file = TestUtils.tempFile().toPath();
+ // On non-Windows/z/OS platforms this fsyncs the file; on Windows/z/OS it is skipped. Either way it must not throw.
+ assertDoesNotThrow(() -> Utils.flushPath(file));
+ }
+
+ @Test
+ public void testFlushPathWithExistingDirectory() {
+ Path dir = TestUtils.tempDirectory().toPath();
+ assertDoesNotThrow(() -> Utils.flushPath(dir));
+ }
+
+ @Test
+ public void testFlushPathWithNullIsNoOp() {
+ assertDoesNotThrow(() -> Utils.flushPath(null));
+ }
+
+ @Test
+ public void testFlushPathThrowsForMissingPath() {
+ // The platform guard short-circuits before the path is opened on Windows/z/OS, so nothing is thrown there.
+ assumeFalse(OperatingSystem.IS_WINDOWS || OperatingSystem.IS_ZOS);
+ Path missing = TestUtils.tempDirectory().toPath().resolve("does-not-exist");
+ assertThrows(NoSuchFileException.class, () -> Utils.flushPath(missing));
+ }
+
+ @Test
+ public void testFlushPathIfExistsSwallowsNoSuchFileException() {
+ Path missing = TestUtils.tempDirectory().toPath().resolve("does-not-exist");
+ // A missing path must be swallowed (NoSuchFileException), not propagated.
+ assertDoesNotThrow(() -> Utils.flushPathIfExists(missing));
+ }
+
/**
* Tests that `readFullyOrFail` behaves correctly if multiple `FileChannel.read` operations are required to fill
* the destination buffer.
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/AbstractIndex.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/AbstractIndex.java
index cb32b8ba6b3..5f5187a23ff 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/log/AbstractIndex.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/AbstractIndex.java
@@ -246,7 +246,7 @@ public abstract class AbstractIndex implements Closeable {
inLock(() -> {
if (mmap != null) {
mmap.force();
- Utils.flushFileIfExists(file.toPath());
+ Utils.flushPathIfExists(file.toPath());
}
});
}
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LocalLog.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LocalLog.java
index 1de72e12b12..c230c4506ca 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/log/LocalLog.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LocalLog.java
@@ -270,7 +270,7 @@ public class LocalLog {
if (segmentsToFlush.stream().anyMatch(s -> s.baseOffset() >= currentRecoveryPoint)) {
// The directory might be renamed concurrently for topic deletion, which may cause NoSuchFileException here.
// Since the directory is to be deleted anyways, we just swallow NoSuchFileException and let it go.
- Utils.flushDirIfExists(dir.toPath());
+ Utils.flushPathIfExists(dir.toPath());
}
}
}
@@ -1059,7 +1059,7 @@ public class LocalLog {
for (LogSegment logSegment : sortedNewSegments) {
logSegment.changeFileSuffixes(SWAP_FILE_SUFFIX, "");
}
- Utils.flushDir(dir.toPath());
+ Utils.flushPath(dir.toPath());
return deletedNotReplaced;
}
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogManager.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogManager.java
index 095e1e86560..562bb6009a0 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogManager.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogManager.java
@@ -355,7 +355,7 @@ public class LogManager {
if (!created) {
throw new IOException("Failed to create data directory " + dir.getAbsolutePath());
}
- Utils.flushDir(dir.toPath().toAbsolutePath().normalize().getParent());
+ Utils.flushPath(dir.toPath().toAbsolutePath().normalize().getParent());
}
if (!dir.isDirectory() || !dir.canRead()) {
throw new IOException(dir.getAbsolutePath() + " is not a readable log directory.");
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java
index dca6f478e9d..bdf03e6e88d 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java
@@ -2342,7 +2342,7 @@ public class UnifiedLog implements AutoCloseable {
maybeHandleIOException(
() -> "Error while flushing producer state snapshot " + snapshot + " for " + topicPartition() + " in dir " + dir().getParent(),
() -> {
- Utils.flushFileIfExists(snapshot);
+ Utils.flushPathIfExists(snapshot);
return null;
});
}