Copilot commented on code in PR #4860:
URL: https://github.com/apache/bookkeeper/pull/4860#discussion_r3974833479
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLoggerAllocator.java:
##########
@@ -228,7 +241,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("Skipping preallocated entry log cleanup because
allocator is stopping", e);
+ }
Review Comment:
If `closePreAllocateLog` is skipped due to `RejectedExecutionException`, the
preallocated log may remain open/unclosed, which can leak file descriptors at
shutdown. A safer pattern is to perform `closePreAllocateLog()` inline when
task submission is rejected (or before shutting down the executor), so cleanup
is guaranteed even during racey shutdown sequences.
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BufferedChannel.java:
##########
@@ -295,4 +319,17 @@ public synchronized int getNumOfBytesInWriteBuffer() {
long getUnpersistedBytes() {
return unpersistedBytes.get();
}
-}
\ No newline at end of file
+
+ final void checkWritable() throws IOException {
+ IOException failure = writeFailure;
+ if (failure != null) {
Review Comment:
`checkWritable()` always throws a new generic `IOException` when the channel
is poisoned. If the original failure is an `EntryLogWriteException`, this loses
the exception type and can prevent upstream `catch (EntryLogWriteException)`
blocks from triggering fail-closed behavior. Consider rethrowing the original
exception when it is already an `EntryLogWriteException` (or otherwise
preserving its type), while still wrapping non-fatal `IOException`s.
##########
bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageEntryLogFlushFailureE2ETest.java:
##########
@@ -0,0 +1,75 @@
+/*
+ *
+ * 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 {
Review Comment:
`LedgerHandle lh` is never closed. Even if the cluster teardown closes the
client, explicitly closing the ledger handle in a `finally` block makes the
test more robust and avoids leaking resources (especially if this test is run
in isolation or failures occur mid-test).
##########
bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageTest.java:
##########
@@ -252,7 +260,63 @@ 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
+ 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));
Review Comment:
This test relies on fixed `Thread.sleep(...)` timing to trigger eviction
behavior, which makes it prone to flakiness under slow/loaded CI. Consider
replacing the sleep with a polling wait (e.g., Awaitility) that repeatedly
triggers/observes the eviction condition until it holds or a timeout is reached.
##########
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);
+ }
Review Comment:
The test uses reflection to reach into private fields (`gcThread`,
`running`, `cleanupExecutor`), making it brittle to internal refactors (e.g.,
field renames). If feasible, expose minimal test-only accessors (e.g.,
package-private getters annotated/marked for testing) or observable behaviors
that avoid reflection while still validating that GC and cleanup executors are
stopped.
##########
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);
+ }
Review Comment:
The failure injection flag is `static`, so it is shared across all instances
and tests in the same JVM. This can cause cross-test interference if tests are
ever run concurrently or re-ordered unexpectedly. Prefer instance-scoped
injection (e.g., a field on the storage instance) or a mechanism tied to the
specific Bookie/storage under test to improve test isolation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
