Issue 4855: fail closed on entry log flush failure

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

@@ -489,6 +489,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);
 
@@ -979,6 +982,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);
@@ -1073,6 +1079,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 {
@@ -118,25 +119,31 @@ public void write(ByteBuf src) throws IOException {
         int copied = 0;
         boolean shouldForceWrite = false;
         synchronized (this) {
+            checkWritable();
             int len = src.readableBytes();
-            while (copied < len) {
-                int bytesToCopy = Math.min(src.readableBytes() - copied, writeBuffer.writableBytes());
-                writeBuffer.writeBytes(src, src.readerIndex() + copied, bytesToCopy);
-                copied += bytesToCopy;
+            try {
+                while (copied < len) {
+                    int bytesToCopy = Math.min(src.readableBytes() - copied, writeBuffer.writableBytes());
+                    writeBuffer.writeBytes(src, src.readerIndex() + copied, bytesToCopy);
+                    copied += bytesToCopy;
 
-                // if we have run out of buffer space, we should flush to the
-                // file
-                if (!writeBuffer.isWritable()) {
-                    flush();
+                    // if we have run out of buffer space, we should flush to the
+                    // file
+                    if (!writeBuffer.isWritable()) {
+                        flush();
+                    }
                 }
-            }
-            position += copied;
-            if (doRegularFlushes) {
-                unpersistedBytes.addAndGet(copied);
-                if (unpersistedBytes.get() >= unpersistedBytesBound) {
-                    flush();
-                    shouldForceWrite = true;
+                position += copied;
+                if (doRegularFlushes) {
+                    unpersistedBytes.addAndGet(copied);
+                    if (unpersistedBytes.get() >= unpersistedBytesBound) {
+                        flush();
+                        shouldForceWrite = true;
+                    }
                 }
+            } catch (IOException e) {
+                markWriteFailure(e);
+                throw e;
             }
         }
         if (shouldForceWrite) {
@@ -170,6 +177,7 @@ public long getFileChannelPosition() {
      * @throws IOException
      */
     public void flushAndForceWrite(boolean forceMetadata) throws IOException {
+        checkWritable();
         flush();
         forceWrite(forceMetadata);
     }
@@ -184,6 +192,7 @@ public void flushAndForceWrite(boolean forceMetadata) throws IOException {
      * @throws IOException
      */
     public void flushAndForceWriteIfRegularFlush(boolean forceMetadata) throws IOException {
+        checkWritable();
         if (doRegularFlushes) {
             flushAndForceWrite(forceMetadata);
         }
@@ -196,10 +205,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());
     }
@@ -211,6 +229,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
@@ -237,7 +256,12 @@ public long forceWrite(boolean forceMetadata) throws IOException {
             }
         }
 
-        fileChannel.force(forceMetadata);
+        try {
+            fileChannel.force(forceMetadata);
+        } catch (IOException e) {
+            markWriteFailure(e);
+            throw e;
+        }
         return positionForceWrite;
     }
 
@@ -295,4 +319,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,7 +56,9 @@
 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 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;
@@ -137,6 +139,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();
 
@@ -205,7 +208,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;
+            }
         }
     }
 
@@ -223,6 +242,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>();
 
@@ -375,6 +395,10 @@ EntryLogManager getEntryLogManager() {
         return entryLogManager;
     }
 
+    public void setFatalErrorListener(LedgerDirsListener fatalErrorListener) {
+        entryLogManager.setFatalErrorListener(fatalErrorListener);
+    }
+
     void addListener(EntryLogListener listener) {
         if (null != listener) {
             listeners.add(listener);
@@ -1192,6 +1216,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.