You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ratis.apache.org by sz...@apache.org on 2026/10/16 02:31:31 UTC
(ratis) branch master updated: RATIS-2647. Trigger shutdown for async log flush failure (#1643)
This is an automated email from the ASF dual-hosted git repository.
szetszwo pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ratis.git
The following commit(s) were added to refs/heads/master by this push:
new 78a1f27d3 RATIS-2647. Trigger shutdown for async log flush failure (#1643)
78a1f27d3 is described below
commit 78a1f27d39910695ee222fad952fccecfe28cb56
Author: Ivan Andika <iv...@apache.org>
AuthorDate: Fri Oct 16 10:31:25 2026 +0800
RATIS-2647. Trigger shutdown for async log flush failure (#1643)
---
.../raftlog/segmented/BufferedWriteChannel.java | 15 ++++++-
.../raftlog/segmented/SegmentedRaftLogWorker.java | 8 +++-
.../raftlog/segmented/TestSegmentedRaftLog.java | 49 ++++++++++++++++++++--
3 files changed, 66 insertions(+), 6 deletions(-)
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/BufferedWriteChannel.java b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/BufferedWriteChannel.java
index 6b8b201f3..e544fba62 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/BufferedWriteChannel.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/BufferedWriteChannel.java
@@ -17,7 +17,9 @@
*/
package org.apache.ratis.server.raftlog.segmented;
+import org.apache.ratis.util.CodeInjectionForTesting;
import org.apache.ratis.util.FileUtils;
+import org.apache.ratis.util.JavaUtils;
import org.apache.ratis.util.Preconditions;
import org.apache.ratis.util.function.CheckedBiFunction;
import org.apache.ratis.util.function.CheckedConsumer;
@@ -42,6 +44,8 @@ import java.util.concurrent.atomic.AtomicReference;
*/
class BufferedWriteChannel implements Closeable {
static final Logger LOG = LoggerFactory.getLogger(BufferedWriteChannel.class);
+ static final String FILE_CHANNEL_FORCE =
+ JavaUtils.getClassSimpleName(BufferedWriteChannel.class) + ".fileChannelForce";
@SuppressWarnings("java:S2095") // return Closable
static BufferedWriteChannel open(File file, boolean append, ByteBuffer buffer) throws IOException {
@@ -136,7 +140,7 @@ class BufferedWriteChannel implements Closeable {
void flush() throws IOException {
flushBuffer();
if (!forced) {
- fileChannel.force(false);
+ forceFileChannel();
forced = true;
}
}
@@ -153,13 +157,20 @@ class BufferedWriteChannel implements Closeable {
private Void fileChannelForce() {
try {
- fileChannel.force(false);
+ forceFileChannel();
} catch (IOException e) {
throw new CompletionException("Failed to force channel " + this, e);
}
return null;
}
+ private void forceFileChannel() throws IOException {
+ if (CodeInjectionForTesting.execute(FILE_CHANNEL_FORCE, this, null)) {
+ throw new IOException("Injected failure");
+ }
+ fileChannel.force(false);
+ }
+
/** Flush the data from the {@link #writeBuffer} to {@link #fileChannel}. */
private void flushBuffer() throws IOException {
if (writeBufferPosition() == 0) {
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogWorker.java b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogWorker.java
index 5c8c81f9b..1038a84c2 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogWorker.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogWorker.java
@@ -417,12 +417,18 @@ class SegmentedRaftLogWorker {
}
private void asyncFlushOutStream(CompletableFuture<Void> stateMachineFlush) throws IOException {
+ final long scheduledFlushIndex = lastWrittenIndex;
final Timekeeper.Context logSyncTimerContext = raftLogMetrics.getSyncTimer().time();
out.asyncFlush(flushExecutor)
.thenCombine(stateMachineFlush, (async, sm) -> async)
.whenComplete((v, e) -> {
- updateFlushedIndexIncreasingly(lastWrittenIndex);
logSyncTimerContext.stop();
+ if (e == null) {
+ updateFlushedIndexIncreasingly(scheduledFlushIndex);
+ } else {
+ LOG.error("{} Failed asyncFlush", this, e);
+ server.close();
+ }
});
}
diff --git a/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestSegmentedRaftLog.java b/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestSegmentedRaftLog.java
index 445fe341e..7c613272f 100644
--- a/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestSegmentedRaftLog.java
+++ b/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestSegmentedRaftLog.java
@@ -26,6 +26,7 @@ import org.apache.ratis.protocol.RaftGroupId;
import org.apache.ratis.protocol.RaftGroupMemberId;
import org.apache.ratis.protocol.RaftPeerId;
import org.apache.ratis.protocol.exceptions.TimeoutIOException;
+import org.apache.ratis.server.RaftServer;
import org.apache.ratis.server.RaftServerConfigKeys;
import org.apache.ratis.server.impl.RetryCacheTestUtil;
import org.apache.ratis.server.RetryCache;
@@ -82,8 +83,12 @@ import org.slf4j.event.Level;
import static java.lang.Boolean.FALSE;
import static java.lang.Boolean.TRUE;
+import static org.apache.ratis.server.raftlog.segmented.BufferedWriteChannel.FILE_CHANNEL_FORCE;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.apache.ratis.server.raftlog.segmented.SegmentedRaftLogWorker.RUN_WORKER;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.params.provider.Arguments.arguments;
@@ -832,10 +837,10 @@ public class TestSegmentedRaftLog extends BaseTest {
TimeUnit.SECONDS.sleep(1);
assertIndices(raftLog, flush, next);
sm.unblockFlushStateMachineData();
- assertIndicesMultipleAttempts(raftLog, flush + 2, next);
+ // An async flush only advances through the index captured when it was scheduled.
+ final long expectedFlushIndex = useAsyncFlush && !smSyncFlush ? flush + 1 : flush + 2;
+ assertIndicesMultipleAttempts(raftLog, expectedFlushIndex, next);
- // raftLog.appendEntry(entry).get() won't return
- // until sm.unblockFlushStateMachineData() was called.
t.join();
}
}
@@ -890,6 +895,44 @@ public class TestSegmentedRaftLog extends BaseTest {
Assertions.assertInstanceOf(TimeoutIOException.class, ex.getCause());
}
+ @Test
+ public void testAsyncFlushFailure() throws Exception {
+ RaftServerConfigKeys.Log.setAsyncFlushEnabled(properties, true);
+ RaftServerConfigKeys.Log.setForceSyncNum(properties, 1);
+ final AtomicInteger commitUpdates = new AtomicInteger();
+ final RaftServer.Division server = mock(RaftServer.Division.class);
+
+ try (SegmentedRaftLog raftLog = SegmentedRaftLog.newBuilder()
+ .setMemberId(MEMBER_ID)
+ .setServer(server)
+ .setSubmitUpdateCommitEvent(commitUpdates::incrementAndGet)
+ .setStorage(storage)
+ .setProperties(properties)
+ .build()) {
+ raftLog.open(RaftLog.INVALID_LOG_INDEX, null);
+ raftLog.appendEntry(prepareLogEntry(0, 0, null, false));
+ JavaUtils.attemptRepeatedly(() -> {
+ assertEquals(0, raftLog.getFlushIndex());
+ return null;
+ }, 10, HUNDRED_MILLIS, "initial async flush", LOG);
+ commitUpdates.set(0);
+
+ CodeInjectionForTesting.put(FILE_CHANNEL_FORCE, (localId, remoteId, args) -> true);
+ try {
+ raftLog.appendEntry(prepareLogEntry(0, 1, null, false));
+
+ JavaUtils.attemptRepeatedly(() -> {
+ verify(server).close();
+ assertEquals(0, raftLog.getFlushIndex());
+ assertEquals(0, commitUpdates.get());
+ return null;
+ }, 10, HUNDRED_MILLIS, "async flush failure", LOG);
+ } finally {
+ CodeInjectionForTesting.remove(FILE_CHANNEL_FORCE);
+ }
+ }
+ }
+
static Thread startAppendEntryThread(RaftLog raftLog, LogEntryProto entry) {
final Thread t = new Thread(() -> {
try {