Issue 4855: fail closed on entry log flush failure

bookkeeper-common/src/test/java/org/apache/bookkeeper/common/testing/executors/MockExecutorController.java

@@ -28,6 +28,7 @@
 import java.time.Duration;
 import java.util.Iterator;
 import java.util.List;
+import java.util.concurrent.Callable;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Delayed;
 import java.util.concurrent.ExecutionException;
@@ -120,6 +121,7 @@ void run() {
 
     public MockExecutorController controlSubmit(ScheduledExecutorService service) {
         doAnswer(answerNow()).when(service).submit(any(Runnable.class));
+        doAnswer(answerNowCallable()).when(service).submit(org.mockito.ArgumentMatchers.<Callable<Object>>any());
         return this;
     }
 
@@ -183,7 +185,23 @@ private static Answer<Future<?>> answerNow() {
            SettableFuture<Void> future = SettableFuture.create();
            future.set(null);
            return future;
-       };
+        };
+    }
+
+    private static Answer<Future<?>> answerNowCallable() {
+        return invocationOnMock -> {
+           // Keep Callable submission semantics: task failures complete the Future exceptionally.
+           ThreadRegistry.forceClearRegistrationForTests(Thread.currentThread().getId());
+
+           Callable<?> task = invocationOnMock.getArgument(0);
+           SettableFuture<Object> future = SettableFuture.create();
+           try {
+               future.set(task.call());
+           } catch (Throwable t) {
+               future.setException(t);
+           }
+           return future;
+        };
     }
 
     private DeferredTask addDelayedTask(

bookkeeper-common/src/test/java/org/apache/bookkeeper/common/testing/executors/MockExecutorControllerTest.java

@@ -18,12 +18,18 @@
  */
 package org.apache.bookkeeper.common.testing.executors;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.fail;
 import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
 import java.time.Duration;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import org.junit.Before;
@@ -57,6 +63,41 @@ public void testSubmit() {
         verify(task, times(1)).run();
     }
 
+    @Test
+    public void testSubmitCallable() throws Exception {
+        Callable<String> task = () -> "done";
+
+        Future<String> future = executor.submit(task);
+
+        assertEquals("done", future.get());
+    }
+
+    @Test
+    public void testSubmitCallableWithNullResult() throws Exception {
+        Callable<Object> task = () -> null;
+
+        Future<Object> future = executor.submit(task);
+
+        assertNull(future.get());
+    }
+
+    @Test
+    public void testSubmitCallableFailure() throws Exception {
+        RuntimeException failure = new RuntimeException("failure");
+        Callable<String> task = () -> {
+            throw failure;
+        };
+
+        Future<String> future = executor.submit(task);
+
+        try {
+            future.get();
+            fail("Expected the submitted Callable failure to be visible from Future.get()");
+        } catch (ExecutionException e) {
+            assertEquals(failure, e.getCause());
+        }
+    }
+
     @Test
     public void testExecute() {
         Runnable task = mock(Runnable.class);

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookieImpl.java

@@ -500,6 +500,9 @@ public void ledgerDeleted(long ledgerId) {
         ledgerStorage.setStateManager(stateManager);
         ledgerStorage.setCheckpointSource(checkpointSource);
         ledgerStorage.setCheckpointer(syncThread);
+        if (isDbLedgerStorage) {
+            ((DbLedgerStorage) ledgerStorage).setFatalErrorListener(getLedgerDirsListener());
+        }
         ledgerStorage.registerLedgerDeletionListener(ledgerDeletionListener);
         handles = new HandleFactoryImpl(ledgerStorage);
 
@@ -871,6 +874,9 @@ public void run() {
     // because shutdown can be called from sync thread which would be
     // interrupted by shutdown call.
     AtomicBoolean shutdownTriggered = new AtomicBoolean(false);
+    // Startup flush runs before stateManager.initState(), so isRunning() is false if it fails.
+    // Track shutdown independently to ensure the cleanup path still runs exactly once.
+    private final AtomicBoolean shutdownStarted = new AtomicBoolean(false);
     void triggerBookieShutdown(final int exitCode) {
         if (!shutdownTriggered.compareAndSet(false, true)) {
             return;
@@ -899,7 +905,7 @@ public int shutdown() {
     int shutdown(int exitCode) {
         lock.lock();
         try {
-            if (isRunning()) {
+            if (shutdownStarted.compareAndSet(false, true)) {
                 // the exitCode only set when first shutdown usually due to exception found
                 log.info()
                         .attr("bookiePort", conf.getBookiePort())
@@ -1033,6 +1039,9 @@ public void recoveryAddEntry(ByteBuf entry, WriteCallback cb, Object ctx, byte[]
                 addEntryInternal(handle, entry, false /* ackBeforeSync */, cb, ctx, masterKey);
             }
             success = true;
+        } catch (EntryLogWriteException e) {
+            triggerBookieShutdown(ExitCode.BOOKIE_EXCEPTION);
+            throw e;
         } catch (NoWritableLedgerDirException e) {
             stateManager.transitionToReadOnlyMode();
             throw new IOException(e);
@@ -1125,6 +1134,9 @@ public void addEntry(ByteBuf entry, boolean ackBeforeSync, WriteCallback cb, Obj
                 addEntryInternal(handle, entry, ackBeforeSync, cb, ctx, masterKey);
             }
             success = true;
+        } catch (EntryLogWriteException e) {
+            triggerBookieShutdown(ExitCode.BOOKIE_EXCEPTION);
+            throw e;
         } catch (NoWritableLedgerDirException e) {
             stateManager.transitionToReadOnlyMode();
             throw new IOException(e);

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BufferedChannel.java

@@ -71,6 +71,7 @@ public class BufferedChannel extends BufferedReadChannel implements Closeable {
     protected final AtomicLong unpersistedBytes;
 
     private boolean closed = false;
+    private volatile IOException writeFailure;
 
     // make constructor to be public for unit test
     public BufferedChannel(ByteBufAllocator allocator, FileChannel fc, int capacity) throws IOException {
@@ -117,8 +118,14 @@ public synchronized void close() throws IOException {
     public void write(ByteBuf src) throws IOException {
         boolean shouldForceWrite = false;
         synchronized (this) {
-            int copied = copyIntoWriteBuffer(src);
-            shouldForceWrite = updatePositionAndFlushIfNeeded(copied);
+            checkWritable();
+            try {
+                int copied = copyIntoWriteBuffer(src);
+                shouldForceWrite = updatePositionAndFlushIfNeeded(copied);
+            } catch (IOException e) {
+                markWriteFailure(e);
+                throw e;
+            }
         }
         if (shouldForceWrite) {
             forceWrite(false);
@@ -137,9 +144,15 @@ public void write(ByteBuf src) throws IOException {
     public void write(ByteBuf src1, ByteBuf src2) throws IOException {
         boolean shouldForceWrite = false;
         synchronized (this) {
-            int copied = copyIntoWriteBuffer(src1);
-            copied += copyIntoWriteBuffer(src2);
-            shouldForceWrite = updatePositionAndFlushIfNeeded(copied);
+            checkWritable();
+            try {
+                int copied = copyIntoWriteBuffer(src1);
+                copied += copyIntoWriteBuffer(src2);
+                shouldForceWrite = updatePositionAndFlushIfNeeded(copied);
+            } catch (IOException e) {
+                markWriteFailure(e);
+                throw e;
+            }
         }
         if (shouldForceWrite) {
             forceWrite(false);
@@ -203,6 +216,7 @@ public long getFileChannelPosition() {
      * @throws IOException
      */
     public void flushAndForceWrite(boolean forceMetadata) throws IOException {
+        checkWritable();
         flush();
         forceWrite(forceMetadata);
     }
@@ -217,6 +231,7 @@ public void flushAndForceWrite(boolean forceMetadata) throws IOException {
      * @throws IOException
      */
     public void flushAndForceWriteIfRegularFlush(boolean forceMetadata) throws IOException {
+        checkWritable();
         if (doRegularFlushes) {
             flushAndForceWrite(forceMetadata);
         }
@@ -229,10 +244,19 @@ public void flushAndForceWriteIfRegularFlush(boolean forceMetadata) throws IOExc
      * @throws IOException if the write fails.
      */
     public synchronized void flush() throws IOException {
+        checkWritable();
         ByteBuffer toWrite = writeBuffer.internalNioBuffer(0, writeBuffer.writerIndex());
-        do {
-            fileChannel.write(toWrite);
-        } while (toWrite.hasRemaining());
+        try {
+            while (toWrite.hasRemaining()) {
+                int written = fileChannel.write(toWrite);
+                if (written <= 0) {
+                    throw new IOException("Unable to make progress while flushing buffered channel");
+                }
+            }
+        } catch (IOException e) {
+            markWriteFailure(e);
+            throw e;
+        }
         writeBuffer.clear();
         writeBufferStartPosition.set(fileChannel.position());
     }
@@ -244,6 +268,7 @@ public synchronized void flush() throws IOException {
      * @throws IOException
      */
     public long forceWrite(boolean forceMetadata) throws IOException {
+        checkWritable();
         // This is the point up to which we had flushed to the file system page cache
         // before issuing this force write hence is guaranteed to be made durable by
         // the force write, any flush that happens after this may or may
@@ -270,7 +295,12 @@ public long forceWrite(boolean forceMetadata) throws IOException {
             }
         }
 
-        fileChannel.force(forceMetadata);
+        try {
+            fileChannel.force(forceMetadata);
+        } catch (IOException e) {
+            markWriteFailure(e);
+            throw e;
+        }
         return positionForceWrite;
     }
 
@@ -333,4 +363,17 @@ public synchronized int getNumOfBytesInWriteBuffer() {
     long getUnpersistedBytes() {
         return unpersistedBytes.get();
     }
+
+    final void checkWritable() throws IOException {
+        IOException failure = writeFailure;
+        if (failure != null) {
+            throw new IOException("BufferedChannel is in failed state", failure);
+        }
+    }
+
+    final void markWriteFailure(IOException e) {
+        if (writeFailure == null) {
+            writeFailure = e;
+        }
+    }
 }

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/DefaultEntryLogger.java

@@ -56,8 +56,10 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.regex.Pattern;
 import lombok.CustomLog;
+import org.apache.bookkeeper.bookie.LedgerDirsManager.LedgerDirsListener;
 import org.apache.bookkeeper.bookie.storage.CompactionEntryLog;
 import org.apache.bookkeeper.bookie.storage.EntryLogScanner;
 import org.apache.bookkeeper.bookie.storage.EntryLogger;
@@ -136,6 +138,7 @@ public String toString() {
          * Updates the entry log file header with the offset and size of the map.
          */
         void appendLedgersMap() throws IOException {
+            checkWritable();
 
             long ledgerMapOffset = this.position();
 
@@ -204,7 +207,23 @@ public void accept(long ledgerId, long size) {
             mapInfo.putLong(ledgerMapOffset);
             mapInfo.putInt(numberOfLedgers);
             mapInfo.flip();
-            this.fileChannel.write(mapInfo, LEDGERS_MAP_OFFSET_POSITION);
+            try {
+                writeFully(this.fileChannel, mapInfo, LEDGERS_MAP_OFFSET_POSITION);
+            } catch (IOException e) {
+                markWriteFailure(e);
+                throw e;
+            }
+        }
+
+        private static void writeFully(FileChannel fileChannel, ByteBuffer buffer, long position) throws IOException {
+            long writePosition = position;
+            while (buffer.hasRemaining()) {
+                int written = fileChannel.write(buffer, writePosition);
+                if (written <= 0) {
+                    throw new IOException("Unable to make progress while updating entry log header");
+                }
+                writePosition += written;
+            }
         }
     }
 
@@ -222,6 +241,7 @@ public void accept(long ledgerId, long size) {
 
     final EntryLoggerAllocator entryLoggerAllocator;
     private final EntryLogManager entryLogManager;
+    private final AtomicBoolean closed = new AtomicBoolean(false);
 
     private final CopyOnWriteArrayList<EntryLogListener> listeners = new CopyOnWriteArrayList<EntryLogListener>();
 
@@ -365,6 +385,10 @@ EntryLogManager getEntryLogManager() {
         return entryLogManager;
     }
 
+    public void setFatalErrorListener(LedgerDirsListener fatalErrorListener) {
+        entryLogManager.setFatalErrorListener(fatalErrorListener);
+    }
+
     void addListener(EntryLogListener listener) {
         if (null != listener) {
             listeners.add(listener);
@@ -1208,6 +1232,10 @@ public boolean accept(long ledgerId) {
      */
     @Override
     public void close() {
+        if (!closed.compareAndSet(false, true)) {
+            log.debug("EntryLogger is already stopped");
+            return;
+        }
         // since logChannel is buffered channel, do flush when shutting down
         log.info("Stopping EntryLogger");
         try {

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManager.java

@@ -26,6 +26,7 @@
 import java.io.IOException;
 import java.util.List;
 import org.apache.bookkeeper.bookie.DefaultEntryLogger.BufferedLogChannel;
+import org.apache.bookkeeper.bookie.LedgerDirsManager.LedgerDirsListener;
 
 interface EntryLogManager {
 
@@ -66,6 +67,11 @@ interface EntryLogManager {
      */
     void forceClose();
 
+    /*
+     * notify the owning bookie when entry-log-level writes become fatal.
+     */
+    void setFatalErrorListener(LedgerDirsListener fatalErrorListener);
+
     /*
      * prepare entrylogger/entrylogmanager before doing SortedLedgerStorage
      * Checkpoint.