[fix] Failed read entries after multiple decommissioning

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java

@@ -1218,14 +1218,52 @@ public void asyncCreateLedgerAdv(final long ledgerId,
      */
     public void asyncOpenLedger(final long lId, final DigestType digestType, final byte[] passwd,
                                 final OpenCallback cb, final Object ctx) {
+        asyncOpenLedger(lId, digestType, passwd, cb, ctx, false);
+    }
+
+    /**
+     * Open existing ledger asynchronously for reading.
+     *
+     * <p>Opening a ledger with this method invokes fencing and recovery on the ledger
+     * if the ledger has not been closed. Fencing will block all other clients from
+     * writing to the ledger. Recovery will make sure that the ledger is closed
+     * before reading from it.
+     *
+     * <p>Recovery also makes sure that any entries which reached one bookie, but not a
+     * quorum, will be replicated to a quorum of bookies. This occurs in cases were
+     * the writer of a ledger crashes after sending a write request to one bookie but
+     * before being able to send it to the rest of the bookies in the quorum.
+     *
+     * <p>If the ledger is already closed, neither fencing nor recovery will be applied.
+     *
+     * @see LedgerHandle#asyncClose
+     *
+     * @param lId
+     *          ledger identifier
+     * @param digestType
+     *          digest type, either MAC or CRC32
+     * @param passwd
+     *          password
+     * @param ctx
+     *          optional control object
+     * @param keepUpdateMetadata
+     *          Whether update ledger metadata if the auto-recover component modified the ledger's ensemble.
+     */
+    public void asyncOpenLedger(final long lId, final DigestType digestType, final byte[] passwd,
+                                final OpenCallback cb, final Object ctx, boolean keepUpdateMetadata) {
         closeLock.readLock().lock();
         try {
             if (closed) {
                 cb.openComplete(BKException.Code.ClientClosedException, null, ctx);
                 return;
             }
-            new LedgerOpenOp(BookKeeper.this, clientStats,
-                             lId, digestType, passwd, cb, ctx).initiate();
+            LedgerOpenOp ledgerOpenOp = new LedgerOpenOp(BookKeeper.this, clientStats,
+                 lId, digestType, passwd, cb, ctx);
+            if (keepUpdateMetadata) {
+                ledgerOpenOp.initiateWithKeepUpdateMetadata();
+            } else {
+                ledgerOpenOp.initiate();
+            }
         } finally {
             closeLock.readLock().unlock();
         }
@@ -1293,13 +1331,36 @@ public void asyncOpenLedgerNoRecovery(final long lId, final DigestType digestTyp
      */
     public LedgerHandle openLedger(long lId, DigestType digestType, byte[] passwd)
             throws BKException, InterruptedException {
+        return openLedger(lId, digestType, passwd, false);
+    }
+
+
+    /**
+     * Synchronous open ledger call.
+     *
+     * @see #asyncOpenLedger
+     * @param lId
+     *          ledger identifier
+     * @param digestType
+     *          digest type, either MAC or CRC32
+     * @param passwd
+     *          password
+     *
+     * @param keepUpdateMetadata
+     *          Whether update ledger metadata if the auto-recover component modified the ledger's ensemble.
+     * @return a handle to the open ledger
+     * @throws InterruptedException
+     * @throws BKException
+     */
+    public LedgerHandle openLedger(long lId, DigestType digestType, byte[] passwd, boolean keepUpdateMetadata)
+            throws BKException, InterruptedException {
         CompletableFuture<LedgerHandle> future = new CompletableFuture<>();
         SyncOpenCallback result = new SyncOpenCallback(future);
 
         /*
          * Calls async open ledger
          */
-        asyncOpenLedger(lId, digestType, passwd, result, null);
+        asyncOpenLedger(lId, digestType, passwd, result, null, keepUpdateMetadata);
 
         return SyncCallbackUtils.waitForResult(future);
     }

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerOpenOp.java

@@ -57,6 +57,18 @@ class LedgerOpenOp {
     ReadOnlyLedgerHandle lh;
     final byte[] passwd;
     boolean doRecovery = true;
+    // The ledger metadata may be modified even if it has been closed, because the auto-recovery component may rewrite
+    // the ledger's metadata. Keep receiving a notification from ZK to avoid the following issue: an opened ledger
+    // handle in memory still accesses to a BK instance who has been decommissioned. The issue that solved happens as
+    // follows:
+    // 1. Client service open a readonly ledger handle, which has been closed.
+    // 2. All BKs that relates to the ledger have been decommissioned.
+    // 3. Auto recovery component moved the data into other BK instances who is alive.
+    // 4. The ledger handle in the client memory keeps connects to the BKs who in the original ensemble set, and the
+    //    connection will always fail.
+    // For minimum modification, to add a new configuration named "keepUpdateMetadata", users can use the
+    // new API to create a readonly ledger handle that will auto-updates metadata.
+    boolean keepUpdateMetadata = false;
     boolean administrativeOpen = false;
     long startTime;
     final OpStatsLogger openOpLogger;
@@ -126,6 +138,15 @@ public void initiateWithoutRecovery() {
         initiate();
     }
 
+    /**
+     * Different with {@link #initiate()}, the method keep update metadata once the auto-recover component modified
+     * the ensemble.
+     */
+    public void initiateWithKeepUpdateMetadata() {
+        this.keepUpdateMetadata = true;
+        initiate();
+    }
+
     private CompletableFuture<Void> closeLedgerHandleAsync() {
         if (lh != null) {
             return lh.closeAsync();
@@ -174,9 +195,25 @@ private void openWithMetadata(Versioned<LedgerMetadata> versionedMetadata) {
         }
 
         // get the ledger metadata back
+        // The cases that need to register listener immediately are:
+        // 1. The ledger is not in recovery opening, which is the original case.
+        // 2. The ledger is closed and need to keep update metadata. There is other cases that do not need to
+        //   register listener. e.g. The ledger is opening by Auto-Recovery component.
+        final boolean watchImmediately = !doRecovery || (keepUpdateMetadata && metadata.isClosed());
         try {
+            // The ledger metadata may be modified even if it has been closed, because the auto-recovery component may
+            // rewrite the ledger's metadata. Keep receiving a notification from ZK to avoid the following issue: an
+            // opened ledger handle in memory still accesses to a BK instance who has been decommissioned. The issue
+            // that solved happens as follows:
+            // 1. Client service open a readonly ledger handle, which has been closed.
+            // 2. All BKs that relates to the ledger have been decommissioned.
+            // 3. Auto recovery component moved the data into other BK instances who is alive.
+            // 4. The ledger handle in the client memory keeps connects to the BKs who in the original ensemble set,
+            //    and the connection will always fail.
+            // Therefore, if a user needs to the feature that update metadata automatically, he will set
+            // "keepUpdateMetadata" to "true",
             lh = new ReadOnlyLedgerHandle(bk.getClientCtx(), ledgerId, versionedMetadata, digestType,
-                                          passwd, !doRecovery);
+                                          passwd, watchImmediately);
         } catch (GeneralSecurityException e) {
             LOG.error("Security exception while opening ledger: " + ledgerId, e);
             openComplete(BKException.Code.DigestNotInitializedException, null);
@@ -199,6 +236,9 @@ private void openWithMetadata(Versioned<LedgerMetadata> versionedMetadata) {
                 public void safeOperationComplete(int rc, Void result) {
                     if (rc == BKException.Code.OK) {
                         openComplete(BKException.Code.OK, lh);
+                        if (!watchImmediately && keepUpdateMetadata) {
+                            lh.registerLedgerMetadataListener();
+                        }
                     } else {
                         closeLedgerHandleAsync().whenComplete((ignore, ex) -> {
                             if (ex != null) {

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ReadOnlyLedgerHandle.java

@@ -95,14 +95,18 @@ public String toString() {
     ReadOnlyLedgerHandle(ClientContext clientCtx,
                          long ledgerId, Versioned<LedgerMetadata> metadata,
                          BookKeeper.DigestType digestType, byte[] password,
-                         boolean watch)
+                         boolean watchImmediately)
             throws GeneralSecurityException, NumberFormatException {
         super(clientCtx, ledgerId, metadata, digestType, password, WriteFlag.NONE);
-        if (watch) {
-            clientCtx.getLedgerManager().registerLedgerMetadataListener(ledgerId, this);
+        if (watchImmediately) {
+            registerLedgerMetadataListener();
         }
     }
 
+    void registerLedgerMetadataListener() {
+        clientCtx.getLedgerManager().registerLedgerMetadataListener(ledgerId, this);
+    }
+
     @Override
     public void close()
             throws InterruptedException, BKException {

bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/BookieAutoRecoveryTest.java

@@ -96,6 +96,7 @@ public BookieAutoRecoveryTest() throws IOException, KeeperException,
 
     @Override
     public void setUp() throws Exception {
+        LOG.info("Start setUp");
         super.setUp();
         baseConf.setMetadataServiceUri(zkUtil.getMetadataServiceUri());
         baseClientConf.setMetadataServiceUri(zkUtil.getMetadataServiceUri());
@@ -117,10 +118,12 @@ public void setUp() throws Exception {
         mFactory = metadataClientDriver.getLedgerManagerFactory();
         underReplicationManager = mFactory.newLedgerUnderreplicationManager();
         ledgerManager = mFactory.newLedgerManager();
+        LOG.info("Finished setUp");
     }
 
     @Override
     public void tearDown() throws Exception {
+        LOG.info("Start tearDown");
         super.tearDown();
 
         if (null != underReplicationManager) {
@@ -138,6 +141,7 @@ public void tearDown() throws Exception {
         if (null != scheduler) {
             scheduler.shutdown();
         }
+        LOG.info("Finished tearDown");
     }
 
     /**
@@ -146,6 +150,7 @@ public void tearDown() throws Exception {
      */
     @Test
     public void testOpenLedgers() throws Exception {
+        LOG.info("Start testOpenLedgers");
         List<LedgerHandle> listOfLedgerHandle = createLedgersAndAddEntries(1, 5);
         LedgerHandle lh = listOfLedgerHandle.get(0);
         int ledgerReplicaIndex = 0;
@@ -186,6 +191,7 @@ public void testOpenLedgers() throws Exception {
 
         verifyLedgerEnsembleMetadataAfterReplication(newBookieServer,
                 listOfLedgerHandle.get(0), ledgerReplicaIndex);
+        LOG.info("Finished testOpenLedgers");
     }
 
     /**
@@ -194,6 +200,7 @@ public void testOpenLedgers() throws Exception {
      */
     @Test
     public void testClosedLedgers() throws Exception {
+        LOG.info("Start testClosedLedgers");
         List<Integer> listOfReplicaIndex = new ArrayList<Integer>();
         List<LedgerHandle> listOfLedgerHandle = createLedgersAndAddEntries(1, 5);
         closeLedgers(listOfLedgerHandle);
@@ -247,6 +254,7 @@ public void testClosedLedgers() throws Exception {
                     listOfLedgerHandle.get(index),
                     listOfReplicaIndex.get(index));
         }
+        LOG.info("Finished testClosedLedgers");
     }
 
     /**
@@ -256,6 +264,7 @@ public void testClosedLedgers() throws Exception {
      */
     @Test
     public void testStopWhileReplicationInProgress() throws Exception {
+        LOG.info("Start testStopWhileReplicationInProgress");
         int numberOfLedgers = 2;
         List<Integer> listOfReplicaIndex = new ArrayList<Integer>();
         List<LedgerHandle> listOfLedgerHandle = createLedgersAndAddEntries(
@@ -327,6 +336,7 @@ public void testStopWhileReplicationInProgress() throws Exception {
                     listOfLedgerHandle.get(index),
                     listOfReplicaIndex.get(index));
         }
+        LOG.info("Finished testStopWhileReplicationInProgress");
     }
 
     /**
@@ -336,6 +346,7 @@ public void testStopWhileReplicationInProgress() throws Exception {
      */
     @Test
     public void testNoSuchLedgerExists() throws Exception {
+        LOG.info("Start testNoSuchLedgerExists");
         List<LedgerHandle> listOfLedgerHandle = createLedgersAndAddEntries(2, 5);
         CountDownLatch latch = new CountDownLatch(listOfLedgerHandle.size());
         for (LedgerHandle lh : listOfLedgerHandle) {
@@ -372,6 +383,7 @@ public void testNoSuchLedgerExists() throws Exception {
             assertNull("UrLedger still exists after rereplication",
                     watchUrLedgerNode(getUrLedgerZNode(lh), latch));
         }
+        LOG.info("Finished testNoSuchLedgerExists");
     }
 
     /**
@@ -380,6 +392,7 @@ public void testNoSuchLedgerExists() throws Exception {
      */
     @Test
     public void testEmptyLedgerLosesQuorumEventually() throws Exception {
+        LOG.info("Start testEmptyLedgerLosesQuorumEventually");
         LedgerHandle lh = bkc.createLedger(3, 2, 2, DigestType.CRC32, PASSWD);
         CountDownLatch latch = new CountDownLatch(1);
         String urZNode = getUrLedgerZNode(lh);
@@ -420,6 +433,7 @@ public void testEmptyLedgerLosesQuorumEventually() throws Exception {
 
         // should be able to open ledger without issue
         bkc.openLedger(lh.getId(), DigestType.CRC32, PASSWD);
+        LOG.info("Finished testEmptyLedgerLosesQuorumEventually");
     }
 
     /**
@@ -429,6 +443,7 @@ public void testEmptyLedgerLosesQuorumEventually() throws Exception {
     @Test
     public void testLedgerMetadataContainsIpAddressAsBookieID()
             throws Exception {
+        LOG.info("Start testLedgerMetadataContainsIpAddressAsBookieID");
         stopBKCluster();
         bkc = new BookKeeperTestClient(baseClientConf);
         // start bookie with useHostNameAsBookieID=false, as old bookie
@@ -494,7 +509,7 @@ public void testLedgerMetadataContainsIpAddressAsBookieID()
 
         verifyLedgerEnsembleMetadataAfterReplication(newBookieServer,
                 listOfLedgerHandle.get(0), ledgerReplicaIndex);
-
+        LOG.info("Finished testLedgerMetadataContainsIpAddressAsBookieID");
     }
 
     /**
@@ -504,6 +519,7 @@ public void testLedgerMetadataContainsIpAddressAsBookieID()
     @Test
     public void testLedgerMetadataContainsHostNameAsBookieID()
             throws Exception {
+        LOG.info("Start testLedgerMetadataContainsHostNameAsBookieID");
         stopBKCluster();
 
         bkc = new BookKeeperTestClient(baseClientConf);
@@ -572,7 +588,7 @@ public void testLedgerMetadataContainsHostNameAsBookieID()
 
         verifyLedgerEnsembleMetadataAfterReplication(newBookieServer,
                 listOfLedgerHandle.get(0), ledgerReplicaIndex);
-
+        LOG.info("Finished testLedgerMetadataContainsHostNameAsBookieID");
     }
 
     private int getReplicaIndexInLedger(LedgerHandle lh, BookieId replicaToKill) {
@@ -634,13 +650,13 @@ private Stat watchUrLedgerNode(final String znode,
             @Override
             public void process(WatchedEvent event) {
                 if (event.getType() == EventType.NodeDeleted) {
-                    LOG.info("Received Ledger rereplication completion event :"
-                            + event.getType());
+                    LOG.info("Received Ledger replication completion. event : {}, path: {}, latchCount: {}",
+                            event.getType(), event.getPath(), latch.getCount());
                     latch.countDown();
                 }
                 if (event.getType() == EventType.NodeCreated) {
-                    LOG.info("Received urLedger publishing event :"
-                            + event.getType());
+                    LOG.info("Received urLedger publishing event: {}, path: {}, latchCount: {}",
+                            event.getType(), event.getPath(), latch.getCount());
                     latch.countDown();
                 }
             }