[fix]Wrong error code(-107) of opening a deleted ledger (#4657)

poorbarcode · web-flow · commit 5870922754b9 · 2025-09-02T12:26:50.000+08:00
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookieException.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookieException.java
@@ -69,6 +69,8 @@ public static BookieException create(int code) {
             return new MetadataStoreException();
         case Code.UnknownBookieIdException:
             return new UnknownBookieIdException();
+        case Code.LedgerFencedAndDeletedException:
+            return new LedgerFencedAndDeletedException();
         case Code.DataUnknownException:
             return new DataUnknownException();
         default:
@@ -95,6 +97,7 @@ public interface Code {
         int CookieExistsException = -109;
         int EntryLogMetadataMapException = -110;
         int DataUnknownException = -111;
+        int LedgerFencedAndDeletedException = -112;
     }
 
     public int getCode() {
@@ -199,6 +202,15 @@ public LedgerFencedException() {
         }
     }
 
+    /**
+     * Signals that a ledger has been fenced in a bookie. No more entries can be appended to that ledger.
+     */
+    public static class LedgerFencedAndDeletedException extends BookieException {
+        public LedgerFencedAndDeletedException() {
+            super(Code.LedgerFencedException);
+        }
+    }
+
     /**
      * Signals that a ledger's operation has been rejected by an internal component because of the resource saturation.
      */
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/HandleFactoryImpl.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/HandleFactoryImpl.java
@@ -57,7 +57,7 @@ public LedgerDescriptor getHandle(final long ledgerId, final byte[] masterKey, b
 
         if (handle == null) {
             if (!journalReplay && recentlyFencedAndDeletedLedgers.getIfPresent(ledgerId) != null) {
-                throw BookieException.create(BookieException.Code.LedgerFencedException);
+                throw BookieException.create(BookieException.Code.LedgerFencedAndDeletedException);
             }
             handle = LedgerDescriptor.create(masterKey, ledgerId, ledgerStorage);
             ledgers.putIfAbsent(ledgerId, handle);
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java
@@ -1194,6 +1194,11 @@ private void writeAndFlush(final Channel channel,
                             completion.setOutstanding();
                         }
                     } else {
+                        try {
+                            future.get();
+                        } catch (Exception ex) {
+                            LOG.warn("Failed to request to the bookie: {}", bookieId, ex);
+                        }
                         nettyOpLogger.registerFailedEvent(MathUtils.elapsedNanos(startTime), TimeUnit.NANOSECONDS);
                         errorOut(key);
                     }
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/ReadEntryProcessor.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/ReadEntryProcessor.java
@@ -94,7 +94,7 @@ protected void processPacket() {
                 handleReadResultForFenceRead(fenceResult, data, startTimeNanos);
                 return;
             }
-        } catch (Bookie.NoLedgerException e) {
+        } catch (Bookie.NoLedgerException | BookieException.LedgerFencedAndDeletedException e) {
             if (LOG.isDebugEnabled()) {
                 LOG.debug("Error reading {}", request, e);
             }
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/ReadEntryProcessorV3.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/ReadEntryProcessorV3.java
@@ -215,9 +215,10 @@ protected ReadResponse getReadResponse() {
                 }
             }
             return readEntry(readResponse, entryId, startTimeSw);
-        } catch (Bookie.NoLedgerException e) {
+        } catch (Bookie.NoLedgerException | BookieException.LedgerFencedAndDeletedException e) {
             if (RequestUtils.isFenceRequest(readRequest)) {
-                LOG.info("No ledger found reading entry {} when fencing ledger {}", entryId, ledgerId);
+                LOG.info("No ledger found(or it has been deleted) reading entry {} when fencing ledger {}",
+                    entryId, ledgerId);
             } else if (entryId != BookieProtocol.LAST_ADD_CONFIRMED) {
                 LOG.info("No ledger found while reading entry: {} from ledger: {}", entryId, ledgerId);
             } else if (LOG.isDebugEnabled()) {
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/WriteEntryProcessor.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/WriteEntryProcessor.java
@@ -90,8 +90,8 @@ protected void processPacket() {
         } catch (IOException e) {
             LOG.error("Error writing {}", request, e);
             rc = BookieProtocol.EIO;
-        } catch (BookieException.LedgerFencedException lfe) {
-            LOG.warn("Write attempt on fenced ledger {} by client {}", request.getLedgerId(),
+        } catch (BookieException.LedgerFencedException | BookieException.LedgerFencedAndDeletedException lfe) {
+            LOG.warn("Write attempt on fenced/deleted ledger {} by client {}", request.getLedgerId(),
                     requestHandler.ctx().channel().remoteAddress());
             rc = BookieProtocol.EFENCED;
         } catch (BookieException e) {
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/WriteEntryProcessorV3.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/WriteEntryProcessorV3.java
@@ -136,8 +136,8 @@ public void writeComplete(int rc, long ledgerId, long entryId,
             logger.error("Error writing entry:{} to ledger:{}",
                     entryId, ledgerId, e);
             status = StatusCode.EIO;
-        } catch (BookieException.LedgerFencedException e) {
-            logger.error("Ledger fenced while writing entry:{} to ledger:{}",
+        } catch (BookieException.LedgerFencedException | BookieException.LedgerFencedAndDeletedException e) {
+            logger.error("Ledger fenced/deleted while writing entry:{} to ledger:{}",
                     entryId, ledgerId, e);
             status = StatusCode.EFENCED;
         } catch (BookieException e) {
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/WriteLacProcessorV3.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/WriteLacProcessorV3.java
@@ -105,6 +105,10 @@ public void writeComplete(int rc, long ledgerId, long entryId, BookieId addr, Ob
             requestProcessor.bookie.setExplicitLac(Unpooled.wrappedBuffer(lacToAdd),
                     writeCallback, requestHandler, masterKey);
             status = StatusCode.EOK;
+        } catch (BookieException.LedgerFencedAndDeletedException e) {
+            logger.error("Error saving lac {} for ledger:{}, which has been deleted",
+                    lac, ledgerId, e);
+            status = StatusCode.ENOLEDGER;
         } catch (IOException e) {
             logger.error("Error saving lac {} for ledger:{}",
                     lac, ledgerId, e);
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestFencing.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestFencing.java
@@ -24,6 +24,9 @@
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.UnpooledByteBufAllocator;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.CyclicBarrier;
 import lombok.extern.slf4j.Slf4j;
@@ -33,10 +36,16 @@
 import org.apache.bookkeeper.bookie.SortedLedgerStorage;
 import org.apache.bookkeeper.bookie.storage.ldb.SingleDirectoryDbLedgerStorage;
 import org.apache.bookkeeper.client.BookKeeper.DigestType;
+import org.apache.bookkeeper.client.api.WriteFlag;
 import org.apache.bookkeeper.conf.ClientConfiguration;
 import org.apache.bookkeeper.net.BookieId;
+import org.apache.bookkeeper.proto.BookieClientImpl;
+import org.apache.bookkeeper.proto.BookieProtocol;
+import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks;
+import org.apache.bookkeeper.proto.PerChannelBookieClient;
 import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
 import org.apache.bookkeeper.test.TestStatsProvider;
+import org.apache.bookkeeper.util.ByteBufList;
 import org.awaitility.reflect.WhiteboxImpl;
 import org.junit.Test;
 import org.slf4j.Logger;
@@ -152,6 +161,110 @@ private void triggerGC(Bookie bookie) {
         }
     }
 
+    @Test(timeout = 3000 * 1000)
+    public void testConcurrentFenceAndDeleteLedger() throws Exception {
+        LedgerHandle writeLedger;
+        writeLedger = bkc.createLedger(digestType, "password".getBytes());
+
+        String tmp = "BookKeeper is cool!";
+        long lac = 0;
+        for (int i = 0; i < 10; i++) {
+            long entryId = writeLedger.addEntry(tmp.getBytes());
+            LOG.info("entryId: {}", entryId);
+            lac = entryId;
+        }
+
+        // Fence and delete.
+        final BookieId bookieId = writeLedger.getLedgerMetadata().getEnsembleAt(0).get(0);
+        ClientConfiguration clientConfiguration2 = newClientConfiguration();
+        clientConfiguration2.setUseV2WireProtocol(true);
+        ClientConfiguration clientConfiguration3 = newClientConfiguration();
+        BookKeeperTestClient bkcV2 = new BookKeeperTestClient(clientConfiguration2, new TestStatsProvider());
+        LedgerHandle  writeLedgerV2 = bkcV2.createLedger(digestType, "password".getBytes());
+        BookKeeperTestClient bkcV3 = new BookKeeperTestClient(clientConfiguration3, new TestStatsProvider());
+        LedgerHandle  writeLedgerV3 = bkcV3.createLedger(digestType, "password".getBytes());
+        ReadOnlyLedgerHandle readLedgerV2 =
+                (ReadOnlyLedgerHandle) bkcV2.openLedger(writeLedger.getId(), digestType, "password".getBytes());
+        ReadOnlyLedgerHandle readLedgerV3 =
+                (ReadOnlyLedgerHandle) bkcV3.openLedger(writeLedger.getId(), digestType, "password".getBytes());
+        BookieClientImpl bookieClientV2 = (BookieClientImpl) readLedgerV2.clientCtx.getBookieClient();
+        BookieClientImpl bookieClientV3 = (BookieClientImpl) readLedgerV3.clientCtx.getBookieClient();
+        // Trigger opening connection.
+        CompletableFuture<Integer> obtainV2 = new CompletableFuture<>();
+        bookieClientV2.lookupClient(bookieId).obtain(
+            new BookkeeperInternalCallbacks.GenericCallback<PerChannelBookieClient>() {
+                @Override
+                public void operationComplete(int rc, PerChannelBookieClient result) {
+                    obtainV2.complete(rc);
+                }
+            }, writeLedger.getId());
+        assertEquals(obtainV2.get().intValue(), BKException.Code.OK);
+        CompletableFuture<Integer> obtainV3 = new CompletableFuture<>();
+        bookieClientV3.lookupClient(bookieId).obtain(
+            new BookkeeperInternalCallbacks.GenericCallback<PerChannelBookieClient>() {
+                @Override
+                public void operationComplete(int rc, PerChannelBookieClient result) {
+                    obtainV3.complete(rc);
+                }
+            }, writeLedger.getId());
+        assertEquals(obtainV3.get().intValue(), BKException.Code.OK);
+        bkcV3.deleteLedger(readLedgerV3.ledgerId);
+
+        // Waiting for GC.
+        for (ServerTester server : servers) {
+            triggerGC(server.getServer().getBookie());
+        }
+
+        // Verify: read requests with V2 protocol will receive a NoSuchLedgerException.
+        final byte readEntryFlagFencing = 1;
+        CompletableFuture<Integer> readResV2 = new CompletableFuture<>();
+        bookieClientV2.readEntry(bookieId,
+                writeLedger.getId(), 0, (rc, ledgerId, entryId1, buffer, ctx) -> {
+                    readResV2.complete(rc);
+                }, null, readEntryFlagFencing, readLedgerV2.ledgerKey);
+        assertEquals(BKException.Code.NoSuchLedgerExistsException, readResV2.get().intValue());
+        // Verify: read requests with V3 protocol will receive a NoSuchLedgerException.
+        CompletableFuture<Integer> readResV3 = new CompletableFuture<>();
+        bookieClientV3.readEntry(bookieId,
+                writeLedger.getId(), 0, (rc, ledgerId, entryId1, buffer, ctx) -> {
+                    readResV3.complete(rc);
+                }, null, readEntryFlagFencing, readLedgerV3.ledgerKey);
+        assertEquals(BKException.Code.NoSuchLedgerExistsException, readResV3.get().intValue());
+        // Verify: add requests with V2 protocol will receive a NoSuchLedgerException.
+        log.info("Try to add the next entry: {}:{}", writeLedger.getId(), lac + 1);
+        final ByteBuf dataV2 = UnpooledByteBufAllocator.DEFAULT.heapBuffer();
+        // Combine add request, and rewrite ledgerId of the request.
+        dataV2.writeByte(1);
+        final ByteBuf toSendV2 = (ByteBuf) writeLedgerV2.macManager.computeDigestAndPackageForSending(
+                lac + 1, lac, 1, dataV2, writeLedger.ledgerKey, BookieProtocol.FLAG_NONE);
+        toSendV2.setLong(28, writeLedger.getId());
+        CompletableFuture<Integer> addResV2 = new CompletableFuture<>();
+        bookieClientV2.addEntry(bookieId, writeLedger.getId(), writeLedger.ledgerKey, lac + 1, toSendV2,
+                (rc, ledgerId, entryId1, addr, ctx) -> {
+                    addResV2.complete(rc);
+                }, null, BookieProtocol.FLAG_NONE, false, WriteFlag.NONE);
+        assertEquals(BKException.Code.LedgerFencedException, addResV2.get().intValue());
+        // Verify: read requests with V3 protocol will receive a NoSuchLedgerException.
+        final ByteBuf dataV3 = UnpooledByteBufAllocator.DEFAULT.heapBuffer();
+        dataV3.writeByte(1);
+        // Combine add request, and rewrite ledgerId of the request.
+        final ByteBufList toSendV3 = (ByteBufList) writeLedgerV3.macManager.computeDigestAndPackageForSending(
+                lac + 1, lac, 1, dataV3, writeLedger.ledgerKey, BookieProtocol.FLAG_NONE);
+        toSendV3.getBuffer(0).setLong(0, writeLedger.getId());
+        CompletableFuture<Integer> addResV3 = new CompletableFuture<>();
+        bookieClientV3.addEntry(bookieId, writeLedger.getId(), writeLedger.ledgerKey, lac + 1, toSendV3,
+                (rc, ledgerId, entryId1, addr, ctx) -> {
+                    addResV3.complete(rc);
+                }, null, BookieProtocol.FLAG_NONE, false, WriteFlag.NONE);
+        assertEquals(BKException.Code.LedgerFencedException, addResV3.get().intValue());
+
+        // cleanup.
+        writeLedgerV2.close();
+        writeLedgerV3.close();
+        bkcV2.close();
+        bkcV3.close();
+    }
+
     private static int threadCount = 0;
 
     class LedgerOpenThread extends Thread {

Original file line number	Diff line number	Diff line change
`@@ -57,7 +57,7 @@ public LedgerDescriptor getHandle(final long ledgerId, final byte[] masterKey, b`
`57`	`57`
`58`	`58`	`if (handle == null) {`
`59`	`59`	`if (!journalReplay && recentlyFencedAndDeletedLedgers.getIfPresent(ledgerId) != null) {`
`60`		`- throw BookieException.create(BookieException.Code.LedgerFencedException);`
	`60`	`+ throw BookieException.create(BookieException.Code.LedgerFencedAndDeletedException);`
`61`	`61`	`}`
`62`	`62`	`handle = LedgerDescriptor.create(masterKey, ledgerId, ledgerStorage);`
`63`	`63`	`ledgers.putIfAbsent(ledgerId, handle);`
Original file line number	Diff line number	Diff line change
`@@ -1194,6 +1194,11 @@ private void writeAndFlush(final Channel channel,`
`1194`	`1194`	`completion.setOutstanding();`
`1195`	`1195`	`}`
`1196`	`1196`	`} else {`
	`1197`	`+ try {`
	`1198`	`+ future.get();`
	`1199`	`+ } catch (Exception ex) {`
	`1200`	`+ LOG.warn("Failed to request to the bookie: {}", bookieId, ex);`
	`1201`	`+ }`
`1197`	`1202`	`nettyOpLogger.registerFailedEvent(MathUtils.elapsedNanos(startTime), TimeUnit.NANOSECONDS);`
`1198`	`1203`	`errorOut(key);`
`1199`	`1204`	`}`
Original file line number	Diff line number	Diff line change
`@@ -94,7 +94,7 @@ protected void processPacket() {`
`94`	`94`	`handleReadResultForFenceRead(fenceResult, data, startTimeNanos);`
`95`	`95`	`return;`
`96`	`96`	`}`
`97`		`- } catch (Bookie.NoLedgerException e) {`
	`97`	`+ } catch (Bookie.NoLedgerException \| BookieException.LedgerFencedAndDeletedException e) {`
`98`	`98`	`if (LOG.isDebugEnabled()) {`
`99`	`99`	`LOG.debug("Error reading {}", request, e);`
`100`	`100`	`}`
Original file line number	Diff line number	Diff line change
`@@ -215,9 +215,10 @@ protected ReadResponse getReadResponse() {`
`215`	`215`	`}`
`216`	`216`	`}`
`217`	`217`	`return readEntry(readResponse, entryId, startTimeSw);`
`218`		`- } catch (Bookie.NoLedgerException e) {`
	`218`	`+ } catch (Bookie.NoLedgerException \| BookieException.LedgerFencedAndDeletedException e) {`
`219`	`219`	`if (RequestUtils.isFenceRequest(readRequest)) {`
`220`		`- LOG.info("No ledger found reading entry {} when fencing ledger {}", entryId, ledgerId);`
	`220`	`+ LOG.info("No ledger found(or it has been deleted) reading entry {} when fencing ledger {}",`
	`221`	`+ entryId, ledgerId);`
`221`	`222`	`} else if (entryId != BookieProtocol.LAST_ADD_CONFIRMED) {`
`222`	`223`	`LOG.info("No ledger found while reading entry: {} from ledger: {}", entryId, ledgerId);`
`223`	`224`	`} else if (LOG.isDebugEnabled()) {`