Copilot commented on code in PR #4863:
URL: https://github.com/apache/bookkeeper/pull/4863#discussion_r3974830888
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerBase.java:
##########
@@ -117,6 +127,21 @@ List<BufferedLogChannel> 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().exception(cause).log(message);
+ fatalErrorListener.fatalError();
+ for (LedgerDirsListener listener : ledgerDirsManager.getListeners()) {
+ if (listener != fatalErrorListener) {
+ listener.fatalError();
Review Comment:
A single misbehaving `LedgerDirsListener` that throws at `fatalError()` can
prevent notifying the remaining listeners, which makes shutdown/transition
behavior less reliable under stress. Wrap each `fatalError()` invocation in its
own try/catch and continue notifying others (logging any listener failure).
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BufferedChannel.java:
##########
@@ -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;
+ }
+ }
Review Comment:
`writeFailure` is `volatile`, but `markWriteFailure()` performs a non-atomic
check-then-set. If multiple threads hit failures concurrently, the
original/root failure can be overwritten (or lost), making diagnosis harder and
potentially changing the thrown cause nondeterministically. Consider using an
`AtomicReference<IOException>` with `compareAndSet(null, e)` (or synchronizing
`markWriteFailure`) so the first failure is preserved deterministically.
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/DefaultEntryLogger.java:
##########
@@ -204,7 +207,23 @@ public void accept(long ledgerId, long size) {
mapInfo.putLong(ledgerMapOffset);
mapInfo.putInt(numberOfLedgers);
mapInfo.flip();
- this.fileChannel.write(mapInfo, LEDGERS_MAP_OFFSET_POSITION);
+ try {
+ writeFully(this.fileChannel, mapInfo,
LEDGERS_MAP_OFFSET_POSITION);
+ } catch (IOException e) {
+ markWriteFailure(e);
+ throw e;
+ }
+ }
+
+ private static void writeFully(FileChannel fileChannel, ByteBuffer
buffer, long position) throws IOException {
+ long writePosition = position;
+ while (buffer.hasRemaining()) {
+ int written = fileChannel.write(buffer, writePosition);
+ if (written <= 0) {
+ throw new IOException("Unable to make progress while
updating entry log header");
+ }
+ writePosition += written;
+ }
}
Review Comment:
The new error message is generic and will be hard to action operationally.
Including context (e.g., logId/file path if available at the call site, and the
target position/remaining bytes) would make diagnosing partial/header-write
issues significantly easier.
##########
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 is stored in a static flag, which can leak state
across tests and can become flaky if the test suite is run with parallelism (or
multiple clusters/tests in the same JVM). Prefer instance-scoped injection
(e.g., per-storage field) or tying it to the specific bookie/test instance
(e.g., via configuration or dependency injection) to avoid cross-test
interference.
##########
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();
Review Comment:
`File.createTempFile()` + `delete()` + `mkdir()` is a brittle pattern (e.g.,
delete can fail on some platforms, `mkdir()` can fail silently here, and it’s
not guaranteed to be a directory). Prefer `Files.createTempDirectory(...)` (or
JUnit's `TemporaryFolder`) and assert directory creation succeeds to make the
test more robust across OS/filesystems.
##########
bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/DefaultEntryLogTest.java:
##########
@@ -1316,6 +1397,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(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();
Review Comment:
This test explicitly closes `fileChannel` and then unconditionally calls
`logChannel.close()` in `finally`. If `BufferedLogChannel.close()` propagates
an `IOException` when the underlying channel is already closed, the test can
fail during cleanup and mask the real assertion. Consider either (a) closing
only via `logChannel` (and inject failure another way), or (b) wrapping
`logChannel.close()` in a try/catch and ignoring/recording the expected close
failure.
--
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]
