From 12b1a948027a2b38a8915b8640ca23bd5f567236 Mon Sep 17 00:00:00 2001 From: root Date: Tue, 4 Aug 2026 10:29:02 +0800 Subject: [PATCH 1/9] fix: fail closed after buffered entrylog write failure --- .../bookkeeper/bookie/BufferedChannel.java | 61 ++- .../bookkeeper/bookie/DefaultEntryLogger.java | 19 +- .../bookie/BufferedChannelTest.java | 358 ++++++++++++++++++ 3 files changed, 428 insertions(+), 10 deletions(-) diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BufferedChannel.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BufferedChannel.java index cddeff9cf29..1fb706742dd 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BufferedChannel.java +++ b/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; + } + } } diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/DefaultEntryLogger.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/DefaultEntryLogger.java index 3bfc0d8d377..a9cfb3bfee3 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/DefaultEntryLogger.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/DefaultEntryLogger.java @@ -136,6 +136,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 +205,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; + } } } diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BufferedChannelTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BufferedChannelTest.java index 0114dbc1253..30e5db0b7c8 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BufferedChannelTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BufferedChannelTest.java @@ -30,7 +30,12 @@ import java.io.File; import java.io.IOException; import java.io.RandomAccessFile; +import java.nio.ByteBuffer; +import java.nio.MappedByteBuffer; import java.nio.channels.FileChannel; +import java.nio.channels.FileLock; +import java.nio.channels.ReadableByteChannel; +import java.nio.channels.WritableByteChannel; import java.nio.file.Files; import java.util.Random; import org.junit.Assert; @@ -75,6 +80,191 @@ public void testBufferedChannelFlushForceWrite() throws Exception { testBufferedChannel(5000, 30, 0, true, true); } + @Test + public void testPartialFlushFailurePoisonsBufferedChannel() throws Exception { + File newLogFile = File.createTempFile("test", "log"); + newLogFile.deleteOnExit(); + FileChannel delegate = new RandomAccessFile(newLogFile, "rw").getChannel(); + PartialFailingFileChannel fileChannel = new PartialFailingFileChannel(delegate, 8); + + BufferedChannel logChannel = new BufferedChannel(UnpooledByteBufAllocator.DEFAULT, fileChannel, + 16, INTERNAL_BUFFER_READ_CAPACITY, 0); + + try { + logChannel.write(Unpooled.wrappedBuffer(new byte[16])); + Assert.fail("Expected the internal flush to fail"); + } catch (IOException expected) { + // Expected. + } + + Assert.assertEquals(0, logChannel.position()); + Assert.assertEquals(8, fileChannel.position()); + + try { + logChannel.write(Unpooled.wrappedBuffer(new byte[1])); + Assert.fail("Expected writes after a partial flush failure to fail"); + } catch (IOException expected) { + // Expected. + } + try { + logChannel.flush(); + Assert.fail("Expected flush after a partial flush failure to fail"); + } catch (IOException expected) { + // Expected. + } + try { + logChannel.forceWrite(false); + Assert.fail("Expected forceWrite after a partial flush failure to fail"); + } catch (IOException expected) { + // Expected. + } + + Assert.assertEquals(0, logChannel.position()); + Assert.assertEquals(8, fileChannel.position()); + + logChannel.close(); + } + + @Test + public void testPartialFlushFailurePreventsLedgersMapHeaderUpdate() throws Exception { + File newLogFile = File.createTempFile("test", "log"); + newLogFile.deleteOnExit(); + FileChannel delegate = new RandomAccessFile(newLogFile, "rw").getChannel(); + delegate.position(DefaultEntryLogger.LOGFILE_HEADER_SIZE); + PartialFailingFileChannel fileChannel = new PartialFailingFileChannel(delegate, 8); + + DefaultEntryLogger.BufferedLogChannel logChannel = new DefaultEntryLogger.BufferedLogChannel( + UnpooledByteBufAllocator.DEFAULT, fileChannel, 16, INTERNAL_BUFFER_READ_CAPACITY, 1L, newLogFile, 0); + + try { + logChannel.write(Unpooled.wrappedBuffer(new byte[16])); + Assert.fail("Expected the internal flush to fail"); + } catch (IOException expected) { + // Expected. + } + + logChannel.registerWrittenEntry(1234L, 99L); + try { + logChannel.appendLedgersMap(); + Assert.fail("Expected appendLedgersMap after a partial flush failure to fail"); + } catch (IOException expected) { + // Expected. + } + + ByteBuffer mapInfo = ByteBuffer.allocate(Long.BYTES + Integer.BYTES); + Assert.assertEquals(mapInfo.capacity(), + fileChannel.read(mapInfo, DefaultEntryLogger.LEDGERS_MAP_OFFSET_POSITION)); + mapInfo.flip(); + Assert.assertEquals(0L, mapInfo.getLong()); + Assert.assertEquals(0, mapInfo.getInt()); + + logChannel.close(); + } + + @Test + public void testForceWriteFailurePoisonsBufferedChannel() throws Exception { + File newLogFile = File.createTempFile("test", "log"); + newLogFile.deleteOnExit(); + FileChannel delegate = new RandomAccessFile(newLogFile, "rw").getChannel(); + PartialFailingFileChannel fileChannel = PartialFailingFileChannel.failingForce(delegate); + + BufferedChannel logChannel = new BufferedChannel(UnpooledByteBufAllocator.DEFAULT, fileChannel, + 16, INTERNAL_BUFFER_READ_CAPACITY, 0); + + logChannel.write(Unpooled.wrappedBuffer(new byte[] { 1 })); + logChannel.flush(); + + try { + logChannel.forceWrite(false); + Assert.fail("Expected forceWrite failure"); + } catch (IOException expected) { + // Expected. + } + try { + logChannel.write(Unpooled.wrappedBuffer(new byte[] { 2 })); + Assert.fail("Expected writes after forceWrite failure to fail"); + } catch (IOException expected) { + // Expected. + } + + Assert.assertEquals(1, logChannel.position()); + + logChannel.close(); + } + + @Test + public void testLedgersMapHeaderPositionedWriteIsFullyWritten() throws Exception { + File newLogFile = File.createTempFile("test", "log"); + newLogFile.deleteOnExit(); + FileChannel delegate = new RandomAccessFile(newLogFile, "rw").getChannel(); + writeEntryLogHeader(delegate); + PartialFailingFileChannel fileChannel = PartialFailingFileChannel.shortPositionedWrite( + delegate, DefaultEntryLogger.LEDGERS_MAP_OFFSET_POSITION, Long.BYTES); + + DefaultEntryLogger.BufferedLogChannel logChannel = new DefaultEntryLogger.BufferedLogChannel( + UnpooledByteBufAllocator.DEFAULT, fileChannel, 64, INTERNAL_BUFFER_READ_CAPACITY, 1L, newLogFile, 0); + + logChannel.write(Unpooled.wrappedBuffer(new byte[] { 1 })); + logChannel.flush(); + + long actualMapOffset = logChannel.position(); + logChannel.registerWrittenEntry(1234L, 1L); + logChannel.appendLedgersMap(); + + ByteBuffer mapInfo = ByteBuffer.allocate(Long.BYTES + Integer.BYTES); + Assert.assertEquals(mapInfo.capacity(), + fileChannel.read(mapInfo, DefaultEntryLogger.LEDGERS_MAP_OFFSET_POSITION)); + mapInfo.flip(); + Assert.assertEquals(actualMapOffset, mapInfo.getLong()); + Assert.assertEquals(1, mapInfo.getInt()); + Assert.assertTrue(fileChannel.shortPositionedWriteDone); + + logChannel.close(); + } + + @Test + public void testLedgersMapHeaderWriteFailurePoisonsBufferedChannel() throws Exception { + File newLogFile = File.createTempFile("test", "log"); + newLogFile.deleteOnExit(); + FileChannel delegate = new RandomAccessFile(newLogFile, "rw").getChannel(); + writeEntryLogHeader(delegate); + PartialFailingFileChannel fileChannel = PartialFailingFileChannel.failingPositionedWrite( + delegate, DefaultEntryLogger.LEDGERS_MAP_OFFSET_POSITION); + + DefaultEntryLogger.BufferedLogChannel logChannel = new DefaultEntryLogger.BufferedLogChannel( + UnpooledByteBufAllocator.DEFAULT, fileChannel, 64, INTERNAL_BUFFER_READ_CAPACITY, 1L, newLogFile, 0); + + logChannel.write(Unpooled.wrappedBuffer(new byte[] { 1 })); + logChannel.flush(); + long mapOffset = logChannel.position(); + logChannel.registerWrittenEntry(1234L, 1L); + + try { + logChannel.appendLedgersMap(); + Assert.fail("Expected header write failure"); + } catch (IOException expected) { + // Expected. + } + try { + logChannel.write(Unpooled.wrappedBuffer(new byte[] { 2 })); + Assert.fail("Expected writes after header write failure to fail"); + } catch (IOException expected) { + // Expected. + } + + ByteBuffer mapInfo = ByteBuffer.allocate(Long.BYTES + Integer.BYTES); + Assert.assertEquals(mapInfo.capacity(), + fileChannel.read(mapInfo, DefaultEntryLogger.LEDGERS_MAP_OFFSET_POSITION)); + mapInfo.flip(); + Assert.assertEquals(0L, mapInfo.getLong()); + Assert.assertEquals(0, mapInfo.getInt()); + + Assert.assertEquals(mapOffset + DefaultEntryLogger.LEDGERS_MAP_HEADER_SIZE + + DefaultEntryLogger.LEDGERS_MAP_ENTRY_SIZE, logChannel.position()); + + logChannel.close(); + } + public void testBufferedChannel(int byteBufLength, int numOfWrites, int unpersistedBytesBound, boolean flush, boolean shouldForceWrite) throws Exception { File newLogFile = File.createTempFile("test", "log"); @@ -211,4 +401,172 @@ private static ByteBuf generateEntry(int length) { bb.writeBytes(data); return bb; } + + private static void writeEntryLogHeader(FileChannel fileChannel) throws IOException { + ByteBuffer header = ByteBuffer.allocate(DefaultEntryLogger.LOGFILE_HEADER_SIZE); + header.put("BKLO".getBytes("UTF-8")); + header.putInt(DefaultEntryLogger.HEADER_CURRENT_VERSION); + header.position(DefaultEntryLogger.LOGFILE_HEADER_SIZE); + header.flip(); + while (header.hasRemaining()) { + fileChannel.write(header); + } + fileChannel.position(DefaultEntryLogger.LOGFILE_HEADER_SIZE); + } + + private static final class PartialFailingFileChannel extends FileChannel { + private final FileChannel delegate; + private final int bytesBeforeFailure; + private final boolean failRegularWrite; + private final long shortPositionedWritePosition; + private final int shortPositionedWriteBytes; + private final long failingPositionedWritePosition; + private final boolean failForce; + private boolean partialWriteDone; + private boolean failureInjected; + private boolean shortPositionedWriteDone; + private boolean positionedWriteFailureInjected; + + private PartialFailingFileChannel(FileChannel delegate, int bytesBeforeFailure) { + this(delegate, bytesBeforeFailure, true, -1L, -1, -1L, false); + } + + private PartialFailingFileChannel(FileChannel delegate, int bytesBeforeFailure, + boolean failRegularWrite, long shortPositionedWritePosition, + int shortPositionedWriteBytes, long failingPositionedWritePosition, + boolean failForce) { + this.delegate = delegate; + this.bytesBeforeFailure = bytesBeforeFailure; + this.failRegularWrite = failRegularWrite; + this.shortPositionedWritePosition = shortPositionedWritePosition; + this.shortPositionedWriteBytes = shortPositionedWriteBytes; + this.failingPositionedWritePosition = failingPositionedWritePosition; + this.failForce = failForce; + } + + private static PartialFailingFileChannel shortPositionedWrite(FileChannel delegate, long position, int bytes) { + return new PartialFailingFileChannel(delegate, 0, false, position, bytes, -1L, false); + } + + private static PartialFailingFileChannel failingPositionedWrite(FileChannel delegate, long position) { + return new PartialFailingFileChannel(delegate, 0, false, -1L, -1, position, false); + } + + private static PartialFailingFileChannel failingForce(FileChannel delegate) { + return new PartialFailingFileChannel(delegate, 0, false, -1L, -1, -1L, true); + } + + @Override + public int write(ByteBuffer src) throws IOException { + if (failRegularWrite && !partialWriteDone) { + int oldLimit = src.limit(); + src.limit(src.position() + bytesBeforeFailure); + int written = delegate.write(src); + src.limit(oldLimit); + partialWriteDone = true; + return written; + } else if (failRegularWrite && !failureInjected) { + failureInjected = true; + throw new IOException("simulated write failure after partial write"); + } + return delegate.write(src); + } + + @Override + public long position() throws IOException { + return delegate.position(); + } + + @Override + public FileChannel position(long newPosition) throws IOException { + delegate.position(newPosition); + return this; + } + + @Override + public int read(ByteBuffer dst) throws IOException { + return delegate.read(dst); + } + + @Override + public long read(ByteBuffer[] dsts, int offset, int length) throws IOException { + return delegate.read(dsts, offset, length); + } + + @Override + public long write(ByteBuffer[] srcs, int offset, int length) throws IOException { + return delegate.write(srcs, offset, length); + } + + @Override + public long size() throws IOException { + return delegate.size(); + } + + @Override + public FileChannel truncate(long size) throws IOException { + delegate.truncate(size); + return this; + } + + @Override + public void force(boolean metaData) throws IOException { + if (failForce) { + throw new IOException("simulated force failure"); + } + delegate.force(metaData); + } + + @Override + public long transferTo(long position, long count, WritableByteChannel target) throws IOException { + return delegate.transferTo(position, count, target); + } + + @Override + public long transferFrom(ReadableByteChannel src, long position, long count) throws IOException { + return delegate.transferFrom(src, position, count); + } + + @Override + public int read(ByteBuffer dst, long position) throws IOException { + return delegate.read(dst, position); + } + + @Override + public int write(ByteBuffer src, long position) throws IOException { + if (!positionedWriteFailureInjected && position == failingPositionedWritePosition) { + positionedWriteFailureInjected = true; + throw new IOException("simulated positioned write failure"); + } + if (!shortPositionedWriteDone && position == shortPositionedWritePosition) { + int oldLimit = src.limit(); + src.limit(src.position() + Math.min(shortPositionedWriteBytes, src.remaining())); + int written = delegate.write(src, position); + src.limit(oldLimit); + shortPositionedWriteDone = true; + return written; + } + return delegate.write(src, position); + } + + @Override + public MappedByteBuffer map(MapMode mode, long position, long size) throws IOException { + return delegate.map(mode, position, size); + } + + @Override + public FileLock lock(long position, long size, boolean shared) throws IOException { + return delegate.lock(position, size, shared); + } + + @Override + public FileLock tryLock(long position, long size, boolean shared) throws IOException { + return delegate.tryLock(position, size, shared); + } + + @Override + protected void implCloseChannel() throws IOException { + delegate.close(); + } + } } From 0b74a3190762a026fce57e3625ef12c8228857a7 Mon Sep 17 00:00:00 2001 From: root Date: Tue, 4 Aug 2026 11:43:16 +0800 Subject: [PATCH 2/9] fix: fail closed on entry log flush failure --- .../apache/bookkeeper/bookie/BookieImpl.java | 9 ++ .../bookie/EntryLogManagerBase.java | 78 +++++++++++--- .../EntryLogManagerForEntryLogPerLedger.java | 6 +- .../EntryLogManagerForSingleEntryLog.java | 4 +- .../bookie/EntryLogWriteException.java | 35 ++++++ .../bookie/EntryLoggerAllocator.java | 41 ++++--- .../bookie/InterleavedLedgerStorage.java | 2 + .../bookie/SortedLedgerStorage.java | 14 +++ .../apache/bookkeeper/bookie/SyncThread.java | 8 ++ .../bookie/storage/ldb/DbLedgerStorage.java | 7 ++ .../ldb/SingleDirectoryDbLedgerStorage.java | 15 +++ .../bookkeeper/bookie/BookieImplTest.java | 101 ++++++++++++++++++ .../bookkeeper/bookie/SyncThreadTest.java | 55 ++++++++++ .../ldb/DbLedgerStorageWriteCacheTest.java | 91 ++++++++++++---- 14 files changed, 411 insertions(+), 55 deletions(-) create mode 100644 bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogWriteException.java diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookieImpl.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookieImpl.java index 362a20129a1..8b7bac2793b 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookieImpl.java +++ b/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); @@ -1033,6 +1036,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 +1131,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); diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerBase.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerBase.java index df8afb4d238..de2cd5d34ce 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerBase.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerBase.java @@ -73,13 +73,20 @@ public long addEntry(long ledger, ByteBuf entry, boolean rollLog) throws IOExcep ByteBuf sizeBuffer = sizeBufferForAdd.get(); sizeBuffer.clear(); sizeBuffer.writeInt(entry.readableBytes()); - logChannel.write(sizeBuffer); - - long pos = logChannel.position(); - logChannel.write(entry); - logChannel.registerWrittenEntry(ledger, entrySize); - - return (logChannel.getLogId() << 32L) | pos; + try { + logChannel.write(sizeBuffer); + + long pos = logChannel.position(); + logChannel.write(entry); + logChannel.registerWrittenEntry(ledger, entrySize); + + return (logChannel.getLogId() << 32L) | pos; + } catch (EntryLogWriteException e) { + throw e; + } catch (IOException e) { + throw new EntryLogWriteException( + "Failed to write entry to entry log " + logChannel.getLogId() + " for ledger " + ledger, e); + } } boolean reachEntryLogLimit(BufferedLogChannel logChannel, long size) { @@ -125,11 +132,31 @@ public void flush() throws IOException { void flushLogChannel(BufferedLogChannel logChannel, boolean forceMetadata) throws IOException { if (logChannel != null) { - logChannel.flushAndForceWrite(forceMetadata); + flushAndForceWrite(logChannel, forceMetadata); log.debug().attr("logId", () -> logChannel.getLogId()).log("Flush and sync current entry logger"); } } + void flushAndForceWrite(BufferedLogChannel logChannel, boolean forceMetadata) throws IOException { + try { + logChannel.flushAndForceWrite(forceMetadata); + } catch (EntryLogWriteException e) { + throw e; + } catch (IOException e) { + throw new EntryLogWriteException("Failed to flush entry log " + logChannel.getLogId(), e); + } + } + + void flushAndForceWriteIfRegularFlush(BufferedLogChannel logChannel, boolean forceMetadata) throws IOException { + try { + logChannel.flushAndForceWriteIfRegularFlush(forceMetadata); + } catch (EntryLogWriteException e) { + throw e; + } catch (IOException e) { + throw new EntryLogWriteException("Failed to flush entry log " + logChannel.getLogId(), e); + } + } + /* * Creates a new log file. This method should be guarded by a lock, * so callers of this method should be in right scope of the lock. @@ -156,12 +183,27 @@ void createNewLog(long ledgerId, String reason) throws IOException { if (null != logChannel) { // flush the internal buffer back to filesystem but not sync disk - logChannel.flush(); - - // Append ledgers map at the end of entry log - logChannel.appendLedgersMap(); + try { + logChannel.flush(); + + // Append ledgers map at the end of entry log + logChannel.appendLedgersMap(); + } catch (EntryLogWriteException e) { + throw e; + } catch (IOException e) { + throw new EntryLogWriteException( + "Failed to rotate entry log " + logChannel.getLogId() + " for ledger " + ledgerId, e); + } - BufferedLogChannel newLogChannel = entryLoggerAllocator.createNewLog(selectDirForNextEntryLog()); + File dirForNextEntryLog = selectDirForNextEntryLog(); + BufferedLogChannel newLogChannel; + try { + newLogChannel = entryLoggerAllocator.createNewLog(dirForNextEntryLog); + } catch (EntryLogWriteException e) { + throw e; + } catch (IOException e) { + throw new EntryLogWriteException("Failed to create a new entry log for ledger " + ledgerId, e); + } entryLoggerAllocator.setWritingLogId(newLogChannel.getLogId()); setCurrentLogForLedgerAndAddToRotate(ledgerId, newLogChannel); log.info() @@ -172,7 +214,15 @@ void createNewLog(long ledgerId, String reason) throws IOException { listener.onRotateEntryLog(); } } else { - BufferedLogChannel newLogChannel = entryLoggerAllocator.createNewLog(selectDirForNextEntryLog()); + File dirForNextEntryLog = selectDirForNextEntryLog(); + BufferedLogChannel newLogChannel; + try { + newLogChannel = entryLoggerAllocator.createNewLog(dirForNextEntryLog); + } catch (EntryLogWriteException e) { + throw e; + } catch (IOException e) { + throw new EntryLogWriteException("Failed to create a new entry log for ledger " + ledgerId, e); + } entryLoggerAllocator.setWritingLogId(newLogChannel.getLogId()); setCurrentLogForLedgerAndAddToRotate(ledgerId, newLogChannel); } diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerForEntryLogPerLedger.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerForEntryLogPerLedger.java index ee11ee55fd2..1caa423e5d0 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerForEntryLogPerLedger.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerForEntryLogPerLedger.java @@ -680,10 +680,10 @@ BufferedLogChannel getCurrentLogForLedgerForAddEntry(long ledgerId, int entrySiz * logChannel, since Bookie must have turned to readonly mode and * the addEntry traffic would be from GC and it is ok to proceed in * this case. - */ + */ if ((diskFull && (!allDisksFull)) || reachEntryLogLimit || (logChannel == null)) { if (logChannel != null) { - logChannel.flushAndForceWriteIfRegularFlush(false); + flushAndForceWriteIfRegularFlush(logChannel, false); } createNewLog(ledgerId, ": diskFull = " + diskFull + ", allDisksFull = " + allDisksFull @@ -699,7 +699,7 @@ BufferedLogChannel getCurrentLogForLedgerForAddEntry(long ledgerId, int entrySiz @Override public void flushRotatedLogs() throws IOException { for (BufferedLogChannel channel : rotatedLogChannels) { - channel.flushAndForceWrite(true); + flushAndForceWrite(channel, true); // since this channel is only used for writing, after flushing the channel, // we had to close the underlying file channel. Otherwise, we might end up // leaking fds which cause the disk spaces could not be reclaimed. diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerForSingleEntryLog.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerForSingleEntryLog.java index 71f4f9d10ce..136b2bfb05e 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerForSingleEntryLog.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerForSingleEntryLog.java @@ -102,7 +102,7 @@ synchronized BufferedLogChannel getCurrentLogForLedgerForAddEntry(long ledgerId, boolean createNewLog = shouldCreateNewEntryLog.get(); if (createNewLog || reachEntryLogLimit) { if (activeLogChannel != null) { - activeLogChannel.flushAndForceWriteIfRegularFlush(false); + flushAndForceWriteIfRegularFlush(activeLogChannel, false); } createNewLog(UNASSIGNED_LEDGERID, ": createNewLog = " + createNewLog + ", reachEntryLogLimit = " + reachEntryLogLimit); @@ -188,7 +188,7 @@ void flushRotatedLogs() throws IOException { while (chIter.hasNext()) { BufferedLogChannel channel = chIter.next(); try { - channel.flushAndForceWrite(true); + flushAndForceWrite(channel, true); } catch (IOException ioe) { // rescue from flush exception, add unflushed channels back synchronized (this) { diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogWriteException.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogWriteException.java new file mode 100644 index 00000000000..11d5ad32d14 --- /dev/null +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogWriteException.java @@ -0,0 +1,35 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ + +package org.apache.bookkeeper.bookie; + +import java.io.IOException; + +/** + * Indicates an entry log write, flush, or fsync failure that leaves the writer state uncertain. + */ +public class EntryLogWriteException extends IOException { + private static final long serialVersionUID = -6456783806451195011L; + + public EntryLogWriteException(String message, IOException cause) { + super(message, cause); + } +} diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLoggerAllocator.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLoggerAllocator.java index 669d72e1094..13475d4f051 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLoggerAllocator.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLoggerAllocator.java @@ -187,22 +187,37 @@ private synchronized BufferedLogChannel allocateNewLog(File dirForNextEntryLog, BufferedLogChannel logChannel = new BufferedLogChannel(byteBufAllocator, channel, conf.getWriteBufferBytes(), conf.getReadBufferBytes(), preallocatedLogId, newLogFile, conf.getFlushIntervalInBytes()); - logfileHeader.readerIndex(0); - logChannel.write(logfileHeader); + boolean success = false; + try { + logfileHeader.readerIndex(0); + logChannel.write(logfileHeader); - for (File f : ledgersDirs) { - setLastLogId(f, preallocatedLogId); - } + for (File f : ledgersDirs) { + setLastLogId(f, preallocatedLogId); + } - if (suffix.equals(DefaultEntryLogger.LOG_FILE_SUFFIX)) { - recentlyCreatedEntryLogsStatus.createdEntryLog(preallocatedLogId); - } + if (suffix.equals(DefaultEntryLogger.LOG_FILE_SUFFIX)) { + recentlyCreatedEntryLogsStatus.createdEntryLog(preallocatedLogId); + } - log.info() - .attr("file", newLogFile) - .attr("logId", preallocatedLogId) - .log("Created new entry log file"); - return logChannel; + log.info() + .attr("file", newLogFile) + .attr("logId", preallocatedLogId) + .log("Created new entry log file"); + success = true; + return logChannel; + } finally { + if (!success) { + try { + logChannel.close(); + } catch (IOException closeError) { + log.warn() + .exception(closeError) + .attr("file", newLogFile) + .log("Failed to close entry log file after allocation failure"); + } + } + } } diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/InterleavedLedgerStorage.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/InterleavedLedgerStorage.java index fa1b724622d..0db7b93294b 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/InterleavedLedgerStorage.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/InterleavedLedgerStorage.java @@ -466,6 +466,8 @@ private void flushOrCheckpoint(boolean isCheckpointFlush) } else { entryLogger.flush(); } + } catch (EntryLogWriteException e) { + throw e; } catch (LedgerDirsManager.NoWritableLedgerDirException e) { throw e; } catch (IOException ioe) { diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SortedLedgerStorage.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SortedLedgerStorage.java index bc3d586008c..8da0bb7f6a1 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SortedLedgerStorage.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SortedLedgerStorage.java @@ -321,6 +321,8 @@ public void run() { if (interleavedLedgerStorage.getEntryLogger().commitEntryMemTableFlush()) { interleavedLedgerStorage.checkpointer.startCheckpoint(cp); } + } catch (EntryLogWriteException e) { + fatalEntryLogWriteFailure(e); } catch (Exception e) { stateManager.transitionToReadOnlyMode(); log.error().exception(e).log("Exception thrown while flushing skip list cache."); @@ -337,6 +339,18 @@ public void onRotateEntryLog() { // flushed to the entry log file. } + private void fatalEntryLogWriteFailure(EntryLogWriteException e) { + LOG.error("Fatal entry log write failure while flushing skip list cache.", e); + if (stateManager instanceof BookieStateManager) { + StateManager.ShutdownHandler shutdownHandler = ((BookieStateManager) stateManager).getShutdownHandler(); + if (shutdownHandler != null) { + shutdownHandler.shutdown(ExitCode.BOOKIE_EXCEPTION); + return; + } + } + stateManager.transitionToReadOnlyMode(); + } + BookieStateManager getStateManager(){ return (BookieStateManager) stateManager; } diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SyncThread.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SyncThread.java index f6a4b9def0c..a838b837ed6 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SyncThread.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SyncThread.java @@ -146,6 +146,10 @@ private void flush() { log.error().exception(e).log("No writeable ledger directories"); dirsListener.allDisksFull(true); return; + } catch (EntryLogWriteException e) { + log.error("Fatal entry log write failure while flushing ledgers", e); + dirsListener.fatalError(); + return; } catch (IOException e) { log.error().exception(e).log("Exception flushing ledgers"); return; @@ -177,6 +181,10 @@ public void checkpoint(Checkpoint checkpoint) { log.error().exception(e).log("No writeable ledger directories"); dirsListener.allDisksFull(true); return; + } catch (EntryLogWriteException e) { + log.error("Fatal entry log write failure while checkpointing ledgers", e); + dirsListener.fatalError(); + return; } catch (IOException e) { log.error().exception(e).log("Exception flushing ledgers"); return; diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorage.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorage.java index 0f672de085d..6d668a8d61b 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorage.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorage.java @@ -53,6 +53,7 @@ import org.apache.bookkeeper.bookie.LastAddConfirmedUpdateNotification; import org.apache.bookkeeper.bookie.LedgerCache; import org.apache.bookkeeper.bookie.LedgerDirsManager; +import org.apache.bookkeeper.bookie.LedgerDirsManager.LedgerDirsListener; import org.apache.bookkeeper.bookie.LedgerStorage; import org.apache.bookkeeper.bookie.StateManager; import org.apache.bookkeeper.bookie.storage.EntryLogger; @@ -301,6 +302,12 @@ public void setCheckpointer(Checkpointer checkpointer) { ledgerStorageList.forEach(s -> s.setCheckpointer(checkpointer)); } + public void setFatalErrorListener(LedgerDirsListener fatalErrorListener) { + if (fatalErrorListener != null) { + ledgerStorageList.forEach(s -> s.setFatalErrorListener(fatalErrorListener)); + } + } + @Override public void start() { ledgerStorageList.forEach(LedgerStorage::start); diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java index 6778de2f824..15b71e2c501 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java @@ -57,6 +57,7 @@ import org.apache.bookkeeper.bookie.CheckpointSource.Checkpoint; import org.apache.bookkeeper.bookie.Checkpointer; import org.apache.bookkeeper.bookie.CompactableLedgerStorage; +import org.apache.bookkeeper.bookie.EntryLogWriteException; import org.apache.bookkeeper.bookie.EntryLocation; import org.apache.bookkeeper.bookie.GarbageCollectionStatus; import org.apache.bookkeeper.bookie.GarbageCollectorThread; @@ -132,6 +133,7 @@ protected Thread newThread(Runnable r, String name) { private CheckpointSource checkpointSource = CheckpointSource.DEFAULT; private Checkpoint lastCheckpoint = Checkpoint.MIN; + private volatile LedgerDirsListener fatalErrorListener = new LedgerDirsListener() { }; private final long writeCacheMaxSize; private final long readCacheMaxSize; @@ -252,6 +254,12 @@ public void setCheckpointSource(CheckpointSource checkpointSource) { @Override public void setCheckpointer(Checkpointer checkpointer) { } + void setFatalErrorListener(LedgerDirsListener fatalErrorListener) { + if (fatalErrorListener != null) { + this.fatalErrorListener = fatalErrorListener; + } + } + /** * Evict all the ledger info object that were not used recently. */ @@ -530,6 +538,8 @@ private void triggerFlushAndAddEntry(long ledgerId, long entryId, ByteBuf entry) long startTime = System.nanoTime(); try { flush(); + } catch (EntryLogWriteException e) { + notifyFatalEntryLogWriteFailure(e); } catch (IOException e) { log.error().exception(e).log("Error during flush"); } finally { @@ -564,6 +574,11 @@ private void triggerFlushAndAddEntry(long ledgerId, long entryId, ByteBuf entry) throw new OperationRejectedException(); } + private void notifyFatalEntryLogWriteFailure(EntryLogWriteException e) { + log.error("Fatal entry log write failure during background flush", e); + fatalErrorListener.fatalError(); + } + @Override public ByteBuf getEntry(long ledgerId, long entryId) throws IOException, BookieException { long startTime = MathUtils.nowInNano(); diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieImplTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieImplTest.java index 748c3fbc4cd..56f125830b6 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieImplTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieImplTest.java @@ -27,20 +27,29 @@ import static org.mockito.Mockito.spy; import io.netty.buffer.ByteBuf; +import io.netty.buffer.ByteBufAllocator; import io.netty.buffer.Unpooled; import io.netty.buffer.UnpooledByteBufAllocator; +import java.io.IOException; import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import org.apache.bookkeeper.bookie.storage.EntryLogger; +import org.apache.bookkeeper.bookie.storage.ldb.DbLedgerStorage; +import org.apache.bookkeeper.bookie.storage.ldb.SingleDirectoryDbLedgerStorage; import org.apache.bookkeeper.client.BookKeeper; import org.apache.bookkeeper.conf.ServerConfiguration; import org.apache.bookkeeper.conf.TestBKConfiguration; import org.apache.bookkeeper.discover.RegistrationManager; +import org.apache.bookkeeper.meta.LedgerManager; import org.apache.bookkeeper.meta.MetadataBookieDriver; import org.apache.bookkeeper.net.BookieId; import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks; import org.apache.bookkeeper.proto.checksum.DigestManager; import org.apache.bookkeeper.stats.NullStatsLogger; +import org.apache.bookkeeper.stats.StatsLogger; import org.apache.bookkeeper.test.BookKeeperClusterTestCase; import org.apache.bookkeeper.util.ByteBufList; import org.apache.bookkeeper.util.PortManager; @@ -128,6 +137,43 @@ public void testRecoveryAddEntry() throws Exception { mockAddEntryReleased(RECOVERY_ADD); } + @Test + public void testBackgroundEntryLogFlushFailureShutsDownBookie() throws Exception { + ServerConfiguration conf = newServerConfiguration(); + conf.setLedgerStorageClass(FailOnFlushDbLedgerStorage.class.getName()); + conf.setJournalWriteData(false); + conf.setProperty(DbLedgerStorage.WRITE_CACHE_MAX_SIZE_MB, 1); + conf.setProperty("dbStorage_maxThrottleTimeMs", 1000); + FailOnFlushDbLedgerStorage.failNextFlushWithEntryLogWriteException.set(false); + + CountDownLatch shutdownLatch = new CountDownLatch(1); + TestBookieImpl.Resources resources = new TestBookieImpl.ResourceBuilder(conf).build(); + BookieImpl bookie = new TestBookieImpl(resources) { + @Override + int shutdown(int exitCode) { + int result = super.shutdown(exitCode); + if (exitCode == ExitCode.BOOKIE_EXCEPTION) { + shutdownLatch.countDown(); + } + return result; + } + }; + + try { + bookie.start(); + FailOnFlushDbLedgerStorage.failNextFlushWithEntryLogWriteException.set(true); + writeUntilCacheFlushIsTriggered(bookie, 10L, "masterKey".getBytes(StandardCharsets.UTF_8)); + + assertTrue("Entry log flush failure should shut down the bookie", + shutdownLatch.await(10, TimeUnit.SECONDS)); + } finally { + FailOnFlushDbLedgerStorage.failNextFlushWithEntryLogWriteException.set(false); + if (bookie.isRunning()) { + bookie.shutdown(); + } + } + } + public void mockAddEntryReleased(int flag) throws Exception { final String metadataServiceUri = zkUtil.getMetadataServiceUri(); ServerConfiguration conf = TestBKConfiguration.newServerConfiguration(); @@ -192,4 +238,59 @@ private ByteBuf generateEntry(long ledger, long entry) { bb.writeBytes(data); return bb; } + + private void writeUntilCacheFlushIsTriggered(BookieImpl bookie, long ledgerId, byte[] masterKey) + throws Exception { + for (int i = 0; i < 20; i++) { + try { + bookie.addEntry(generateLargeEntry(ledgerId, i), false, + (rc, lid, eid, addr, ctx) -> { }, null, masterKey); + } catch (BookieException.OperationRejectedException e) { + return; + } + } + } + + private ByteBuf generateLargeEntry(long ledger, long entry) { + ByteBuf bb = Unpooled.buffer(100 * 1024 + 3 * Long.BYTES); + bb.writeLong(ledger); + bb.writeLong(entry); + bb.writeLong(entry - 1); + bb.writeZero(100 * 1024); + return bb; + } + + public static class FailOnFlushDbLedgerStorage extends DbLedgerStorage { + private static final AtomicBoolean failNextFlushWithEntryLogWriteException = new AtomicBoolean(false); + + @Override + protected SingleDirectoryDbLedgerStorage newSingleDirectoryDbLedgerStorage(ServerConfiguration conf, + LedgerManager ledgerManager, LedgerDirsManager ledgerDirsManager, LedgerDirsManager indexDirsManager, + EntryLogger entryLogger, StatsLogger statsLogger, long writeCacheSize, long readCacheSize, + int readAheadCacheBatchSize, long readAheadCacheBatchBytesSize) + throws IOException { + return new FailOnFlushSingleDirectoryDbLedgerStorage(conf, ledgerManager, ledgerDirsManager, + indexDirsManager, entryLogger, statsLogger, allocator, writeCacheSize, readCacheSize, + readAheadCacheBatchSize, readAheadCacheBatchBytesSize); + } + } + + private static class FailOnFlushSingleDirectoryDbLedgerStorage extends SingleDirectoryDbLedgerStorage { + FailOnFlushSingleDirectoryDbLedgerStorage(ServerConfiguration conf, LedgerManager ledgerManager, + LedgerDirsManager ledgerDirsManager, LedgerDirsManager indexDirsManager, EntryLogger entryLogger, + StatsLogger statsLogger, ByteBufAllocator allocator, long writeCacheSize, long readCacheSize, + int readAheadCacheBatchSize, long readAheadCacheBatchBytesSize) + throws IOException { + super(conf, ledgerManager, ledgerDirsManager, indexDirsManager, entryLogger, statsLogger, allocator, + writeCacheSize, readCacheSize, readAheadCacheBatchSize, readAheadCacheBatchBytesSize); + } + + @Override + public void flush() throws IOException { + if (FailOnFlushDbLedgerStorage.failNextFlushWithEntryLogWriteException.compareAndSet(true, false)) { + throw new EntryLogWriteException("entry log flush failed", new IOException("injected")); + } + super.flush(); + } + } } diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/SyncThreadTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/SyncThreadTest.java index 6df1bacb80d..c5548d112b2 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/SyncThreadTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/SyncThreadTest.java @@ -223,6 +223,61 @@ public void checkpoint(Checkpoint checkpoint) t.shutdown(); } + @Test + public void testSyncThreadShutdownOnEntryLogWriteFailureDuringCheckpoint() throws Exception { + int flushInterval = 100; + ServerConfiguration conf = TestBKConfiguration.newServerConfiguration(); + conf.setFlushInterval(flushInterval); + CheckpointSource checkpointSource = new DummyCheckpointSource(); + final CountDownLatch fatalLatch = new CountDownLatch(1); + LedgerDirsListener listener = new LedgerDirsListener() { + @Override + public void fatalError() { + fatalLatch.countDown(); + } + }; + + LedgerStorage storage = new DummyLedgerStorage() { + @Override + public void checkpoint(Checkpoint checkpoint) + throws IOException { + throw new EntryLogWriteException( + "entry log flush failed", new IOException("injected")); + } + }; + final SyncThread t = new SyncThread(conf, listener, storage, checkpointSource, NullStatsLogger.INSTANCE); + t.startCheckpoint(Checkpoint.MAX); + assertTrue("Should have called fatal error", fatalLatch.await(10, TimeUnit.SECONDS)); + t.shutdown(); + } + + @Test + public void testSyncThreadShutdownOnEntryLogWriteFailureDuringFlush() throws Exception { + int flushInterval = 100; + ServerConfiguration conf = TestBKConfiguration.newServerConfiguration(); + conf.setFlushInterval(flushInterval); + CheckpointSource checkpointSource = new DummyCheckpointSource(); + final CountDownLatch fatalLatch = new CountDownLatch(1); + LedgerDirsListener listener = new LedgerDirsListener() { + @Override + public void fatalError() { + fatalLatch.countDown(); + } + }; + + LedgerStorage storage = new DummyLedgerStorage() { + @Override + public void flush() throws IOException { + throw new EntryLogWriteException( + "entry log flush failed", new IOException("injected")); + } + }; + final SyncThread t = new SyncThread(conf, listener, storage, checkpointSource, NullStatsLogger.INSTANCE); + t.requestFlush().get(10, TimeUnit.SECONDS); + assertTrue("Should have called fatal error", fatalLatch.await(10, TimeUnit.SECONDS)); + t.shutdown(); + } + /** * Test that if the ledger storage throws * a disk full exception, the owner of the sync diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageWriteCacheTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageWriteCacheTest.java index 102f7f5addc..0ddb3fe59e4 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageWriteCacheTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageWriteCacheTest.java @@ -21,6 +21,7 @@ package org.apache.bookkeeper.bookie.storage.ldb; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import io.netty.buffer.ByteBuf; @@ -28,10 +29,15 @@ import io.netty.buffer.Unpooled; import java.io.File; import java.io.IOException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import org.apache.bookkeeper.bookie.Bookie; import org.apache.bookkeeper.bookie.BookieException.OperationRejectedException; import org.apache.bookkeeper.bookie.BookieImpl; +import org.apache.bookkeeper.bookie.EntryLogWriteException; import org.apache.bookkeeper.bookie.LedgerDirsManager; +import org.apache.bookkeeper.bookie.LedgerDirsManager.LedgerDirsListener; import org.apache.bookkeeper.bookie.TestBookieImpl; import org.apache.bookkeeper.bookie.storage.EntryLogger; import org.apache.bookkeeper.conf.ServerConfiguration; @@ -64,6 +70,8 @@ protected SingleDirectoryDbLedgerStorage newSingleDirectoryDbLedgerStorage(Serve } private static class MockedSingleDirectoryDbLedgerStorage extends SingleDirectoryDbLedgerStorage { + private static final AtomicBoolean failNextFlushWithEntryLogWriteException = new AtomicBoolean(false); + public MockedSingleDirectoryDbLedgerStorage(ServerConfiguration conf, LedgerManager ledgerManager, LedgerDirsManager ledgerDirsManager, LedgerDirsManager indexDirsManager, EntryLogger entryLogger, StatsLogger statsLogger, @@ -75,29 +83,33 @@ public MockedSingleDirectoryDbLedgerStorage(ServerConfiguration conf, LedgerMana readAheadCacheBatchBytesSize); } - @Override - public void flush() throws IOException { - flushMutex.lock(); - try { - // Swap the write caches and block indefinitely to simulate a slow disk - WriteCache tmp = writeCacheBeingFlushed; - writeCacheBeingFlushed = writeCache; - writeCache = tmp; - - // since the cache is switched, we can allow flush to be triggered - hasFlushBeenTriggered.set(false); - - // Block the flushing thread - try { - Thread.sleep(1000); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - return; - } - } finally { - flushMutex.unlock(); - } - } + @Override + public void flush() throws IOException { + if (failNextFlushWithEntryLogWriteException.compareAndSet(true, false)) { + throw new EntryLogWriteException("entry log flush failed", new IOException("injected")); + } + + flushMutex.lock(); + try { + // Swap the write caches and block indefinitely to simulate a slow disk + WriteCache tmp = writeCacheBeingFlushed; + writeCacheBeingFlushed = writeCache; + writeCache = tmp; + + // since the cache is switched, we can allow flush to be triggered + hasFlushBeenTriggered.set(false); + + // Block the flushing thread + try { + Thread.sleep(1000); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + } finally { + flushMutex.unlock(); + } + } } } @@ -116,6 +128,7 @@ public void setup() throws Exception { conf.setProperty(DbLedgerStorage.WRITE_CACHE_MAX_SIZE_MB, 1); conf.setProperty(DbLedgerStorage.MAX_THROTTLE_TIME_MILLIS, 1000); conf.setLedgerDirNames(new String[] { tmpDir.toString() }); + MockedDbLedgerStorage.MockedSingleDirectoryDbLedgerStorage.failNextFlushWithEntryLogWriteException.set(false); Bookie bookie = new TestBookieImpl(conf); storage = (DbLedgerStorage) bookie.getLedgerStorage(); @@ -165,4 +178,36 @@ public void writeCacheFull() throws Exception { // Expected } } + + @Test + public void writeCacheFullEntryLogWriteFailureTriggersFatalError() throws Exception { + storage.setMasterKey(4, "key".getBytes()); + CountDownLatch fatalLatch = new CountDownLatch(1); + storage.setFatalErrorListener(new LedgerDirsListener() { + @Override + public void fatalError() { + fatalLatch.countDown(); + } + }); + MockedDbLedgerStorage.MockedSingleDirectoryDbLedgerStorage.failNextFlushWithEntryLogWriteException.set(true); + + try { + for (int i = 0; i < 10; i++) { + storage.addEntry(newEntry(4, i)); + } + fail("Should have thrown exception"); + } catch (OperationRejectedException e) { + // Expected because the background flush failed before rotating the write cache. + } + + assertTrue("Should have called fatal error", fatalLatch.await(10, TimeUnit.SECONDS)); + } + + private static ByteBuf newEntry(long ledgerId, long entryId) { + ByteBuf entry = Unpooled.buffer(100 * 1024 + 2 * 8); + entry.writeLong(ledgerId); + entry.writeLong(entryId); + entry.writeZero(100 * 1024); + return entry; + } } From 0a181558fd771c8922a248a19cd5fe4f64b8f3d8 Mon Sep 17 00:00:00 2001 From: root Date: Tue, 4 Aug 2026 12:04:33 +0800 Subject: [PATCH 3/9] test: add e2e coverage for entry log flush failure --- .../bookkeeper/bookie/BookieImplTest.java | 46 +----------- ...gerStorageEntryLogFlushFailureE2ETest.java | 74 +++++++++++++++++++ .../ldb/FailOnFlushDbLedgerStorage.java | 73 ++++++++++++++++++ 3 files changed, 151 insertions(+), 42 deletions(-) create mode 100644 bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageEntryLogFlushFailureE2ETest.java create mode 100644 bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/FailOnFlushDbLedgerStorage.java diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieImplTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieImplTest.java index 56f125830b6..057fc2f9f9b 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieImplTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieImplTest.java @@ -27,29 +27,24 @@ import static org.mockito.Mockito.spy; import io.netty.buffer.ByteBuf; -import io.netty.buffer.ByteBufAllocator; import io.netty.buffer.Unpooled; import io.netty.buffer.UnpooledByteBufAllocator; -import java.io.IOException; import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; -import org.apache.bookkeeper.bookie.storage.EntryLogger; import org.apache.bookkeeper.bookie.storage.ldb.DbLedgerStorage; -import org.apache.bookkeeper.bookie.storage.ldb.SingleDirectoryDbLedgerStorage; +import org.apache.bookkeeper.bookie.storage.ldb.FailOnFlushDbLedgerStorage; import org.apache.bookkeeper.client.BookKeeper; import org.apache.bookkeeper.conf.ServerConfiguration; import org.apache.bookkeeper.conf.TestBKConfiguration; import org.apache.bookkeeper.discover.RegistrationManager; -import org.apache.bookkeeper.meta.LedgerManager; import org.apache.bookkeeper.meta.MetadataBookieDriver; import org.apache.bookkeeper.net.BookieId; import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks; import org.apache.bookkeeper.proto.checksum.DigestManager; import org.apache.bookkeeper.stats.NullStatsLogger; -import org.apache.bookkeeper.stats.StatsLogger; import org.apache.bookkeeper.test.BookKeeperClusterTestCase; import org.apache.bookkeeper.util.ByteBufList; import org.apache.bookkeeper.util.PortManager; @@ -144,7 +139,7 @@ public void testBackgroundEntryLogFlushFailureShutsDownBookie() throws Exception conf.setJournalWriteData(false); conf.setProperty(DbLedgerStorage.WRITE_CACHE_MAX_SIZE_MB, 1); conf.setProperty("dbStorage_maxThrottleTimeMs", 1000); - FailOnFlushDbLedgerStorage.failNextFlushWithEntryLogWriteException.set(false); + FailOnFlushDbLedgerStorage.resetFailure(); CountDownLatch shutdownLatch = new CountDownLatch(1); TestBookieImpl.Resources resources = new TestBookieImpl.ResourceBuilder(conf).build(); @@ -161,13 +156,13 @@ int shutdown(int exitCode) { try { bookie.start(); - FailOnFlushDbLedgerStorage.failNextFlushWithEntryLogWriteException.set(true); + FailOnFlushDbLedgerStorage.injectFailureOnNextFlush(); writeUntilCacheFlushIsTriggered(bookie, 10L, "masterKey".getBytes(StandardCharsets.UTF_8)); assertTrue("Entry log flush failure should shut down the bookie", shutdownLatch.await(10, TimeUnit.SECONDS)); } finally { - FailOnFlushDbLedgerStorage.failNextFlushWithEntryLogWriteException.set(false); + FailOnFlushDbLedgerStorage.resetFailure(); if (bookie.isRunning()) { bookie.shutdown(); } @@ -260,37 +255,4 @@ private ByteBuf generateLargeEntry(long ledger, long entry) { return bb; } - public static class FailOnFlushDbLedgerStorage extends DbLedgerStorage { - private static final AtomicBoolean failNextFlushWithEntryLogWriteException = new AtomicBoolean(false); - - @Override - protected SingleDirectoryDbLedgerStorage newSingleDirectoryDbLedgerStorage(ServerConfiguration conf, - LedgerManager ledgerManager, LedgerDirsManager ledgerDirsManager, LedgerDirsManager indexDirsManager, - EntryLogger entryLogger, StatsLogger statsLogger, long writeCacheSize, long readCacheSize, - int readAheadCacheBatchSize, long readAheadCacheBatchBytesSize) - throws IOException { - return new FailOnFlushSingleDirectoryDbLedgerStorage(conf, ledgerManager, ledgerDirsManager, - indexDirsManager, entryLogger, statsLogger, allocator, writeCacheSize, readCacheSize, - readAheadCacheBatchSize, readAheadCacheBatchBytesSize); - } - } - - private static class FailOnFlushSingleDirectoryDbLedgerStorage extends SingleDirectoryDbLedgerStorage { - FailOnFlushSingleDirectoryDbLedgerStorage(ServerConfiguration conf, LedgerManager ledgerManager, - LedgerDirsManager ledgerDirsManager, LedgerDirsManager indexDirsManager, EntryLogger entryLogger, - StatsLogger statsLogger, ByteBufAllocator allocator, long writeCacheSize, long readCacheSize, - int readAheadCacheBatchSize, long readAheadCacheBatchBytesSize) - throws IOException { - super(conf, ledgerManager, ledgerDirsManager, indexDirsManager, entryLogger, statsLogger, allocator, - writeCacheSize, readCacheSize, readAheadCacheBatchSize, readAheadCacheBatchBytesSize); - } - - @Override - public void flush() throws IOException { - if (FailOnFlushDbLedgerStorage.failNextFlushWithEntryLogWriteException.compareAndSet(true, false)) { - throw new EntryLogWriteException("entry log flush failed", new IOException("injected")); - } - super.flush(); - } - } } diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageEntryLogFlushFailureE2ETest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageEntryLogFlushFailureE2ETest.java new file mode 100644 index 00000000000..ce093ecd367 --- /dev/null +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageEntryLogFlushFailureE2ETest.java @@ -0,0 +1,74 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.bookkeeper.bookie.storage.ldb; + +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; + +import java.util.concurrent.TimeUnit; +import org.apache.bookkeeper.bookie.BookieImpl; +import org.apache.bookkeeper.client.BKException; +import org.apache.bookkeeper.client.BookKeeper.DigestType; +import org.apache.bookkeeper.client.LedgerHandle; +import org.apache.bookkeeper.test.BookKeeperClusterTestCase; +import org.awaitility.Awaitility; +import org.junit.Test; + +public class DbLedgerStorageEntryLogFlushFailureE2ETest extends BookKeeperClusterTestCase { + private static final byte[] PASSWD = "passwd".getBytes(UTF_8); + + public DbLedgerStorageEntryLogFlushFailureE2ETest() { + super(1); + baseConf.setLedgerStorageClass(FailOnFlushDbLedgerStorage.class.getName()); + baseConf.setFlushInterval(60000); + baseConf.setGcWaitTime(60000); + baseConf.setProperty(DbLedgerStorage.WRITE_CACHE_MAX_SIZE_MB, 1); + baseConf.setProperty(DbLedgerStorage.MAX_THROTTLE_TIME_MILLIS, 1000); + baseClientConf.setAddEntryTimeout(5); + } + + @Test + public void testClientWriteFailsAndBookieShutsDownAfterEntryLogFlushFailure() throws Exception { + BookieImpl bookie = (BookieImpl) serverByIndex(0).getBookie(); + LedgerHandle lh = bkc.createLedger(1, 1, 1, DigestType.CRC32, PASSWD); + byte[] payload = new byte[100 * 1024]; + BKException clientFailure = null; + + FailOnFlushDbLedgerStorage.injectFailureOnNextFlush(); + try { + for (int i = 0; i < 20; i++) { + try { + lh.addEntry(payload); + } catch (BKException e) { + clientFailure = e; + break; + } + } + + assertNotNull("Client should observe a write failure after the entry log flush failure", clientFailure); + Awaitility.await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> + assertFalse("Bookie should be shut down after entry log flush failure", bookie.isRunning())); + } finally { + FailOnFlushDbLedgerStorage.resetFailure(); + } + } +} diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/FailOnFlushDbLedgerStorage.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/FailOnFlushDbLedgerStorage.java new file mode 100644 index 00000000000..cc6286de042 --- /dev/null +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/FailOnFlushDbLedgerStorage.java @@ -0,0 +1,73 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.bookkeeper.bookie.storage.ldb; + +import io.netty.buffer.ByteBufAllocator; +import java.io.IOException; +import java.util.concurrent.atomic.AtomicBoolean; +import org.apache.bookkeeper.bookie.EntryLogWriteException; +import org.apache.bookkeeper.bookie.LedgerDirsManager; +import org.apache.bookkeeper.bookie.storage.EntryLogger; +import org.apache.bookkeeper.conf.ServerConfiguration; +import org.apache.bookkeeper.meta.LedgerManager; +import org.apache.bookkeeper.stats.StatsLogger; + +public class FailOnFlushDbLedgerStorage extends DbLedgerStorage { + private static final AtomicBoolean failNextFlushWithEntryLogWriteException = new AtomicBoolean(false); + + public static void injectFailureOnNextFlush() { + failNextFlushWithEntryLogWriteException.set(true); + } + + public static void resetFailure() { + failNextFlushWithEntryLogWriteException.set(false); + } + + @Override + protected SingleDirectoryDbLedgerStorage newSingleDirectoryDbLedgerStorage(ServerConfiguration conf, + LedgerManager ledgerManager, LedgerDirsManager ledgerDirsManager, LedgerDirsManager indexDirsManager, + EntryLogger entryLogger, StatsLogger statsLogger, long writeCacheSize, long readCacheSize, + int readAheadCacheBatchSize, long readAheadCacheBatchBytesSize) + throws IOException { + return new FailOnFlushSingleDirectoryDbLedgerStorage(conf, ledgerManager, ledgerDirsManager, + indexDirsManager, entryLogger, statsLogger, allocator, writeCacheSize, readCacheSize, + readAheadCacheBatchSize, readAheadCacheBatchBytesSize); + } + + private static class FailOnFlushSingleDirectoryDbLedgerStorage extends SingleDirectoryDbLedgerStorage { + FailOnFlushSingleDirectoryDbLedgerStorage(ServerConfiguration conf, LedgerManager ledgerManager, + LedgerDirsManager ledgerDirsManager, LedgerDirsManager indexDirsManager, EntryLogger entryLogger, + StatsLogger statsLogger, ByteBufAllocator allocator, long writeCacheSize, long readCacheSize, + int readAheadCacheBatchSize, long readAheadCacheBatchBytesSize) + throws IOException { + super(conf, ledgerManager, ledgerDirsManager, indexDirsManager, entryLogger, statsLogger, allocator, + writeCacheSize, readCacheSize, readAheadCacheBatchSize, readAheadCacheBatchBytesSize); + } + + @Override + public void flush() throws IOException { + if (failNextFlushWithEntryLogWriteException.compareAndSet(true, false)) { + throw new EntryLogWriteException("entry log flush failed", new IOException("injected")); + } + super.flush(); + } + } +} From 3f42820d81ae6cf9bdf67c6604dc189075fd680e Mon Sep 17 00:00:00 2001 From: yangxianjungree <714696209@qq.com> Date: Tue, 4 Aug 2026 16:58:55 +0800 Subject: [PATCH 4/9] fix: close entrylog failure propagation gaps --- .../EntryLogManagerForEntryLogPerLedger.java | 9 +++- .../apache/bookkeeper/bookie/SyncThread.java | 7 ++- .../bookkeeper/bookie/BookieImplTest.java | 36 ++++++++++++++ .../bookie/DefaultEntryLogTest.java | 49 +++++++++++++++++++ .../bookkeeper/bookie/SyncThreadTest.java | 10 +++- .../ldb/FailOnFlushDbLedgerStorage.java | 5 +- 6 files changed, 110 insertions(+), 6 deletions(-) diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerForEntryLogPerLedger.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerForEntryLogPerLedger.java index 1caa423e5d0..85ee8edc5a9 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerForEntryLogPerLedger.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerForEntryLogPerLedger.java @@ -359,10 +359,15 @@ private void onCacheEntryRemoval(RemovalNotification // Append ledgers map at the end of entry log try { logChannel.appendLedgersMap(); - } catch (Exception e) { + } catch (IOException e) { log.error() .exception(e) - .log("Got IOException while trying to appendLedgersMap in cacheEntryRemoval callback"); + .log("Fatal entry log write failure while trying to appendLedgersMap " + + "in cacheEntryRemoval callback"); + for (LedgerDirsListener listener : ledgerDirsManager.getListeners()) { + listener.fatalError(); + } + return; } replicaOfCurrentLogChannels.remove(logChannel.getLogId()); rotatedLogChannels.add(logChannel); diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SyncThread.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SyncThread.java index a838b837ed6..d36c62b6020 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SyncThread.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SyncThread.java @@ -130,15 +130,18 @@ public Future requestFlush() { long startTime = System.nanoTime(); try { flush(); + } catch (EntryLogWriteException e) { + throw e; } catch (Throwable t) { log.error().exception(t).log("Exception flushing ledgers"); } finally { syncExecutorTime.addLatency(MathUtils.elapsedNanos(startTime), TimeUnit.NANOSECONDS); } + return null; }); } - private void flush() { + private void flush() throws EntryLogWriteException { Checkpoint checkpoint = checkpointSource.newCheckpoint(); try { ledgerStorage.flush(); @@ -149,7 +152,7 @@ private void flush() { } catch (EntryLogWriteException e) { log.error("Fatal entry log write failure while flushing ledgers", e); dirsListener.fatalError(); - return; + throw e; } catch (IOException e) { log.error().exception(e).log("Exception flushing ledgers"); return; diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieImplTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieImplTest.java index 057fc2f9f9b..2b173a51637 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieImplTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieImplTest.java @@ -21,6 +21,7 @@ package org.apache.bookkeeper.bookie; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.doReturn; @@ -169,6 +170,41 @@ int shutdown(int exitCode) { } } + @Test + public void testStartupEntryLogFlushFailureStopsBookieBeforeRunning() throws Exception { + ServerConfiguration conf = newServerConfiguration(); + conf.setLedgerStorageClass(FailOnFlushDbLedgerStorage.class.getName()); + conf.setJournalWriteData(false); + FailOnFlushDbLedgerStorage.resetFailure(); + + CountDownLatch shutdownLatch = new CountDownLatch(1); + TestBookieImpl.Resources resources = new TestBookieImpl.ResourceBuilder(conf).build(); + BookieImpl bookie = new TestBookieImpl(resources) { + @Override + int shutdown(int exitCode) { + int result = super.shutdown(exitCode); + if (exitCode == ExitCode.BOOKIE_EXCEPTION) { + shutdownLatch.countDown(); + } + return result; + } + }; + + try { + FailOnFlushDbLedgerStorage.injectFailureOnNextFlush(); + bookie.start(); + + assertTrue("Startup entry log flush failure should shut down the bookie", + shutdownLatch.await(10, TimeUnit.SECONDS)); + assertFalse("Bookie should not keep running after startup entry log flush failure", bookie.isRunning()); + } finally { + FailOnFlushDbLedgerStorage.resetFailure(); + if (bookie.isRunning()) { + bookie.shutdown(); + } + } + } + public void mockAddEntryReleased(int flag) throws Exception { final String metadataServiceUri = zkUtil.getMetadataServiceUri(); ServerConfiguration conf = TestBKConfiguration.newServerConfiguration(); diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/DefaultEntryLogTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/DefaultEntryLogTest.java index 3f14a33d53f..acd4c763b55 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/DefaultEntryLogTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/DefaultEntryLogTest.java @@ -59,11 +59,13 @@ import java.util.concurrent.CyclicBarrier; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLongArray; import java.util.concurrent.locks.Lock; import org.apache.bookkeeper.bookie.DefaultEntryLogger.BufferedLogChannel; +import org.apache.bookkeeper.bookie.LedgerDirsManager.LedgerDirsListener; import org.apache.bookkeeper.bookie.LedgerDirsManager.NoWritableLedgerDirException; import org.apache.bookkeeper.common.testing.annotations.FlakyTest; import org.apache.bookkeeper.conf.ServerConfiguration; @@ -1316,6 +1318,53 @@ public void testAppendLedgersMapOnCacheRemoval() throws Exception { assertEquals((entrySize + 4) * numOfEntries, ledgersMap.get(ledgerId), "Total size of entries"); } + @Test + public void testAppendLedgersMapFailureOnCacheRemovalTriggersFatalError() throws Exception { + int evictionPeriod = 1; + + ServerConfiguration conf = TestBKConfiguration.newServerConfiguration(); + conf.setEntryLogFilePreAllocationEnabled(false); + conf.setEntryLogPerLedgerEnabled(true); + conf.setLedgerDirNames(createAndGetLedgerDirs(1)); + conf.setEntrylogMapAccessExpiryTimeInSeconds(evictionPeriod); + LedgerDirsManager ledgerDirsManager = new LedgerDirsManager(conf, conf.getLedgerDirs(), + new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold())); + + CountDownLatch fatalLatch = new CountDownLatch(1); + ledgerDirsManager.addLedgerDirsListener(new LedgerDirsListener() { + @Override + public void fatalError() { + fatalLatch.countDown(); + } + }); + + DefaultEntryLogger entryLogger = new DefaultEntryLogger(conf, ledgerDirsManager); + EntryLogManagerForEntryLogPerLedger entryLogManager = (EntryLogManagerForEntryLogPerLedger) entryLogger + .getEntryLogManager(); + + long ledgerId = 0L; + File tmpFile = File.createTempFile("entrylog", "failed-eviction"); + tmpFile.deleteOnExit(); + FileChannel fileChannel = new RandomAccessFile(tmpFile, "rw").getChannel(); + BufferedLogChannel logChannel = new BufferedLogChannel(UnpooledByteBufAllocator.DEFAULT, fileChannel, 10, 10, + 0L, tmpFile, conf.getFlushIntervalInBytes()); + + try { + entryLogManager.setCurrentLogForLedgerAndAddToRotate(ledgerId, logChannel); + + fileChannel.close(); + Thread.sleep(evictionPeriod * 1000 + 100); + entryLogManager.doEntryLogMapCleanup(); + + assertTrue("Cache removal appendLedgersMap failure should trigger fatal error", + fatalLatch.await(10, TimeUnit.SECONDS)); + Assert.assertFalse("Failed log channel should not be added to rotated logs", + entryLogManager.getRotatedLogChannels().contains(logChannel)); + } finally { + logChannel.close(); + } + } + /** * test EntryLogManager.EntryLogManagerForEntryLogPerLedger doesn't removes * the ledger from its cache map if ledger's corresponding state is accessed diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/SyncThreadTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/SyncThreadTest.java index c5548d112b2..98857c616c3 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/SyncThreadTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/SyncThreadTest.java @@ -31,6 +31,7 @@ import java.util.EnumSet; import java.util.PrimitiveIterator.OfLong; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; @@ -273,7 +274,14 @@ public void flush() throws IOException { } }; final SyncThread t = new SyncThread(conf, listener, storage, checkpointSource, NullStatsLogger.INSTANCE); - t.requestFlush().get(10, TimeUnit.SECONDS); + Future flush = t.requestFlush(); + try { + flush.get(10, TimeUnit.SECONDS); + fail("Flush future should fail on entry log write failure"); + } catch (ExecutionException e) { + assertTrue("Flush future should fail with entry log write failure", + e.getCause() instanceof EntryLogWriteException); + } assertTrue("Should have called fatal error", fatalLatch.await(10, TimeUnit.SECONDS)); t.shutdown(); } diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/FailOnFlushDbLedgerStorage.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/FailOnFlushDbLedgerStorage.java index cc6286de042..19d97fe8da2 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/FailOnFlushDbLedgerStorage.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/FailOnFlushDbLedgerStorage.java @@ -32,6 +32,7 @@ public class FailOnFlushDbLedgerStorage extends DbLedgerStorage { private static final AtomicBoolean failNextFlushWithEntryLogWriteException = new AtomicBoolean(false); + private static final AtomicBoolean entryLogFlushFailed = new AtomicBoolean(false); public static void injectFailureOnNextFlush() { failNextFlushWithEntryLogWriteException.set(true); @@ -39,6 +40,7 @@ public static void injectFailureOnNextFlush() { public static void resetFailure() { failNextFlushWithEntryLogWriteException.set(false); + entryLogFlushFailed.set(false); } @Override @@ -64,7 +66,8 @@ private static class FailOnFlushSingleDirectoryDbLedgerStorage extends SingleDir @Override public void flush() throws IOException { - if (failNextFlushWithEntryLogWriteException.compareAndSet(true, false)) { + if (entryLogFlushFailed.get() || failNextFlushWithEntryLogWriteException.compareAndSet(true, false)) { + entryLogFlushFailed.set(true); throw new EntryLogWriteException("entry log flush failed", new IOException("injected")); } super.flush(); From 5e3674b7a50e964221a1c59a27cfba1b755cfb42 Mon Sep 17 00:00:00 2001 From: yangxianjungree <714696209@qq.com> Date: Tue, 4 Aug 2026 20:11:29 +0800 Subject: [PATCH 5/9] fix: keep fatal entrylog notifications compatible --- .../bookkeeper/bookie/DefaultEntryLogger.java | 5 ++ .../bookkeeper/bookie/EntryLogManager.java | 6 ++ .../bookie/EntryLogManagerBase.java | 18 +++++ .../EntryLogManagerForEntryLogPerLedger.java | 12 +-- .../ldb/SingleDirectoryDbLedgerStorage.java | 4 + .../storage/ldb/DbLedgerStorageTest.java | 81 +++++++++++++++++++ 6 files changed, 118 insertions(+), 8 deletions(-) diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/DefaultEntryLogger.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/DefaultEntryLogger.java index a9cfb3bfee3..85001db4ad4 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/DefaultEntryLogger.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/DefaultEntryLogger.java @@ -58,6 +58,7 @@ import java.util.concurrent.CopyOnWriteArrayList; 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; @@ -382,6 +383,10 @@ EntryLogManager getEntryLogManager() { return entryLogManager; } + public void setFatalErrorListener(LedgerDirsListener fatalErrorListener) { + entryLogManager.setFatalErrorListener(fatalErrorListener); + } + void addListener(EntryLogListener listener) { if (null != listener) { listeners.add(listener); diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManager.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManager.java index 7364fb08e56..763243fbbf0 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManager.java +++ b/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. diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerBase.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerBase.java index de2cd5d34ce..e688cfa2b56 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerBase.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerBase.java @@ -33,6 +33,7 @@ import lombok.CustomLog; import org.apache.bookkeeper.bookie.DefaultEntryLogger.BufferedLogChannel; import org.apache.bookkeeper.bookie.DefaultEntryLogger.EntryLogListener; +import org.apache.bookkeeper.bookie.LedgerDirsManager.LedgerDirsListener; import org.apache.bookkeeper.bookie.LedgerDirsManager.NoWritableLedgerDirException; import org.apache.bookkeeper.conf.ServerConfiguration; @@ -42,6 +43,8 @@ abstract class EntryLogManagerBase implements EntryLogManager { final EntryLoggerAllocator entryLoggerAllocator; final LedgerDirsManager ledgerDirsManager; private final List listeners; + private static final LedgerDirsListener NOOP_FATAL_ERROR_LISTENER = new LedgerDirsListener() { }; + private volatile LedgerDirsListener fatalErrorListener = NOOP_FATAL_ERROR_LISTENER; /** * The maximum size of a entry logger file. */ @@ -124,6 +127,21 @@ List getRotatedLogChannels() { return rotatedLogChannels; } + @Override + public void setFatalErrorListener(LedgerDirsListener fatalErrorListener) { + this.fatalErrorListener = fatalErrorListener != null ? fatalErrorListener : NOOP_FATAL_ERROR_LISTENER; + } + + void notifyFatalEntryLogWriteFailure(String message, Throwable cause) { + log.error(message, cause); + fatalErrorListener.fatalError(); + for (LedgerDirsListener listener : ledgerDirsManager.getListeners()) { + if (listener != fatalErrorListener) { + listener.fatalError(); + } + } + } + @Override public void flush() throws IOException { flushCurrentLogs(); diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerForEntryLogPerLedger.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerForEntryLogPerLedger.java index 85ee8edc5a9..7789ffcebe1 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerForEntryLogPerLedger.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerForEntryLogPerLedger.java @@ -359,14 +359,10 @@ private void onCacheEntryRemoval(RemovalNotification // Append ledgers map at the end of entry log try { logChannel.appendLedgersMap(); - } catch (IOException e) { - log.error() - .exception(e) - .log("Fatal entry log write failure while trying to appendLedgersMap " - + "in cacheEntryRemoval callback"); - for (LedgerDirsListener listener : ledgerDirsManager.getListeners()) { - listener.fatalError(); - } + } catch (IOException | RuntimeException e) { + notifyFatalEntryLogWriteFailure( + "Fatal entry log write failure while trying to appendLedgersMap in cacheEntryRemoval callback", + e); return; } replicaOfCurrentLogChannels.remove(logChannel.getLogId()); diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java index 15b71e2c501..e95c7c83408 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java @@ -57,6 +57,7 @@ import org.apache.bookkeeper.bookie.CheckpointSource.Checkpoint; import org.apache.bookkeeper.bookie.Checkpointer; import org.apache.bookkeeper.bookie.CompactableLedgerStorage; +import org.apache.bookkeeper.bookie.DefaultEntryLogger; import org.apache.bookkeeper.bookie.EntryLogWriteException; import org.apache.bookkeeper.bookie.EntryLocation; import org.apache.bookkeeper.bookie.GarbageCollectionStatus; @@ -257,6 +258,9 @@ public void setCheckpointer(Checkpointer checkpointer) { } void setFatalErrorListener(LedgerDirsListener fatalErrorListener) { if (fatalErrorListener != null) { this.fatalErrorListener = fatalErrorListener; + if (entryLogger instanceof DefaultEntryLogger) { + ((DefaultEntryLogger) entryLogger).setFatalErrorListener(fatalErrorListener); + } } } diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageTest.java index a569a638ccf..3bb992888ab 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageTest.java @@ -35,7 +35,9 @@ import java.io.FileOutputStream; import java.io.IOException; import java.lang.reflect.Field; +import java.lang.reflect.Method; import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; import java.util.List; import java.util.Set; import java.util.concurrent.CountDownLatch; @@ -46,6 +48,8 @@ import org.apache.bookkeeper.bookie.Bookie; import org.apache.bookkeeper.bookie.Bookie.NoEntryException; import org.apache.bookkeeper.bookie.BookieException; +import org.apache.bookkeeper.bookie.BufferedChannel; +import org.apache.bookkeeper.bookie.BufferedChannelBase; import org.apache.bookkeeper.bookie.BookieImpl; import org.apache.bookkeeper.bookie.CheckpointSource; import org.apache.bookkeeper.bookie.CheckpointSourceList; @@ -53,6 +57,7 @@ import org.apache.bookkeeper.bookie.EntryLocation; import org.apache.bookkeeper.bookie.Journal; import org.apache.bookkeeper.bookie.LedgerDirsManager; +import org.apache.bookkeeper.bookie.LedgerDirsManager.LedgerDirsListener; import org.apache.bookkeeper.bookie.LedgerDirsManager.NoWritableLedgerDirException; import org.apache.bookkeeper.bookie.LedgerStorage; import org.apache.bookkeeper.bookie.LogMark; @@ -62,6 +67,7 @@ import org.apache.bookkeeper.conf.TestBKConfiguration; import org.apache.bookkeeper.proto.BookieProtocol; import org.apache.bookkeeper.util.DiskChecker; +import org.apache.commons.io.FileUtils; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -264,6 +270,62 @@ public void testBookieCompaction() throws Exception { assertEquals(newEntry3, res); } + @Test + public void testPerLedgerEvictionFailurePropagatesToDbFatalErrorListener() throws Exception { + File perLedgerDir = File.createTempFile("bkTestPerLedger", ".dir"); + perLedgerDir.delete(); + perLedgerDir.mkdir(); + File curDir = BookieImpl.getCurrentDirectory(perLedgerDir); + BookieImpl.checkDirectoryStructure(curDir); + + ServerConfiguration conf = TestBKConfiguration.newServerConfiguration(); + conf.setGcWaitTime(1000); + conf.setLedgerStorageClass(DbLedgerStorage.class.getName()); + conf.setLedgerDirNames(new String[] { perLedgerDir.toString() }); + conf.setEntryLogFilePreAllocationEnabled(false); + conf.setEntryLogPerLedgerEnabled(true); + conf.setEntrylogMapAccessExpiryTimeInSeconds(1); + + BookieImpl bookie = new TestBookieImpl(conf); + DbLedgerStorage storage = (DbLedgerStorage) bookie.getLedgerStorage(); + CountDownLatch fatalLatch = new CountDownLatch(1); + storage.setFatalErrorListener(new LedgerDirsListener() { + @Override + public void fatalError() { + fatalLatch.countDown(); + } + }); + + try { + SingleDirectoryDbLedgerStorage singleDirStorage = storage.getLedgerStorageList().get(0); + DefaultEntryLogger entryLogger = (DefaultEntryLogger) singleDirStorage.getEntryLogger(); + long ledgerId = 4L; + ByteBuf entry = Unpooled.buffer(1024); + try { + entry.writeLong(ledgerId); + entry.writeLong(1L); + entry.writeBytes("entry-1".getBytes()); + entryLogger.addEntry(ledgerId, entry); + } finally { + ReferenceCountUtil.release(entry); + } + + Object entryLogManager = getEntryLogManager(entryLogger); + BufferedChannel currentLogChannel = (BufferedChannel) invoke(entryLogManager, + "getCurrentLogForLedger", new Class[] { long.class }, ledgerId); + closeUnderlyingFileChannel(currentLogChannel); + + Thread.sleep(TimeUnit.SECONDS.toMillis(2)); + invoke(entryLogManager, "doEntryLogMapCleanup", new Class[] { }); + + assertTrue("Per-ledger eviction failure should propagate through DbLedgerStorage fatal listener", + fatalLatch.await(10, TimeUnit.SECONDS)); + } finally { + bookie.shutdown(); + FileUtils.deleteDirectory(perLedgerDir); + } + } + @Test public void doubleDirectory() throws Exception { int gcWaitTime = 1000; @@ -283,6 +345,25 @@ public void doubleDirectory() throws Exception { bookie.shutdown(); } + private Object getEntryLogManager(DefaultEntryLogger entryLogger) throws Exception { + Method method = DefaultEntryLogger.class.getDeclaredMethod("getEntryLogManager"); + method.setAccessible(true); + return method.invoke(entryLogger); + } + + private void closeUnderlyingFileChannel(BufferedChannel channel) throws Exception { + Field fileChannelField = BufferedChannelBase.class.getDeclaredField("fileChannel"); + fileChannelField.setAccessible(true); + ((FileChannel) fileChannelField.get(channel)).close(); + } + + private Object invoke(Object target, String methodName, Class[] parameterTypes, Object... args) + throws Exception { + Method method = target.getClass().getDeclaredMethod(methodName, parameterTypes); + method.setAccessible(true); + return method.invoke(target, args); + } + @Test public void testRewritingEntries() throws Exception { storage.setMasterKey(1, "key".getBytes()); From d761317ef92c093c54844ce220cf4e611aae2aed Mon Sep 17 00:00:00 2001 From: yangxianjungree <714696209@qq.com> Date: Wed, 5 Aug 2026 09:42:09 +0800 Subject: [PATCH 6/9] fix: adapt fatal entrylog logging to master --- .../org/apache/bookkeeper/bookie/EntryLogManagerBase.java | 2 +- .../org/apache/bookkeeper/bookie/SortedLedgerStorage.java | 2 +- .../main/java/org/apache/bookkeeper/bookie/SyncThread.java | 4 ++-- .../bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java | 2 +- 4 files changed, 5 insertions(+), 5 deletions(-) diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerBase.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerBase.java index e688cfa2b56..366267f0ba7 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerBase.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerBase.java @@ -133,7 +133,7 @@ public void setFatalErrorListener(LedgerDirsListener fatalErrorListener) { } void notifyFatalEntryLogWriteFailure(String message, Throwable cause) { - log.error(message, cause); + log.error().exception(cause).log(message); fatalErrorListener.fatalError(); for (LedgerDirsListener listener : ledgerDirsManager.getListeners()) { if (listener != fatalErrorListener) { diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SortedLedgerStorage.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SortedLedgerStorage.java index 8da0bb7f6a1..d6045cc3fa1 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SortedLedgerStorage.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SortedLedgerStorage.java @@ -340,7 +340,7 @@ public void onRotateEntryLog() { } private void fatalEntryLogWriteFailure(EntryLogWriteException e) { - LOG.error("Fatal entry log write failure while flushing skip list cache.", e); + log.error().exception(e).log("Fatal entry log write failure while flushing skip list cache."); if (stateManager instanceof BookieStateManager) { StateManager.ShutdownHandler shutdownHandler = ((BookieStateManager) stateManager).getShutdownHandler(); if (shutdownHandler != null) { diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SyncThread.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SyncThread.java index d36c62b6020..1f978360358 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SyncThread.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SyncThread.java @@ -150,7 +150,7 @@ private void flush() throws EntryLogWriteException { dirsListener.allDisksFull(true); return; } catch (EntryLogWriteException e) { - log.error("Fatal entry log write failure while flushing ledgers", e); + log.error().exception(e).log("Fatal entry log write failure while flushing ledgers"); dirsListener.fatalError(); throw e; } catch (IOException e) { @@ -185,7 +185,7 @@ public void checkpoint(Checkpoint checkpoint) { dirsListener.allDisksFull(true); return; } catch (EntryLogWriteException e) { - log.error("Fatal entry log write failure while checkpointing ledgers", e); + log.error().exception(e).log("Fatal entry log write failure while checkpointing ledgers"); dirsListener.fatalError(); return; } catch (IOException e) { diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java index e95c7c83408..994459b054a 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java @@ -579,7 +579,7 @@ private void triggerFlushAndAddEntry(long ledgerId, long entryId, ByteBuf entry) } private void notifyFatalEntryLogWriteFailure(EntryLogWriteException e) { - log.error("Fatal entry log write failure during background flush", e); + log.error().exception(e).log("Fatal entry log write failure during background flush"); fatalErrorListener.fatalError(); } From 8988984d483f4204936ec1514d785fb52c179538 Mon Sep 17 00:00:00 2001 From: yangxianjungree <714696209@qq.com> Date: Wed, 5 Aug 2026 09:46:53 +0800 Subject: [PATCH 7/9] test: release entrylog test buffers --- .../bookie/DefaultEntryLogTest.java | 569 ++++++++++-------- .../storage/ldb/DbLedgerStorageTest.java | 61 +- 2 files changed, 373 insertions(+), 257 deletions(-) diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/DefaultEntryLogTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/DefaultEntryLogTest.java index acd4c763b55..b07f6dd4d88 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/DefaultEntryLogTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/DefaultEntryLogTest.java @@ -167,16 +167,20 @@ public void testEntryLogIsSealedWithPerLedgerDisabled() throws Exception { statsProvider.getStatsLogger(BookKeeperServerStats.ENTRYLOGGER_SCOPE); DefaultEntryLogger entryLogger = new DefaultEntryLogger(conf, dirsMgr, null, statsLogger, UnpooledByteBufAllocator.DEFAULT); - EntryLogManagerBase entrylogManager = (EntryLogManagerBase) entryLogger.getEntryLogManager(); - entrylogManager.createNewLog(0); - BufferedReadChannel channel = entryLogger.getChannelForLogId(0); - assertFalse(channel.sealed); - entrylogManager.createNewLog(1); - channel = entryLogger.getChannelForLogId(0); - assertFalse(channel.sealed); - entrylogManager.createNewLog(2); - channel = entryLogger.getChannelForLogId(1); - assertTrue(channel.sealed); + try { + EntryLogManagerBase entrylogManager = (EntryLogManagerBase) entryLogger.getEntryLogManager(); + entrylogManager.createNewLog(0); + BufferedReadChannel channel = entryLogger.getChannelForLogId(0); + assertFalse(channel.sealed); + entrylogManager.createNewLog(1); + channel = entryLogger.getChannelForLogId(0); + assertFalse(channel.sealed); + entrylogManager.createNewLog(2); + channel = entryLogger.getChannelForLogId(1); + assertTrue(channel.sealed); + } finally { + entryLogger.close(); + } } @Test @@ -191,16 +195,20 @@ public void testEntryLogIsSealedWithPerLedgerEnabled() throws Exception { statsProvider.getStatsLogger(BookKeeperServerStats.ENTRYLOGGER_SCOPE); DefaultEntryLogger entryLogger = new DefaultEntryLogger(conf, dirsMgr, null, statsLogger, UnpooledByteBufAllocator.DEFAULT); - EntryLogManagerBase entrylogManager = (EntryLogManagerBase) entryLogger.getEntryLogManager(); - entrylogManager.createNewLog(0); - BufferedReadChannel channel = entryLogger.getChannelForLogId(0); - assertFalse(channel.sealed); - entrylogManager.createNewLog(1); - channel = entryLogger.getChannelForLogId(0); - assertFalse(channel.sealed); - entrylogManager.createNewLog(2); - channel = entryLogger.getChannelForLogId(1); - assertFalse(channel.sealed); + try { + EntryLogManagerBase entrylogManager = (EntryLogManagerBase) entryLogger.getEntryLogManager(); + entrylogManager.createNewLog(0); + BufferedReadChannel channel = entryLogger.getChannelForLogId(0); + assertFalse(channel.sealed); + entrylogManager.createNewLog(1); + channel = entryLogger.getChannelForLogId(0); + assertFalse(channel.sealed); + entrylogManager.createNewLog(2); + channel = entryLogger.getChannelForLogId(1); + assertFalse(channel.sealed); + } finally { + entryLogger.close(); + } } @Test @@ -276,6 +284,24 @@ private ByteBuf generateEntry(long ledger, long entry, int length) { return bb; } + private static long addEntryAndRelease(LedgerStorage ledgerStorage, ByteBuf entry) + throws IOException, BookieException { + try { + return ledgerStorage.addEntry(entry); + } finally { + ReferenceCountUtil.release(entry); + } + } + + private static void assertEntryCompareEqualsAndRelease(ByteBuf expected, ByteBuf actual) { + try { + assertEquals(0, expected.compareTo(actual)); + } finally { + ReferenceCountUtil.release(actual); + ReferenceCountUtil.release(expected); + } + } + private static String generateDataString(long ledger, long entry) { return ("ledger-" + ledger + "-" + entry); } @@ -348,24 +374,29 @@ public void testAddEntryFailureOnDiskFull() throws Exception { BookieImpl bookie = new TestBookieImpl(conf); DefaultEntryLogger entryLogger = new DefaultEntryLogger(conf, bookie.getLedgerDirsManager()); - InterleavedLedgerStorage ledgerStorage = - ((InterleavedLedgerStorage) bookie.ledgerStorage.getUnderlyingLedgerStorage()); - ledgerStorage.entryLogger = entryLogger; - // Create ledgers - ledgerStorage.setMasterKey(1, "key".getBytes()); - ledgerStorage.setMasterKey(2, "key".getBytes()); - ledgerStorage.setMasterKey(3, "key".getBytes()); - // Add entries - ledgerStorage.addEntry(generateEntry(1, 1)); - ledgerStorage.addEntry(generateEntry(2, 1)); - // Add entry with disk full failure simulation - bookie.getLedgerDirsManager().addToFilledDirs(((EntryLogManagerBase) entryLogger.getEntryLogManager()) - .getCurrentLogForLedger(DefaultEntryLogger.UNASSIGNED_LEDGERID).getLogFile().getParentFile()); - ledgerStorage.addEntry(generateEntry(3, 1)); - // Verify written entries - assertEquals(0, generateEntry(1, 1).compareTo(ledgerStorage.getEntry(1, 1))); - assertEquals(0, generateEntry(2, 1).compareTo(ledgerStorage.getEntry(2, 1))); - assertEquals(0, generateEntry(3, 1).compareTo(ledgerStorage.getEntry(3, 1))); + try { + InterleavedLedgerStorage ledgerStorage = + ((InterleavedLedgerStorage) bookie.ledgerStorage.getUnderlyingLedgerStorage()); + ledgerStorage.entryLogger = entryLogger; + // Create ledgers + ledgerStorage.setMasterKey(1, "key".getBytes()); + ledgerStorage.setMasterKey(2, "key".getBytes()); + ledgerStorage.setMasterKey(3, "key".getBytes()); + // Add entries + addEntryAndRelease(ledgerStorage, generateEntry(1, 1)); + addEntryAndRelease(ledgerStorage, generateEntry(2, 1)); + // Add entry with disk full failure simulation + bookie.getLedgerDirsManager().addToFilledDirs(((EntryLogManagerBase) entryLogger.getEntryLogManager()) + .getCurrentLogForLedger(DefaultEntryLogger.UNASSIGNED_LEDGERID).getLogFile().getParentFile()); + addEntryAndRelease(ledgerStorage, generateEntry(3, 1)); + // Verify written entries + assertEntryCompareEqualsAndRelease(generateEntry(1, 1), ledgerStorage.getEntry(1, 1)); + assertEntryCompareEqualsAndRelease(generateEntry(2, 1), ledgerStorage.getEntry(2, 1)); + assertEntryCompareEqualsAndRelease(generateEntry(3, 1), ledgerStorage.getEntry(3, 1)); + } finally { + entryLogger.close(); + bookie.shutdown(); + } } /** @@ -702,9 +733,11 @@ static class LedgerStorageReadTask implements Callable { @Override public Boolean call() throws IOException, BookieException { + ByteBuf expectedByteBuf = null; + ByteBuf actualByteBuf = null; try { - ByteBuf expectedByteBuf = generateEntry(ledgerId, entryId); - ByteBuf actualByteBuf = ledgerStorage.getEntry(ledgerId, entryId); + expectedByteBuf = generateEntry(ledgerId, entryId); + actualByteBuf = ledgerStorage.getEntry(ledgerId, entryId); if (!expectedByteBuf.equals(actualByteBuf)) { LOG.error("Expected Entry: {} Actual Entry: {}", expectedByteBuf.toString(Charset.defaultCharset()), actualByteBuf.toString(Charset.defaultCharset())); @@ -715,6 +748,9 @@ public Boolean call() throws IOException, BookieException { LOG.error("Got Exception for GetEntry call. LedgerId: " + ledgerId + " entryId: " + entryId, e); throw new IOException("Got Exception for GetEntry call. LedgerId: " + ledgerId + " entryId: " + entryId, e); + } finally { + ReferenceCountUtil.release(actualByteBuf); + ReferenceCountUtil.release(expectedByteBuf); } return true; } @@ -951,16 +987,24 @@ public void testFlushIntervalInBytes() throws Exception { "EntryLogManager class type"); ByteBuf buf = newEntryLogger.readEntry(ledgerId, 0L, entry0Position); - long readLedgerId = buf.readLong(); - long readEntryId = buf.readLong(); - assertEquals(ledgerId, readLedgerId, "LedgerId"); - assertEquals(0L, readEntryId, "EntryId"); + try { + long readLedgerId = buf.readLong(); + long readEntryId = buf.readLong(); + assertEquals(ledgerId, readLedgerId, "LedgerId"); + assertEquals(0L, readEntryId, "EntryId"); + } finally { + ReferenceCountUtil.release(buf); + } buf = newEntryLogger.readEntry(ledgerId, 1L, entry1Position); - readLedgerId = buf.readLong(); - readEntryId = buf.readLong(); - assertEquals(ledgerId, readLedgerId, "LedgerId"); - assertEquals(1L, readEntryId, "EntryId"); + try { + long readLedgerId = buf.readLong(); + long readEntryId = buf.readLong(); + assertEquals(ledgerId, readLedgerId, "LedgerId"); + assertEquals(1L, readEntryId, "EntryId"); + } finally { + ReferenceCountUtil.release(buf); + } } @Test @@ -969,19 +1013,28 @@ public void testReadEntryWithoutLedgerID() throws Exception { // `+ 1` is not a typo: create one more log file than the max number of o cached readers for (int i = 0; i < 10; i++) { ByteBuf e = makeEntry(1L, i, 100); - long loc = entryLogger.addEntry(1L, e.slice()); - locations.add(loc); + try { + long loc = entryLogger.addEntry(1L, e.slice()); + locations.add(loc); + } finally { + ReferenceCountUtil.release(e); + } } entryLogger.flush(); for (Long loc : locations) { int i = locations.indexOf(loc); ByteBuf data = entryLogger.readEntry(loc); - assertEntryEquals(data, makeEntry(1L, i, 100)); - long readLedgerId = data.readLong(); - long readEntryId = data.readLong(); - assertEquals(1L, readLedgerId, "LedgerId"); - assertEquals(i, readEntryId, "EntryId"); - ReferenceCountUtil.release(data); + ByteBuf expected = makeEntry(1L, i, 100); + try { + assertEntryEquals(data, expected); + long readLedgerId = data.readLong(); + long readEntryId = data.readLong(); + assertEquals(1L, readLedgerId, "LedgerId"); + assertEquals(i, readEntryId, "EntryId"); + } finally { + ReferenceCountUtil.release(data); + ReferenceCountUtil.release(expected); + } } } @@ -1000,69 +1053,74 @@ public void testEntryLogManagerInterfaceForEntryLogPerLedger() throws Exception new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold())); DefaultEntryLogger entryLogger = new DefaultEntryLogger(conf, ledgerDirsManager); - EntryLogManagerForEntryLogPerLedger entryLogManager = (EntryLogManagerForEntryLogPerLedger) entryLogger - .getEntryLogManager(); + try { + EntryLogManagerForEntryLogPerLedger entryLogManager = (EntryLogManagerForEntryLogPerLedger) entryLogger + .getEntryLogManager(); - assertEquals(0, entryLogManager.getCopyOfCurrentLogs().size(), "Number of current active EntryLogs "); - assertEquals(0, entryLogManager.getRotatedLogChannels().size(), "Number of Rotated Logs "); + assertEquals(0, entryLogManager.getCopyOfCurrentLogs().size(), "Number of current active EntryLogs "); + assertEquals(0, entryLogManager.getRotatedLogChannels().size(), "Number of Rotated Logs "); - int numOfLedgers = 5; - int numOfThreadsPerLedger = 10; - validateLockAcquireAndRelease(numOfLedgers, numOfThreadsPerLedger, entryLogManager); + int numOfLedgers = 5; + int numOfThreadsPerLedger = 10; + validateLockAcquireAndRelease(numOfLedgers, numOfThreadsPerLedger, entryLogManager); - for (long i = 0; i < numOfLedgers; i++) { - entryLogManager.setCurrentLogForLedgerAndAddToRotate(i, - createDummyBufferedLogChannel(i, conf)); - } + for (long i = 0; i < numOfLedgers; i++) { + entryLogManager.setCurrentLogForLedgerAndAddToRotate(i, + createDummyBufferedLogChannel(i, conf)); + } - for (long i = 0; i < numOfLedgers; i++) { - assertEquals(entryLogManager.getCurrentLogIfPresent(i), entryLogManager.getCurrentLogForLedger(i), - "LogChannel for ledger: " + i); - } + for (long i = 0; i < numOfLedgers; i++) { + assertEquals(entryLogManager.getCurrentLogIfPresent(i), entryLogManager.getCurrentLogForLedger(i), + "LogChannel for ledger: " + i); + } - assertEquals(numOfLedgers, entryLogManager.getCopyOfCurrentLogs().size(), - "Number of current active EntryLogs "); - assertEquals(0, entryLogManager.getRotatedLogChannels().size(), "Number of Rotated Logs "); + assertEquals(numOfLedgers, entryLogManager.getCopyOfCurrentLogs().size(), + "Number of current active EntryLogs "); + assertEquals(0, entryLogManager.getRotatedLogChannels().size(), "Number of Rotated Logs "); - for (long i = 0; i < numOfLedgers; i++) { - entryLogManager.setCurrentLogForLedgerAndAddToRotate(i, - createDummyBufferedLogChannel(numOfLedgers + i, conf)); - } + for (long i = 0; i < numOfLedgers; i++) { + entryLogManager.setCurrentLogForLedgerAndAddToRotate(i, + createDummyBufferedLogChannel(numOfLedgers + i, conf)); + } - /* - * since new entryLogs are set for all the ledgers, previous entrylogs would be added to rotatedLogChannels - */ - assertEquals(numOfLedgers, entryLogManager.getCopyOfCurrentLogs().size(), - "Number of current active EntryLogs "); - assertEquals(numOfLedgers, entryLogManager.getRotatedLogChannels().size(), "Number of Rotated Logs "); + /* + * since new entryLogs are set for all the ledgers, previous entrylogs would be added to rotatedLogChannels + */ + assertEquals(numOfLedgers, entryLogManager.getCopyOfCurrentLogs().size(), + "Number of current active EntryLogs "); + assertEquals(numOfLedgers, entryLogManager.getRotatedLogChannels().size(), "Number of Rotated Logs "); - for (long i = 0; i < numOfLedgers; i++) { - entryLogManager.setCurrentLogForLedgerAndAddToRotate(i, - createDummyBufferedLogChannel(2 * numOfLedgers + i, conf)); - } + for (long i = 0; i < numOfLedgers; i++) { + entryLogManager.setCurrentLogForLedgerAndAddToRotate(i, + createDummyBufferedLogChannel(2 * numOfLedgers + i, conf)); + } - /* - * again since new entryLogs are set for all the ledgers, previous entrylogs would be added to - * rotatedLogChannels - */ - assertEquals(numOfLedgers, entryLogManager.getCopyOfCurrentLogs().size(), - "Number of current active EntryLogs "); - assertEquals(2 * numOfLedgers, entryLogManager.getRotatedLogChannels().size(), "Number of Rotated Logs "); + /* + * again since new entryLogs are set for all the ledgers, previous entrylogs would be added to + * rotatedLogChannels + */ + assertEquals(numOfLedgers, entryLogManager.getCopyOfCurrentLogs().size(), + "Number of current active EntryLogs "); + assertEquals(2 * numOfLedgers, entryLogManager.getRotatedLogChannels().size(), "Number of Rotated Logs "); - for (BufferedLogChannel logChannel : entryLogManager.getRotatedLogChannels()) { - entryLogManager.getRotatedLogChannels().remove(logChannel); - } - assertEquals(0, entryLogManager.getRotatedLogChannels().size(), "Number of Rotated Logs "); + for (BufferedLogChannel logChannel : new ArrayList<>(entryLogManager.getRotatedLogChannels())) { + entryLogManager.getRotatedLogChannels().remove(logChannel); + logChannel.close(); + } + assertEquals(0, entryLogManager.getRotatedLogChannels().size(), "Number of Rotated Logs "); - // entrylogid is sequential - for (long i = 0; i < numOfLedgers; i++) { - assertEquals(2 * numOfLedgers + i, entryLogManager.getCurrentLogForLedger(i).getLogId(), - "EntryLogId for Ledger " + i); - } + // entrylogid is sequential + for (long i = 0; i < numOfLedgers; i++) { + assertEquals(2 * numOfLedgers + i, entryLogManager.getCurrentLogForLedger(i).getLogId(), + "EntryLogId for Ledger " + i); + } - for (long i = 2 * numOfLedgers; i < (3 * numOfLedgers); i++) { - assertNotNull(entryLogManager.getCurrentLogIfPresent(i), - "EntryLog with logId: " + i + " should be present"); + for (long i = 2 * numOfLedgers; i < (3 * numOfLedgers); i++) { + assertNotNull(entryLogManager.getCurrentLogIfPresent(i), + "EntryLog with logId: " + i + " should be present"); + } + } finally { + entryLogger.close(); } } @@ -1151,34 +1209,38 @@ public void testEntryLogManagerExpiryRemoval() throws Exception { new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold())); DefaultEntryLogger entryLogger = new DefaultEntryLogger(conf, ledgerDirsManager); - EntryLogManagerForEntryLogPerLedger entryLogManager = - (EntryLogManagerForEntryLogPerLedger) entryLogger.getEntryLogManager(); + try { + EntryLogManagerForEntryLogPerLedger entryLogManager = + (EntryLogManagerForEntryLogPerLedger) entryLogger.getEntryLogManager(); - long ledgerId = 0L; + long ledgerId = 0L; - BufferedLogChannel logChannel = createDummyBufferedLogChannel(0, conf); - entryLogManager.setCurrentLogForLedgerAndAddToRotate(ledgerId, logChannel); + BufferedLogChannel logChannel = createDummyBufferedLogChannel(0, conf); + entryLogManager.setCurrentLogForLedgerAndAddToRotate(ledgerId, logChannel); - BufferedLogChannel currentLogForLedger = entryLogManager.getCurrentLogForLedger(ledgerId); - assertEquals(logChannel, currentLogForLedger, "LogChannel for ledger " + ledgerId + " should match"); + BufferedLogChannel currentLogForLedger = entryLogManager.getCurrentLogForLedger(ledgerId); + assertEquals(logChannel, currentLogForLedger, "LogChannel for ledger " + ledgerId + " should match"); - Thread.sleep(evictionPeriod * 1000 + 100); - entryLogManager.doEntryLogMapCleanup(); + Thread.sleep(evictionPeriod * 1000 + 100); + entryLogManager.doEntryLogMapCleanup(); - /* - * since for more than evictionPeriod, that ledger is not accessed and cache is cleaned up, mapping for that - * ledger should not be available anymore - */ - currentLogForLedger = entryLogManager.getCurrentLogForLedger(ledgerId); - assertNull(currentLogForLedger, "LogChannel for ledger " + ledgerId + " should be null"); - assertEquals(0, entryLogManager.getCopyOfCurrentLogs().size(), "Number of current active EntryLogs "); - assertEquals(1, entryLogManager.getRotatedLogChannels().size(), "Number of rotated EntryLogs "); - assertTrue(entryLogManager.getRotatedLogChannels().contains(logChannel), - "CopyOfRotatedLogChannels should contain the created LogChannel"); - - assertTrue((entryLogManager.getCacheAsMap().get(ledgerId) == null) - || (entryLogManager.getCacheAsMap().get(ledgerId).getEntryLogWithDirInfo() == null), - "since map entry must have been evicted, it should be null"); + /* + * since for more than evictionPeriod, that ledger is not accessed and cache is cleaned up, mapping for that + * ledger should not be available anymore + */ + currentLogForLedger = entryLogManager.getCurrentLogForLedger(ledgerId); + assertNull(currentLogForLedger, "LogChannel for ledger " + ledgerId + " should be null"); + assertEquals(0, entryLogManager.getCopyOfCurrentLogs().size(), "Number of current active EntryLogs "); + assertEquals(1, entryLogManager.getRotatedLogChannels().size(), "Number of rotated EntryLogs "); + assertTrue(entryLogManager.getRotatedLogChannels().contains(logChannel), + "CopyOfRotatedLogChannels should contain the created LogChannel"); + + assertTrue((entryLogManager.getCacheAsMap().get(ledgerId) == null) + || (entryLogManager.getCacheAsMap().get(ledgerId).getEntryLogWithDirInfo() == null), + "since map entry must have been evicted, it should be null"); + } finally { + entryLogger.close(); + } } /* @@ -1246,13 +1308,17 @@ public void testLongLedgerIdsWithEntryLogPerLedger() throws Exception { for (int entryId = 0; entryId < numOfEntries; entryId++) { String expectedValue = generateDataString(ledgerId, entryId); ByteBuf buf = entryLogger.readEntry(ledgerId, entryId, pos[i][entryId]); - long readLedgerId = buf.readLong(); - long readEntryId = buf.readLong(); - byte[] readData = new byte[buf.readableBytes()]; - buf.readBytes(readData); - assertEquals(ledgerId, readLedgerId, "LedgerId "); - assertEquals(entryId, readEntryId, "EntryId "); - assertEquals(expectedValue, new String(readData), "Entry Data "); + try { + long readLedgerId = buf.readLong(); + long readEntryId = buf.readLong(); + byte[] readData = new byte[buf.readableBytes()]; + buf.readBytes(readData); + assertEquals(ledgerId, readLedgerId, "LedgerId "); + assertEquals(entryId, readEntryId, "EntryId "); + assertEquals(expectedValue, new String(readData), "Entry Data "); + } finally { + ReferenceCountUtil.release(buf); + } } } } @@ -1356,10 +1422,10 @@ public void fatalError() { Thread.sleep(evictionPeriod * 1000 + 100); entryLogManager.doEntryLogMapCleanup(); - assertTrue("Cache removal appendLedgersMap failure should trigger fatal error", - fatalLatch.await(10, TimeUnit.SECONDS)); - Assert.assertFalse("Failed log channel should not be added to rotated logs", - entryLogManager.getRotatedLogChannels().contains(logChannel)); + assertTrue(fatalLatch.await(10, TimeUnit.SECONDS), + "Cache removal appendLedgersMap failure should trigger fatal error"); + assertFalse(entryLogManager.getRotatedLogChannels().contains(logChannel), + "Failed log channel should not be added to rotated logs"); } finally { logChannel.close(); } @@ -1384,36 +1450,41 @@ public void testExpiryRemovalByAccessingOnAnotherThread() throws Exception { new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold())); DefaultEntryLogger entryLogger = new DefaultEntryLogger(conf, ledgerDirsManager); - EntryLogManagerForEntryLogPerLedger entryLogManager = - (EntryLogManagerForEntryLogPerLedger) entryLogger.getEntryLogManager(); + try { + EntryLogManagerForEntryLogPerLedger entryLogManager = + (EntryLogManagerForEntryLogPerLedger) entryLogger.getEntryLogManager(); - long ledgerId = 0L; + long ledgerId = 0L; - BufferedLogChannel newLogChannel = createDummyBufferedLogChannel(1, conf); - entryLogManager.setCurrentLogForLedgerAndAddToRotate(ledgerId, newLogChannel); + BufferedLogChannel newLogChannel = createDummyBufferedLogChannel(1, conf); + entryLogManager.setCurrentLogForLedgerAndAddToRotate(ledgerId, newLogChannel); - Thread t = new Thread() { - public void run() { - try { - Thread.sleep((evictionPeriod * 1000) / 2); - entryLogManager.getCurrentLogForLedger(ledgerId); - } catch (InterruptedException | IOException e) { + Thread t = new Thread() { + public void run() { + try { + Thread.sleep((evictionPeriod * 1000) / 2); + entryLogManager.getCurrentLogForLedger(ledgerId); + } catch (InterruptedException | IOException e) { + } } - } - }; + }; - t.start(); - Thread.sleep(evictionPeriod * 1000 + 100); - entryLogManager.doEntryLogMapCleanup(); + t.start(); + Thread.sleep(evictionPeriod * 1000 + 100); + entryLogManager.doEntryLogMapCleanup(); + t.join(); - /* - * in this scenario, that ledger is accessed by other thread during - * eviction period time, so it should not be evicted. - */ - BufferedLogChannel currentLogForLedger = entryLogManager.getCurrentLogForLedger(ledgerId); - assertEquals(newLogChannel, currentLogForLedger, "LogChannel for ledger " + ledgerId); - assertEquals(1, entryLogManager.getCopyOfCurrentLogs().size(), "Number of current active EntryLogs "); - assertEquals(0, entryLogManager.getRotatedLogChannels().size(), "Number of rotated EntryLogs "); + /* + * in this scenario, that ledger is accessed by other thread during + * eviction period time, so it should not be evicted. + */ + BufferedLogChannel currentLogForLedger = entryLogManager.getCurrentLogForLedger(ledgerId); + assertEquals(newLogChannel, currentLogForLedger, "LogChannel for ledger " + ledgerId); + assertEquals(1, entryLogManager.getCopyOfCurrentLogs().size(), "Number of current active EntryLogs "); + assertEquals(0, entryLogManager.getRotatedLogChannels().size(), "Number of rotated EntryLogs "); + } finally { + entryLogger.close(); + } } /** @@ -1438,59 +1509,64 @@ public void testExpiryRemovalByAccessingNonCacheRelatedMethods() throws Exceptio new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold())); DefaultEntryLogger entryLogger = new DefaultEntryLogger(conf, ledgerDirsManager); - EntryLogManagerForEntryLogPerLedger entryLogManager = - (EntryLogManagerForEntryLogPerLedger) entryLogger.getEntryLogManager(); - - long ledgerId = 0L; - - BufferedLogChannel newLogChannel = createDummyBufferedLogChannel(1, conf); - entryLogManager.setCurrentLogForLedgerAndAddToRotate(ledgerId, newLogChannel); - - AtomicBoolean exceptionOccurred = new AtomicBoolean(false); - Thread t = new Thread() { - public void run() { - try { - Thread.sleep(500); - /* - * any of the following operations should not access entry - * of 'ledgerId' in the cache - */ - entryLogManager.getCopyOfCurrentLogs(); - entryLogManager.getRotatedLogChannels(); - entryLogManager.getCurrentLogIfPresent(newLogChannel.getLogId()); - entryLogManager.getDirForNextEntryLog(ledgerDirsManager.getWritableLedgerDirs()); - long newLedgerId = 100; - BufferedLogChannel logChannelForNewLedger = - createDummyBufferedLogChannel(newLedgerId, conf); - entryLogManager.setCurrentLogForLedgerAndAddToRotate(newLedgerId, logChannelForNewLedger); - entryLogManager.getCurrentLogIfPresent(newLedgerId); - } catch (Exception e) { - LOG.error("Got Exception in thread", e); - exceptionOccurred.set(true); + try { + EntryLogManagerForEntryLogPerLedger entryLogManager = + (EntryLogManagerForEntryLogPerLedger) entryLogger.getEntryLogManager(); + + long ledgerId = 0L; + + BufferedLogChannel newLogChannel = createDummyBufferedLogChannel(1, conf); + entryLogManager.setCurrentLogForLedgerAndAddToRotate(ledgerId, newLogChannel); + + AtomicBoolean exceptionOccurred = new AtomicBoolean(false); + Thread t = new Thread() { + public void run() { + try { + Thread.sleep(500); + /* + * any of the following operations should not access entry + * of 'ledgerId' in the cache + */ + entryLogManager.getCopyOfCurrentLogs(); + entryLogManager.getRotatedLogChannels(); + entryLogManager.getCurrentLogIfPresent(newLogChannel.getLogId()); + entryLogManager.getDirForNextEntryLog(ledgerDirsManager.getWritableLedgerDirs()); + long newLedgerId = 100; + BufferedLogChannel logChannelForNewLedger = + createDummyBufferedLogChannel(newLedgerId, conf); + entryLogManager.setCurrentLogForLedgerAndAddToRotate(newLedgerId, logChannelForNewLedger); + entryLogManager.getCurrentLogIfPresent(newLedgerId); + } catch (Exception e) { + LOG.error("Got Exception in thread", e); + exceptionOccurred.set(true); + } } - } - }; + }; - t.start(); - Thread.sleep(evictionPeriod * 1000 + 100); - entryLogManager.doEntryLogMapCleanup(); - assertFalse(exceptionOccurred.get(), "Exception occurred in thread, which is not expected"); + t.start(); + Thread.sleep(evictionPeriod * 1000 + 100); + entryLogManager.doEntryLogMapCleanup(); + t.join(); + assertFalse(exceptionOccurred.get(), "Exception occurred in thread, which is not expected"); - /* - * since for more than evictionPeriod, that ledger is not accessed and cache is cleaned up, mapping for that - * ledger should not be available anymore - */ - BufferedLogChannel currentLogForLedger = entryLogManager.getCurrentLogForLedger(ledgerId); - assertNull(currentLogForLedger, "LogChannel for ledger " + ledgerId + " should be null"); - // expected number of current active entryLogs is 1 since we created entrylog for 'newLedgerId' - assertEquals(1, entryLogManager.getCopyOfCurrentLogs().size(), "Number of current active EntryLogs "); - assertEquals(1, entryLogManager.getRotatedLogChannels().size(), "Number of rotated EntryLogs "); - assertTrue(entryLogManager.getRotatedLogChannels().contains(newLogChannel), - "CopyOfRotatedLogChannels should contain the created LogChannel"); - - assertTrue((entryLogManager.getCacheAsMap().get(ledgerId) == null) - || (entryLogManager.getCacheAsMap().get(ledgerId).getEntryLogWithDirInfo() == null), - "since mapentry must have been evicted, it should be null"); + /* + * since for more than evictionPeriod, that ledger is not accessed and cache is cleaned up, mapping for that + * ledger should not be available anymore + */ + BufferedLogChannel currentLogForLedger = entryLogManager.getCurrentLogForLedger(ledgerId); + assertNull(currentLogForLedger, "LogChannel for ledger " + ledgerId + " should be null"); + // expected number of current active entryLogs is 1 since we created entrylog for 'newLedgerId' + assertEquals(1, entryLogManager.getCopyOfCurrentLogs().size(), "Number of current active EntryLogs "); + assertEquals(1, entryLogManager.getRotatedLogChannels().size(), "Number of rotated EntryLogs "); + assertTrue(entryLogManager.getRotatedLogChannels().contains(newLogChannel), + "CopyOfRotatedLogChannels should contain the created LogChannel"); + + assertTrue((entryLogManager.getCacheAsMap().get(ledgerId) == null) + || (entryLogManager.getCacheAsMap().get(ledgerId).getEntryLogWithDirInfo() == null), + "since mapentry must have been evicted, it should be null"); + } finally { + entryLogger.close(); + } } /* @@ -1639,13 +1715,17 @@ public void testReadAddCallsOfMultipleEntryLogs() throws Exception { for (int i = 0; i < numOfActiveLedgers; i++) { String expectedValue = "ledger-" + i + "-" + j; ByteBuf buf = entryLogger.readEntry(i, j, positions[i][j]); - long ledgerId = buf.readLong(); - long entryId = buf.readLong(); - byte[] data = new byte[buf.readableBytes()]; - buf.readBytes(data); - assertEquals(i, ledgerId, "LedgerId "); - assertEquals(j, entryId, "EntryId "); - assertEquals(expectedValue, new String(data), "Entry Data "); + try { + long ledgerId = buf.readLong(); + long entryId = buf.readLong(); + byte[] data = new byte[buf.readableBytes()]; + buf.readBytes(data); + assertEquals(i, ledgerId, "LedgerId "); + assertEquals(j, entryId, "EntryId "); + assertEquals(expectedValue, new String(data), "Entry Data "); + } finally { + ReferenceCountUtil.release(buf); + } } } @@ -1660,13 +1740,17 @@ public void testReadAddCallsOfMultipleEntryLogs() throws Exception { for (int i = 0; i < numOfActiveLedgers; i++) { String expectedValue = "ledger-" + i + "-" + j; ByteBuf buf = entryLogger.readEntry(i, j, positions[i][j]); - long ledgerId = buf.readLong(); - long entryId = buf.readLong(); - byte[] data = new byte[buf.readableBytes()]; - buf.readBytes(data); - assertEquals(i, ledgerId, "LedgerId "); - assertEquals(j, entryId, "EntryId "); - assertEquals(expectedValue, new String(data), "Entry Data "); + try { + long ledgerId = buf.readLong(); + long entryId = buf.readLong(); + byte[] data = new byte[buf.readableBytes()]; + buf.readBytes(data); + assertEquals(i, ledgerId, "LedgerId "); + assertEquals(j, entryId, "EntryId "); + assertEquals(expectedValue, new String(data), "Entry Data "); + } finally { + ReferenceCountUtil.release(buf); + } } } } @@ -1686,9 +1770,11 @@ static class ReadTask implements Callable { @Override public Boolean call() throws IOException { + ByteBuf expectedByteBuf = null; + ByteBuf actualByteBuf = null; try { - ByteBuf expectedByteBuf = generateEntry(ledgerId, entryId); - ByteBuf actualByteBuf = entryLogger.readEntry(ledgerId, entryId, position); + expectedByteBuf = generateEntry(ledgerId, entryId); + actualByteBuf = entryLogger.readEntry(ledgerId, entryId, position); if (!expectedByteBuf.equals(actualByteBuf)) { LOG.error("Expected Entry: {} Actual Entry: {}", expectedByteBuf.toString(Charset.defaultCharset()), actualByteBuf.toString(Charset.defaultCharset())); @@ -1699,6 +1785,9 @@ public Boolean call() throws IOException { LOG.error("Got Exception for GetEntry call. LedgerId: " + ledgerId + " entryId: " + entryId, e); throw new IOException("Got Exception for GetEntry call. LedgerId: " + ledgerId + " entryId: " + entryId, e); + } finally { + ReferenceCountUtil.release(actualByteBuf); + ReferenceCountUtil.release(expectedByteBuf); } return true; } @@ -1975,13 +2064,17 @@ public void testSwappingEntryLogManager(boolean initialEntryLogPerLedgerEnabled, for (int i = 0; i < numOfActiveLedgers; i++) { String expectedValue = "ledger-" + i + "-" + j; ByteBuf buf = newEntryLogger.readEntry(i, j, positions[i][j]); - long ledgerId = buf.readLong(); - long entryId = buf.readLong(); - byte[] data = new byte[buf.readableBytes()]; - buf.readBytes(data); - assertEquals(i, ledgerId, "LedgerId "); - assertEquals(j, entryId, "EntryId "); - assertEquals(expectedValue, new String(data), "Entry Data "); + try { + long ledgerId = buf.readLong(); + long entryId = buf.readLong(); + byte[] data = new byte[buf.readableBytes()]; + buf.readBytes(data); + assertEquals(i, ledgerId, "LedgerId "); + assertEquals(j, entryId, "EntryId "); + assertEquals(expectedValue, new String(data), "Entry Data "); + } finally { + ReferenceCountUtil.release(buf); + } } } } diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageTest.java index 3bb992888ab..8fb1372b5b0 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageTest.java @@ -170,7 +170,7 @@ public void simple() throws Exception { // Read from write cache assertTrue(storage.entryExists(4, 1)); ByteBuf res = storage.getEntry(4, 1); - assertEquals(entry, res); + assertByteBufEqualsAndRelease(entry, res); storage.flush(); @@ -179,7 +179,7 @@ public void simple() throws Exception { // Read from db assertTrue(storage.entryExists(4, 1)); res = storage.getEntry(4, 1); - assertEquals(entry, res); + assertByteBufEqualsAndRelease(entry, res); try { storage.getEntry(4, 2); @@ -198,7 +198,7 @@ public void simple() throws Exception { // Read last entry in ledger res = storage.getEntry(4, BookieProtocol.LAST_ADD_CONFIRMED); - assertEquals(entry2, res); + assertByteBufEqualsAndRelease(entry2, res); // Read last add confirmed in ledger assertEquals(1L, storage.getLastAddConfirmed(4)); @@ -218,7 +218,7 @@ public void simple() throws Exception { storage.addEntry(entry4); res = storage.getEntry(4, 4); - assertEquals(entry4, res); + assertByteBufEqualsAndRelease(entry4, res); assertEquals(3, storage.getLastAddConfirmed(4)); @@ -267,7 +267,7 @@ public void testBookieCompaction() throws Exception { ByteBuf res = storage.getEntry(4, 3); System.out.println("res: " + ByteBufUtil.hexDump(res)); System.out.println("newEntry3: " + ByteBufUtil.hexDump(newEntry3)); - assertEquals(newEntry3, res); + assertByteBufEqualsAndRelease(newEntry3, res); } @Test @@ -392,7 +392,7 @@ public void testRewritingEntries() throws Exception { storage.flush(); ByteBuf response = storage.getEntry(1, 1); - assertEquals(newEntry1, response); + assertByteBufEqualsAndRelease(newEntry1, response); } @Test @@ -414,7 +414,7 @@ public void testEntriesOutOfOrder() throws Exception { } ByteBuf res = storage.getEntry(1, 2); - assertEquals(entry2, res); + assertByteBufEqualsAndRelease(entry2, res); ByteBuf entry1 = Unpooled.buffer(1024); entry1.writeLong(1); // ledger id @@ -424,18 +424,18 @@ public void testEntriesOutOfOrder() throws Exception { storage.addEntry(entry1); res = storage.getEntry(1, 1); - assertEquals(entry1, res); + assertByteBufEqualsAndRelease(entry1, res); res = storage.getEntry(1, 2); - assertEquals(entry2, res); + assertByteBufEqualsAndRelease(entry2, res); storage.flush(); res = storage.getEntry(1, 1); - assertEquals(entry1, res); + assertByteBufEqualsAndRelease(entry1, res); res = storage.getEntry(1, 2); - assertEquals(entry2, res); + assertByteBufEqualsAndRelease(entry2, res); } @Test @@ -535,8 +535,8 @@ public void testAddEntriesAfterDelete() throws Exception { storage.addEntry(entry0); storage.addEntry(entry1); - assertEquals(entry0, storage.getEntry(1, 0)); - assertEquals(entry1, storage.getEntry(1, 1)); + assertByteBufEqualsAndRelease(entry0, storage.getEntry(1, 0)); + assertByteBufEqualsAndRelease(entry1, storage.getEntry(1, 1)); storage.flush(); } @@ -554,10 +554,13 @@ public void testLimboStateSucceedsWhenInLimboButHasEntry() throws Exception { storage.flush(); storage.setLimboState(1); + ByteBuf result = null; try { - storage.getEntry(1, 0); + result = storage.getEntry(1, 0); } catch (BookieException.DataUnknownException e) { fail("Should have been able to read entry"); + } finally { + ReferenceCountUtil.release(result); } } @@ -574,24 +577,30 @@ public void testLimboStateThrowsInLimboWhenNoEntry() throws Exception { storage.flush(); storage.setLimboState(1); + ByteBuf result = null; try { - storage.getEntry(1, 1); + result = storage.getEntry(1, 1); } catch (NoEntryException nee) { fail("Shouldn't have seen NoEntryException"); } catch (BookieException.DataUnknownException e) { // expected + } finally { + ReferenceCountUtil.release(result); } storage.shutdown(); Bookie restartedBookie = new TestBookieImpl(conf); DbLedgerStorage restartedStorage = (DbLedgerStorage) restartedBookie.getLedgerStorage(); try { + result = null; try { - restartedStorage.getEntry(1, 1); + result = restartedStorage.getEntry(1, 1); } catch (NoEntryException nee) { fail("Shouldn't have seen NoEntryException"); } catch (BookieException.DataUnknownException e) { // expected + } finally { + ReferenceCountUtil.release(result); } } finally { restartedStorage.shutdown(); @@ -613,21 +622,27 @@ public void testLimboStateThrowsNoEntryExceptionWhenLimboCleared() throws Except storage.flush(); storage.setLimboState(1); + ByteBuf result = null; try { - storage.getEntry(1, 1); + result = storage.getEntry(1, 1); } catch (NoEntryException nee) { fail("Shouldn't have seen NoEntryException"); } catch (BookieException.DataUnknownException e) { // expected + } finally { + ReferenceCountUtil.release(result); } storage.clearLimboState(1); + result = null; try { - storage.getEntry(1, 1); + result = storage.getEntry(1, 1); } catch (NoEntryException nee) { // expected } catch (BookieException.DataUnknownException e) { fail("Should have seen NoEntryException"); + } finally { + ReferenceCountUtil.release(result); } } @@ -696,7 +711,7 @@ public void testHasEntry() throws Exception { assertFalse(storage.entryExists(ledgerId, 1)); // pull entry into readcache - storage.getEntry(ledgerId, 0); + ReferenceCountUtil.release(storage.getEntry(ledgerId, 0)); // should come from read cache assertTrue(storage.entryExists(ledgerId, 0)); @@ -847,6 +862,14 @@ private LogMark readLogMark(File file) throws IOException { return mark; } + private static void assertByteBufEqualsAndRelease(ByteBuf expected, ByteBuf actual) { + try { + assertEquals(expected, actual); + } finally { + ReferenceCountUtil.release(actual); + } + } + @Test public void testSingleLedgerDirectoryCheckpoint() throws Exception { int gcWaitTime = 1000; From 90885008f7330fc852a340139b4ec594c6d3e33a Mon Sep 17 00:00:00 2001 From: yangxianjungree <714696209@qq.com> Date: Wed, 5 Aug 2026 10:36:47 +0800 Subject: [PATCH 8/9] fix: make entrylog failure shutdown cleanup idempotent --- .../bookkeeper/bookie/DefaultEntryLogger.java | 6 ++++++ .../bookkeeper/bookie/EntryLoggerAllocator.java | 9 ++++++++- .../org/apache/bookkeeper/bookie/SyncThread.java | 9 ++++++++- .../apache/bookkeeper/bookie/BookieImplTest.java | 6 ++++++ .../bookkeeper/bookie/DefaultEntryLogTest.java | 13 +++++++++++++ .../apache/bookkeeper/bookie/SyncThreadTest.java | 14 ++++++++++++++ ...DbLedgerStorageEntryLogFlushFailureE2ETest.java | 1 + .../storage/ldb/FailOnFlushDbLedgerStorage.java | 5 +---- 8 files changed, 57 insertions(+), 6 deletions(-) diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/DefaultEntryLogger.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/DefaultEntryLogger.java index 85001db4ad4..76af61f164e 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/DefaultEntryLogger.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/DefaultEntryLogger.java @@ -56,6 +56,7 @@ 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; @@ -240,6 +241,7 @@ private static void writeFully(FileChannel fileChannel, ByteBuffer buffer, long final EntryLoggerAllocator entryLoggerAllocator; private final EntryLogManager entryLogManager; + private final AtomicBoolean closed = new AtomicBoolean(false); private final CopyOnWriteArrayList listeners = new CopyOnWriteArrayList(); @@ -1230,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 { diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLoggerAllocator.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLoggerAllocator.java index 13475d4f051..e6258214c6a 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLoggerAllocator.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLoggerAllocator.java @@ -42,6 +42,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import lombok.CustomLog; @@ -265,7 +266,13 @@ void setLastLogId(File dir, long logId) throws IOException { */ void stop() { // wait until the preallocation finished. - allocatorExecutor.execute(this::closePreAllocateLog); + if (!allocatorExecutor.isShutdown()) { + try { + allocatorExecutor.execute(this::closePreAllocateLog); + } catch (RejectedExecutionException e) { + log.debug().exception(e).log("Skipping preallocated entry log cleanup because allocator is stopping"); + } + } allocatorExecutor.shutdown(); try { if (!allocatorExecutor.awaitTermination(5, TimeUnit.SECONDS)) { diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SyncThread.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SyncThread.java index 1f978360358..1a2a1e0c380 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SyncThread.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SyncThread.java @@ -26,6 +26,7 @@ import java.io.IOException; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import lombok.AccessLevel; @@ -235,7 +236,13 @@ public void disableCheckpoint() { // shutdown sync thread void shutdown() throws InterruptedException { log.info("Shutting down SyncThread"); - requestFlush(); + if (!executor.isShutdown()) { + try { + requestFlush(); + } catch (RejectedExecutionException e) { + log.debug().exception(e).log("Skipping final flush because SyncThread executor is shutting down"); + } + } executor.shutdown(); long start = System.currentTimeMillis(); diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieImplTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieImplTest.java index 2b173a51637..3f372d49d91 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieImplTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieImplTest.java @@ -147,6 +147,9 @@ public void testBackgroundEntryLogFlushFailureShutsDownBookie() throws Exception BookieImpl bookie = new TestBookieImpl(resources) { @Override int shutdown(int exitCode) { + if (exitCode == ExitCode.BOOKIE_EXCEPTION) { + FailOnFlushDbLedgerStorage.resetFailure(); + } int result = super.shutdown(exitCode); if (exitCode == ExitCode.BOOKIE_EXCEPTION) { shutdownLatch.countDown(); @@ -182,6 +185,9 @@ public void testStartupEntryLogFlushFailureStopsBookieBeforeRunning() throws Exc BookieImpl bookie = new TestBookieImpl(resources) { @Override int shutdown(int exitCode) { + if (exitCode == ExitCode.BOOKIE_EXCEPTION) { + FailOnFlushDbLedgerStorage.resetFailure(); + } int result = super.shutdown(exitCode); if (exitCode == ExitCode.BOOKIE_EXCEPTION) { shutdownLatch.countDown(); diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/DefaultEntryLogTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/DefaultEntryLogTest.java index b07f6dd4d88..9277d931831 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/DefaultEntryLogTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/DefaultEntryLogTest.java @@ -539,6 +539,19 @@ public void testPreAllocateLog() throws Exception { assertNull(entryLogger.getEntryLoggerAllocator().getPreallocationFuture()); } + @Test + public void testCloseIsIdempotentWithPreAllocation() throws Exception { + entryLogger.close(); + + conf.setEntryLogFilePreAllocationEnabled(true); + entryLogger = new DefaultEntryLogger(conf, dirsMgr); + ((EntryLogManagerBase) entryLogger.getEntryLogManager()).createNewLog(DefaultEntryLogger.UNASSIGNED_LEDGERID); + assertNotNull(entryLogger.getEntryLoggerAllocator().getPreallocationFuture()); + + entryLogger.close(); + entryLogger.close(); + } + /** * Test the getEntryLogsSet() method. */ diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/SyncThreadTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/SyncThreadTest.java index 98857c616c3..526a93fbe8b 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/SyncThreadTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/SyncThreadTest.java @@ -147,6 +147,20 @@ public void checkpoint(Checkpoint checkpoint) assertFalse("Shouldn't have failed anywhere", failedSomewhere.get()); } + @Test + public void testSyncThreadShutdownIsIdempotent() throws Exception { + int flushInterval = 100; + ServerConfiguration conf = TestBKConfiguration.newServerConfiguration(); + conf.setFlushInterval(flushInterval); + CheckpointSource checkpointSource = new DummyCheckpointSource(); + LedgerDirsListener listener = new LedgerDirsListener() {}; + LedgerStorage storage = new DummyLedgerStorage(); + + final SyncThread t = new SyncThread(conf, listener, storage, checkpointSource, NullStatsLogger.INSTANCE); + t.shutdown(); + t.shutdown(); + } + /** * Test that sync thread suspension works. * i.e. when we suspend the syncthread, nothing diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageEntryLogFlushFailureE2ETest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageEntryLogFlushFailureE2ETest.java index ce093ecd367..e529113a798 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageEntryLogFlushFailureE2ETest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageEntryLogFlushFailureE2ETest.java @@ -65,6 +65,7 @@ public void testClientWriteFailsAndBookieShutsDownAfterEntryLogFlushFailure() th } assertNotNull("Client should observe a write failure after the entry log flush failure", clientFailure); + FailOnFlushDbLedgerStorage.resetFailure(); Awaitility.await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertFalse("Bookie should be shut down after entry log flush failure", bookie.isRunning())); } finally { diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/FailOnFlushDbLedgerStorage.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/FailOnFlushDbLedgerStorage.java index 19d97fe8da2..cc6286de042 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/FailOnFlushDbLedgerStorage.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/FailOnFlushDbLedgerStorage.java @@ -32,7 +32,6 @@ public class FailOnFlushDbLedgerStorage extends DbLedgerStorage { private static final AtomicBoolean failNextFlushWithEntryLogWriteException = new AtomicBoolean(false); - private static final AtomicBoolean entryLogFlushFailed = new AtomicBoolean(false); public static void injectFailureOnNextFlush() { failNextFlushWithEntryLogWriteException.set(true); @@ -40,7 +39,6 @@ public static void injectFailureOnNextFlush() { public static void resetFailure() { failNextFlushWithEntryLogWriteException.set(false); - entryLogFlushFailed.set(false); } @Override @@ -66,8 +64,7 @@ private static class FailOnFlushSingleDirectoryDbLedgerStorage extends SingleDir @Override public void flush() throws IOException { - if (entryLogFlushFailed.get() || failNextFlushWithEntryLogWriteException.compareAndSet(true, false)) { - entryLogFlushFailed.set(true); + if (failNextFlushWithEntryLogWriteException.compareAndSet(true, false)) { throw new EntryLogWriteException("entry log flush failed", new IOException("injected")); } super.flush(); From acff4638b3b1d53ebc25dfbb2eb2feee21aa1ae6 Mon Sep 17 00:00:00 2001 From: yangxianjungree <714696209@qq.com> Date: Wed, 5 Aug 2026 15:37:14 +0800 Subject: [PATCH 9/9] fix: continue db storage shutdown cleanup after flush failure --- .../ldb/SingleDirectoryDbLedgerStorage.java | 37 ++++- .../bookie/BufferedChannelTest.java | 39 ++++++ ...eDirectoryDbLedgerStorageShutdownTest.java | 130 ++++++++++++++++++ 3 files changed, 199 insertions(+), 7 deletions(-) create mode 100644 bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorageShutdownTest.java diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java index 994459b054a..b1f480c5352 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java @@ -79,6 +79,7 @@ import org.apache.bookkeeper.stats.OpStatsLogger; import org.apache.bookkeeper.stats.StatsLogger; import org.apache.bookkeeper.stats.ThreadRegistry; +import org.apache.bookkeeper.util.IOUtils; import org.apache.bookkeeper.util.collections.ConcurrentLongHashMap; import org.apache.commons.collections4.CollectionUtils; import org.apache.commons.lang3.mutable.MutableLong; @@ -354,25 +355,47 @@ public List getEntryLocationDBPath() { @Override public void shutdown() throws InterruptedException { + InterruptedException interrupted = null; + try { flush(); + } catch (IOException e) { + log.error().exception(e).log("Error flushing db storage during shutdown"); + } finally { + try { + gcThread.shutdown(); + } catch (InterruptedException e) { + interrupted = e; + Thread.currentThread().interrupt(); + } - gcThread.shutdown(); - entryLogger.close(); + try { + entryLogger.close(); + } catch (IOException e) { + log.error().exception(e).log("Error closing entry logger during shutdown"); + } cleanupExecutor.shutdown(); - cleanupExecutor.awaitTermination(1, TimeUnit.SECONDS); + try { + cleanupExecutor.awaitTermination(1, TimeUnit.SECONDS); + } catch (InterruptedException e) { + if (interrupted == null) { + interrupted = e; + } + Thread.currentThread().interrupt(); + } - ledgerIndex.close(); - entryLocationIndex.close(); + IOUtils.close(log, ledgerIndex); + IOUtils.close(log, entryLocationIndex); writeCache.close(); writeCacheBeingFlushed.close(); readCache.close(); executor.shutdown(); + } - } catch (IOException e) { - log.error().exception(e).log("Error closing db storage"); + if (interrupted != null) { + throw interrupted; } } diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BufferedChannelTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BufferedChannelTest.java index 30e5db0b7c8..d605992d829 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BufferedChannelTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BufferedChannelTest.java @@ -125,6 +125,45 @@ public void testPartialFlushFailurePoisonsBufferedChannel() throws Exception { logChannel.close(); } + @Test + public void testPartialFlushFailurePoisonsBufferedChannelForTwoBufferWrite() throws Exception { + File newLogFile = File.createTempFile("test", "log"); + newLogFile.deleteOnExit(); + FileChannel delegate = new RandomAccessFile(newLogFile, "rw").getChannel(); + PartialFailingFileChannel fileChannel = new PartialFailingFileChannel(delegate, 8); + + BufferedChannel logChannel = new BufferedChannel(UnpooledByteBufAllocator.DEFAULT, fileChannel, + 16, INTERNAL_BUFFER_READ_CAPACITY, 0); + + try { + logChannel.write(Unpooled.wrappedBuffer(new byte[8]), Unpooled.wrappedBuffer(new byte[8])); + Assert.fail("Expected the internal flush to fail"); + } catch (IOException expected) { + // Expected. + } + + Assert.assertEquals(0, logChannel.position()); + Assert.assertEquals(8, fileChannel.position()); + + try { + logChannel.write(Unpooled.wrappedBuffer(new byte[1]), Unpooled.wrappedBuffer(new byte[1])); + Assert.fail("Expected writes after a partial flush failure to fail"); + } catch (IOException expected) { + // Expected. + } + try { + logChannel.flush(); + Assert.fail("Expected flush after a partial flush failure to fail"); + } catch (IOException expected) { + // Expected. + } + + Assert.assertEquals(0, logChannel.position()); + Assert.assertEquals(8, fileChannel.position()); + + logChannel.close(); + } + @Test public void testPartialFlushFailurePreventsLedgersMapHeaderUpdate() throws Exception { File newLogFile = File.createTempFile("test", "log"); diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorageShutdownTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorageShutdownTest.java new file mode 100644 index 00000000000..079cb1d251c --- /dev/null +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorageShutdownTest.java @@ -0,0 +1,130 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.bookkeeper.bookie.storage.ldb; + +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +import io.netty.buffer.ByteBufAllocator; +import java.io.File; +import java.io.IOException; +import java.lang.reflect.Field; +import java.util.concurrent.ExecutorService; +import org.apache.bookkeeper.bookie.BookieImpl; +import org.apache.bookkeeper.bookie.EntryLogWriteException; +import org.apache.bookkeeper.bookie.GarbageCollectorThread; +import org.apache.bookkeeper.bookie.LedgerDirsManager; +import org.apache.bookkeeper.bookie.storage.EntryLogger; +import org.apache.bookkeeper.conf.ServerConfiguration; +import org.apache.bookkeeper.conf.TestBKConfiguration; +import org.apache.bookkeeper.meta.LedgerManager; +import org.apache.bookkeeper.stats.NullStatsLogger; +import org.apache.bookkeeper.stats.StatsLogger; +import org.apache.bookkeeper.util.DiskChecker; +import org.apache.commons.io.FileUtils; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +/** + * Tests shutdown cleanup for {@link SingleDirectoryDbLedgerStorage}. + */ +public class SingleDirectoryDbLedgerStorageShutdownTest { + + private static final long MB = 1024 * 1024; + + private File tmpDir; + private EntryLogger entryLogger; + private FailingFlushSingleDirectoryDbLedgerStorage storage; + + @Before + public void setup() throws Exception { + tmpDir = File.createTempFile("bkTest", ".dir"); + tmpDir.delete(); + tmpDir.mkdir(); + File curDir = BookieImpl.getCurrentDirectory(tmpDir); + BookieImpl.checkDirectoryStructure(curDir); + + ServerConfiguration conf = TestBKConfiguration.newServerConfiguration(); + conf.setGcWaitTime(1000); + conf.setLedgerDirNames(new String[] { tmpDir.toString() }); + DiskChecker diskChecker = new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold()); + LedgerDirsManager ledgerDirsManager = new LedgerDirsManager(conf, conf.getLedgerDirs(), diskChecker); + LedgerDirsManager indexDirsManager = new LedgerDirsManager(conf, conf.getLedgerDirs(), diskChecker); + entryLogger = mock(EntryLogger.class); + + storage = new FailingFlushSingleDirectoryDbLedgerStorage(conf, mock(LedgerManager.class), + ledgerDirsManager, indexDirsManager, entryLogger, NullStatsLogger.INSTANCE, + ByteBufAllocator.DEFAULT, MB, MB, 1, 1024); + } + + @After + public void teardown() throws Exception { + if (storage != null) { + storage.shutdown(); + } + FileUtils.deleteDirectory(tmpDir); + } + + @Test + public void shutdownContinuesCleanupAfterFlushFailure() throws Exception { + storage.shutdown(); + verify(entryLogger).close(); + assertFalse(isGcThreadRunning()); + assertTrue(getCleanupExecutor().isShutdown()); + storage = null; + } + + private boolean isGcThreadRunning() throws Exception { + Field gcThreadField = SingleDirectoryDbLedgerStorage.class.getDeclaredField("gcThread"); + gcThreadField.setAccessible(true); + GarbageCollectorThread gcThread = (GarbageCollectorThread) gcThreadField.get(storage); + + Field runningField = GarbageCollectorThread.class.getDeclaredField("running"); + runningField.setAccessible(true); + return runningField.getBoolean(gcThread); + } + + private ExecutorService getCleanupExecutor() throws Exception { + Field cleanupExecutorField = SingleDirectoryDbLedgerStorage.class.getDeclaredField("cleanupExecutor"); + cleanupExecutorField.setAccessible(true); + return (ExecutorService) cleanupExecutorField.get(storage); + } + + private static class FailingFlushSingleDirectoryDbLedgerStorage extends SingleDirectoryDbLedgerStorage { + + FailingFlushSingleDirectoryDbLedgerStorage(ServerConfiguration conf, LedgerManager ledgerManager, + LedgerDirsManager ledgerDirsManager, LedgerDirsManager indexDirsManager, EntryLogger entryLogger, + StatsLogger statsLogger, ByteBufAllocator allocator, long writeCacheSize, long readCacheSize, + int readAheadCacheBatchSize, long readAheadCacheBatchBytesSize) + throws IOException { + super(conf, ledgerManager, ledgerDirsManager, indexDirsManager, entryLogger, statsLogger, allocator, + writeCacheSize, readCacheSize, readAheadCacheBatchSize, readAheadCacheBatchBytesSize); + } + + @Override + public void flush() throws IOException { + throw new EntryLogWriteException("entry log flush failed", new IOException("injected")); + } + } +}