[fix] Fix data lost after when writing ledger and deleting legder exe…

…cute concurrency (#4462)

### Motivation

| step | `BK client 1` | `BK client 2` |
| --- | --- | --- |
| 1 | create ledger `1` | 
| 2 | | open ledger `1` |
| 3 | | delete ledger `1` |
| 4 | write data to ledger `1` |

At the step `4`, the write should fail, but it succeeds. It leads users to assume the data has been written, but it can not be read.

You can reproduce the issue by `testWriteAfterDeleted`

There is a scenario that will lead to Pulsar loss messages

- `broker-2` created a ledger
- `broker-2`'s ZK session is expired, which will lead the topic it owned to be assigned to other brokers
- `broker-0` owned the topic again
  - it will delete the last empty ledger
- consumers connected to `broker-0`
- producers connected to `broker-2`
  - send messages to the topic
- on `broker-2`, the ledger can not be closed due to the ledger metadata has been deleted 

### Changes

Once the ledger is fenced, it can not be wrote anymore.

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

@@ -543,7 +543,7 @@ public void process(int journalVersion, long offset, ByteBuffer recBuff) throws
                             masterKeyCache.put(ledgerId, masterKey);
 
                             // Force to re-insert the master key in ledger storage
-                            handles.getHandle(ledgerId, masterKey);
+                            handles.getHandle(ledgerId, masterKey, true);
                         } else {
                             throw new IOException("Invalid journal. Contains journalKey "
                                     + " but layout version (" + journalVersion
@@ -555,7 +555,7 @@ public void process(int journalVersion, long offset, ByteBuffer recBuff) throws
                             if (key == null) {
                                 key = ledgerStorage.readMasterKey(ledgerId);
                             }
-                            LedgerDescriptor handle = handles.getHandle(ledgerId, key);
+                            LedgerDescriptor handle = handles.getHandle(ledgerId, key, true);
                             handle.setFenced();
                         } else {
                             throw new IOException("Invalid journal. Contains fenceKey "
@@ -573,7 +573,7 @@ public void process(int journalVersion, long offset, ByteBuffer recBuff) throws
                             if (key == null) {
                                 key = ledgerStorage.readMasterKey(ledgerId);
                             }
-                            LedgerDescriptor handle = handles.getHandle(ledgerId, key);
+                            LedgerDescriptor handle = handles.getHandle(ledgerId, key, true);
                             handle.setExplicitLac(explicitLacBuf);
                         } else {
                             throw new IOException("Invalid journal. Contains explicitLAC " + " but layout version ("
@@ -596,7 +596,7 @@ public void process(int journalVersion, long offset, ByteBuffer recBuff) throws
                         if (key == null) {
                             key = ledgerStorage.readMasterKey(ledgerId);
                         }
-                        LedgerDescriptor handle = handles.getHandle(ledgerId, key);
+                        LedgerDescriptor handle = handles.getHandle(ledgerId, key, true);
 
                         recBuff.rewind();
                         handle.addEntry(Unpooled.wrappedBuffer(recBuff));
@@ -933,7 +933,7 @@ LedgerDescriptor getLedgerForEntry(ByteBuf entry, final byte[] masterKey)
             throws IOException, BookieException {
         final long ledgerId = entry.getLong(entry.readerIndex());
 
-        return handles.getHandle(ledgerId, masterKey);
+        return handles.getHandle(ledgerId, masterKey, false);
     }
 
     private Journal getJournal(long ledgerId) {
@@ -1042,7 +1042,7 @@ public void setExplicitLac(ByteBuf entry, WriteCallback writeCallback, Object ct
         ByteBuf explicitLACEntry = null;
         try {
             long ledgerId = entry.getLong(entry.readerIndex());
-            LedgerDescriptor handle = handles.getHandle(ledgerId, masterKey);
+            LedgerDescriptor handle = handles.getHandle(ledgerId, masterKey, false);
             synchronized (handle) {
                 entry.markReaderIndex();
                 handle.setExplicitLac(entry);
@@ -1131,7 +1131,7 @@ public void addEntry(ByteBuf entry, boolean ackBeforeSync, WriteCallback cb, Obj
      */
     public CompletableFuture<Boolean> fenceLedger(long ledgerId, byte[] masterKey)
             throws IOException, BookieException {
-        LedgerDescriptor handle = handles.getHandle(ledgerId, masterKey);
+        LedgerDescriptor handle = handles.getHandle(ledgerId, masterKey, false);
         return handle.fenceAndLogInJournal(getJournal(ledgerId));
     }
 

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

@@ -24,7 +24,7 @@
 import java.io.IOException;
 
 interface HandleFactory {
-    LedgerDescriptor getHandle(long ledgerId, byte[] masterKey)
+    LedgerDescriptor getHandle(long ledgerId, byte[] masterKey, boolean journalReplay)
             throws IOException, BookieException;
 
     LedgerDescriptor getReadOnlyHandle(long ledgerId)

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

@@ -21,14 +21,25 @@
 
 package org.apache.bookkeeper.bookie;
 
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
 import java.io.IOException;
+import java.time.Duration;
 import org.apache.bookkeeper.bookie.LedgerStorage.LedgerDeletionListener;
 import org.apache.bookkeeper.util.collections.ConcurrentLongHashMap;
 
 class HandleFactoryImpl implements HandleFactory, LedgerDeletionListener {
     private final ConcurrentLongHashMap<LedgerDescriptor> ledgers;
     private final ConcurrentLongHashMap<LedgerDescriptor> readOnlyLedgers;
 
+    /**
+     * Once the ledger was marked "fenced" before, the ledger was accessed by multi clients. One client is calling
+     * "delete" now, and other clients may call "write" continuously later. We mark these ledgers can not be written
+     * anymore. And maintains the state for 7 days is safety.
+     */
+    private final Cache<Long, Boolean> recentlyFencedAndDeletedLedgers = CacheBuilder.newBuilder()
+            .expireAfterAccess(Duration.ofDays(7)).build();
+
     final LedgerStorage ledgerStorage;
 
     HandleFactoryImpl(LedgerStorage ledgerStorage) {
@@ -40,10 +51,14 @@ class HandleFactoryImpl implements HandleFactory, LedgerDeletionListener {
     }
 
     @Override
-    public LedgerDescriptor getHandle(final long ledgerId, final byte[] masterKey) throws IOException, BookieException {
+    public LedgerDescriptor getHandle(final long ledgerId, final byte[] masterKey, boolean journalReplay)
+            throws IOException, BookieException {
         LedgerDescriptor handle = ledgers.get(ledgerId);
 
         if (handle == null) {
+            if (!journalReplay && recentlyFencedAndDeletedLedgers.getIfPresent(ledgerId) != null) {
+                throw BookieException.create(BookieException.Code.LedgerFencedException);
+            }
             handle = LedgerDescriptor.create(masterKey, ledgerId, ledgerStorage);
             ledgers.putIfAbsent(ledgerId, handle);
         }
@@ -64,8 +79,22 @@ public LedgerDescriptor getReadOnlyHandle(final long ledgerId) throws IOExceptio
         return handle;
     }
 
+    private void markIfConflictWritingOccurs(long ledgerId) {
+        LedgerDescriptor ledgerDescriptor = ledgers.get(ledgerId);
+        try {
+            if (ledgerDescriptor != null && ledgerDescriptor.isFenced()) {
+                recentlyFencedAndDeletedLedgers.put(ledgerId, true);
+            }
+        } catch (IOException | BookieException ex) {
+            // The ledger is in limbo state.
+            recentlyFencedAndDeletedLedgers.put(ledgerId, true);
+        }
+    }
+
     @Override
     public void ledgerDeleted(long ledgerId) {
+        markIfConflictWritingOccurs(ledgerId);
+        // Do delete.
         ledgers.remove(ledgerId);
         readOnlyLedgers.remove(ledgerId);
     }

bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieJournalTest.java

@@ -452,7 +452,7 @@ public void testV4Journal() throws Exception {
         } catch (Bookie.NoEntryException e) {
             // correct behaviour
         }
-        assertTrue(b.handles.getHandle(1, "testPasswd".getBytes()).isFenced());
+        assertTrue(b.handles.getHandle(1, "testPasswd".getBytes(), false).isFenced());
 
         b.shutdown();
     }
@@ -484,7 +484,7 @@ public void testV5Journal() throws Exception {
         } catch (Bookie.NoEntryException e) {
             // correct behavior
         }
-        assertTrue(b.handles.getHandle(1, "testV5Journal".getBytes()).isFenced());
+        assertTrue(b.handles.getHandle(1, "testV5Journal".getBytes(), false).isFenced());
 
         b.shutdown();
     }

bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestFencing.java

@@ -26,10 +26,18 @@
 
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.CyclicBarrier;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.bookie.Bookie;
+import org.apache.bookkeeper.bookie.InterleavedLedgerStorage;
+import org.apache.bookkeeper.bookie.LedgerStorage;
+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.conf.ClientConfiguration;
 import org.apache.bookkeeper.net.BookieId;
 import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
+import org.apache.bookkeeper.test.TestStatsProvider;
+import org.awaitility.reflect.WhiteboxImpl;
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -38,6 +46,7 @@
  * This unit test tests ledger fencing.
  *
  */
+@Slf4j
 public class TestFencing extends BookKeeperClusterTestCase {
     private static final Logger LOG = LoggerFactory.getLogger(TestFencing.class);
 
@@ -77,6 +86,7 @@ public void testBasicFencing() throws Exception {
             fail("Should have thrown an exception when trying to write");
         } catch (BKException.BKLedgerFencedException e) {
             // correct behaviour
+            log.info("expected a fenced error", e);
         }
 
         /*
@@ -87,6 +97,61 @@ public void testBasicFencing() throws Exception {
                    readlh.getLastAddConfirmed() == writelh.getLastAddConfirmed());
     }
 
+    @Test
+    public void testWriteAfterDeleted() throws Exception {
+        LedgerHandle writeLedger;
+        writeLedger = bkc.createLedger(digestType, "password".getBytes());
+
+        String tmp = "BookKeeper is cool!";
+        for (int i = 0; i < 10; i++) {
+            long entryId = writeLedger.addEntry(tmp.getBytes());
+            LOG.info("entryId: {}", entryId);
+        }
+
+        // Fence and delete.
+        BookKeeperTestClient bkc2 = new BookKeeperTestClient(baseClientConf, new TestStatsProvider());
+        LedgerHandle readLedger = bkc2.openLedger(writeLedger.getId(), digestType, "password".getBytes());
+        bkc2.deleteLedger(readLedger.ledgerId);
+
+        // Waiting for GC.
+        for (ServerTester server : servers) {
+            triggerGC(server.getServer().getBookie());
+        }
+
+        try {
+            long entryId = writeLedger.addEntry(tmp.getBytes());
+            LOG.info("Not expected: entryId: {}", entryId);
+            LOG.error("Should have thrown an exception");
+            fail("Should have thrown an exception when trying to write");
+        } catch (BKException.BKLedgerFencedException e) {
+            log.info("expected a fenced error", e);
+            // correct behaviour
+        }
+
+        /*
+         * Check it has been recovered properly.
+         */
+        assertTrue("Has not recovered correctly: " + readLedger.getLastAddConfirmed()
+                   + " original " + writeLedger.getLastAddConfirmed(),
+                   readLedger.getLastAddConfirmed() == writeLedger.getLastAddConfirmed());
+
+        // cleanup.
+        bkc2.close();
+    }
+
+    private void triggerGC(Bookie bookie) {
+        LedgerStorage ledgerStorage = bookie.getLedgerStorage();
+        if (ledgerStorage instanceof InterleavedLedgerStorage
+                || ledgerStorage instanceof SingleDirectoryDbLedgerStorage) {
+            Runnable gcThread = WhiteboxImpl.getInternalState(ledgerStorage, "gcThread");
+            gcThread.run();
+        } else if (ledgerStorage instanceof SortedLedgerStorage) {
+            Object actLedgerStorage = WhiteboxImpl.getInternalState(ledgerStorage, "interleavedLedgerStorage");
+            Runnable gcThread = WhiteboxImpl.getInternalState(actLedgerStorage, "gcThread");
+            gcThread.run();
+        }
+    }
+
     private static int threadCount = 0;
 
     class LedgerOpenThread extends Thread {