This is an automated email from the ASF dual-hosted git repository.
jsancio pushed a commit to branch 3.6
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.6 by this push:
new 522263d1956 KAFKA-14273; Close file before atomic move (#14354)
522263d1956 is described below
commit 522263d1956231506e1afb660738484c4ce41667
Author: José Armando García Sancio <jsan...@users.noreply.github.com>
AuthorDate: Thu Sep 7 16:17:03 2023 -0700
KAFKA-14273; Close file before atomic move (#14354)
In the Windows OS atomic move are not allowed if the file has another open
handle. E.g
__cluster_metadata-0\quorum-state: The process cannot access the file
because it is being used by another process
at
java.base/sun.nio.fs.WindowsException.translateToIOException(WindowsException.java:92)
at
java.base/sun.nio.fs.WindowsException.rethrowAsIOException(WindowsException.java:103)
at
java.base/sun.nio.fs.WindowsFileCopy.move(WindowsFileCopy.java:403)
at
java.base/sun.nio.fs.WindowsFileSystemProvider.move(WindowsFileSystemProvider.java:293)
at java.base/java.nio.file.Files.move(Files.java:1430)
at
org.apache.kafka.common.utils.Utils.atomicMoveWithFallback(Utils.java:949)
at
org.apache.kafka.common.utils.Utils.atomicMoveWithFallback(Utils.java:932)
at
org.apache.kafka.raft.FileBasedStateStore.writeElectionStateToFile(FileBasedStateStore.java:152)
This is fixed by first closing the temporary quorum-state file before
attempting to move it.
Reviewers: Colin Patrick McCabe <cmcc...@apache.org>
Co-Authored-By: Renaldo Baur Filho <renald...@gmail.com>
---
.../org/apache/kafka/raft/FileBasedStateStore.java | 31 +++++++++++++---------
.../apache/kafka/raft/FileBasedStateStoreTest.java | 16 +++++++++++
2 files changed, 35 insertions(+), 12 deletions(-)
diff --git a/raft/src/main/java/org/apache/kafka/raft/FileBasedStateStore.java
b/raft/src/main/java/org/apache/kafka/raft/FileBasedStateStore.java
index e403613ccd5..b359493e506 100644
--- a/raft/src/main/java/org/apache/kafka/raft/FileBasedStateStore.java
+++ b/raft/src/main/java/org/apache/kafka/raft/FileBasedStateStore.java
@@ -63,6 +63,7 @@ public class FileBasedStateStore implements QuorumStateStore {
private final File stateFile;
static final String DATA_VERSION = "data_version";
+ static final short HIGHEST_SUPPORTED_VERSION = 0;
public FileBasedStateStore(final File stateFile) {
this.stateFile = stateFile;
@@ -139,21 +140,27 @@ public class FileBasedStateStore implements
QuorumStateStore {
log.trace("Writing tmp quorum state {}", temp.getAbsolutePath());
- try (final FileOutputStream fileOutputStream = new
FileOutputStream(temp);
- final BufferedWriter writer = new BufferedWriter(
- new OutputStreamWriter(fileOutputStream,
StandardCharsets.UTF_8))) {
- short version = state.highestSupportedVersion();
-
- ObjectNode jsonState = (ObjectNode)
QuorumStateDataJsonConverter.write(state, version);
- jsonState.set(DATA_VERSION, new ShortNode(version));
- writer.write(jsonState.toString());
- writer.flush();
- fileOutputStream.getFD().sync();
+ try {
+ try (final FileOutputStream fileOutputStream = new
FileOutputStream(temp);
+ final BufferedWriter writer = new BufferedWriter(
+ new OutputStreamWriter(fileOutputStream,
StandardCharsets.UTF_8)
+ )
+ ) {
+ ObjectNode jsonState = (ObjectNode)
QuorumStateDataJsonConverter.write(state, HIGHEST_SUPPORTED_VERSION);
+ jsonState.set(DATA_VERSION, new
ShortNode(HIGHEST_SUPPORTED_VERSION));
+ writer.write(jsonState.toString());
+ writer.flush();
+ fileOutputStream.getFD().sync();
+ }
Utils.atomicMoveWithFallback(temp.toPath(), stateFile.toPath());
} catch (IOException e) {
throw new UncheckedIOException(
- String.format("Error while writing the Quorum status from the
file %s",
- stateFile.getAbsolutePath()), e);
+ String.format(
+ "Error while writing the Quorum status from the file %s",
+ stateFile.getAbsolutePath()
+ ),
+ e
+ );
} finally {
// cleanup the temp file when the write finishes (either success
or fail).
deleteFileIfExists(temp);
diff --git
a/raft/src/test/java/org/apache/kafka/raft/FileBasedStateStoreTest.java
b/raft/src/test/java/org/apache/kafka/raft/FileBasedStateStoreTest.java
index 5fa4f5c6880..d5a6570ef61 100644
--- a/raft/src/test/java/org/apache/kafka/raft/FileBasedStateStoreTest.java
+++ b/raft/src/test/java/org/apache/kafka/raft/FileBasedStateStoreTest.java
@@ -16,7 +16,9 @@
*/
package org.apache.kafka.raft;
+import org.apache.kafka.common.protocol.types.TaggedFields;
import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.raft.generated.QuorumStateData;
import org.apache.kafka.test.TestUtils;
import org.junit.jupiter.api.AfterEach;
@@ -90,6 +92,20 @@ public class FileBasedStateStoreTest {
assertFalse(stateFile.exists());
}
+ @Test
+ public void testSupportedVersion() {
+ // If the next few checks fail, please check that they are compatible
with previous releases of KRaft
+
+ // Check that FileBasedStateStore supports the latest version
+ assertEquals(FileBasedStateStore.HIGHEST_SUPPORTED_VERSION,
QuorumStateData.HIGHEST_SUPPORTED_VERSION);
+ // Check that the supported versions haven't changed
+ assertEquals(0, QuorumStateData.HIGHEST_SUPPORTED_VERSION);
+ assertEquals(0, QuorumStateData.LOWEST_SUPPORTED_VERSION);
+ // For the latest version check that the number of tagged fields
hasn't changed
+ TaggedFields taggedFields = (TaggedFields)
QuorumStateData.SCHEMA_0.get(6).def.type;
+ assertEquals(0, taggedFields.numFields());
+ }
+
@AfterEach
public void cleanup() throws IOException {
if (stateStore != null) {
