address comments

Author: poorbarcode

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

@@ -195,6 +195,7 @@ private void openWithMetadata(Versioned<LedgerMetadata> versionedMetadata) {
         }
 
         // get the ledger metadata back
+        final boolean needRecovery = !doRecovery || !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
@@ -208,7 +209,7 @@ private void openWithMetadata(Versioned<LedgerMetadata> versionedMetadata) {
             // 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 || keepUpdateMetadata);
+                                          passwd, !needRecovery);
         } catch (GeneralSecurityException e) {
             LOG.error("Security exception while opening ledger: " + ledgerId, e);
             openComplete(BKException.Code.DigestNotInitializedException, null);
@@ -231,6 +232,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 (needRecovery && 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/FullEnsembleDecommissionedTest.java

@@ -138,6 +138,44 @@ public void testOpenedLedgerHandleStillWorkAfterDecommissioning() throws Excepti
         lh.addEntry(data);
         lh.close();
         List<BookieId> originalEnsemble = lh.getLedgerMetadata().getAllEnsembles().get(0L);
+        LedgerHandle readonlyLh = bkc.openLedger(lh.getId(), digestType, PASSWD, false);
+        assertTrue(originalEnsemble.size() == 2);
+
+        startNewBookie();
+        BookieServer newBookieServer3 = serverByIndex(lastBookieIndex());
+        killBookie(originalEnsemble.get(0));
+        waitAutoRecoveryFinished(lh.getId(), originalEnsemble.get(0), newBookieServer3.getBookieId());
+
+        startNewBookie();
+        int newBookieIndex4 = lastBookieIndex();
+        BookieServer newBookieServer4 = serverByIndex(newBookieIndex4);
+        killBookie(originalEnsemble.get(1));
+        waitAutoRecoveryFinished(lh.getId(), originalEnsemble.get(1), newBookieServer4.getBookieId());
+
+        Awaitility.await().untilAsserted(() -> {
+            LedgerEntries ledgerEntries = readonlyLh.read(0, 0);
+            assertNotNull(ledgerEntries);
+            byte[] entryBytes = ledgerEntries.getEntry(0L).getEntryBytes();
+            assertEquals(new String(data), new String(entryBytes));
+            ledgerEntries.close();
+        });
+        readonlyLh.close();
+    }
+
+    /**
+     * The purpose of this test:
+     * 1. Client service open a readonly ledger handle with recovery, which has not been closed yet.
+     * 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. Verify: lhe ledger handle in the client memory keeps updating the ledger ensemble set, and the new read
+     *    request works.
+     */
+    @Test
+    public void testRecoverOpenLedgerHandleStillWorkAfterDecommissioning() throws Exception {
+        LedgerHandle lh = bkc.createLedger(2, 2, digestType, PASSWD);
+        assertTrue(lh.getLedgerMetadata().getAllEnsembles().get(0L).size() == 2);
+        lh.addEntry(data);
+        List<BookieId> originalEnsemble = lh.getLedgerMetadata().getAllEnsembles().get(0L);
         LedgerHandle readonlyLh = bkc.openLedger(lh.getId(), digestType, PASSWD, true);
         assertTrue(originalEnsemble.size() == 2);