QuorumCoverage should only count unknown nodes

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

@@ -1407,7 +1407,13 @@ public void readLastConfirmedDataComplete(int rc, DigestManager.RecoveryData dat
                 }
             };
 
-        new ReadLastConfirmedOp(this, clientCtx.getBookieClient(), getCurrentEnsemble(), innercb).initiate();
+        new ReadLastConfirmedOp(clientCtx.getBookieClient(),
+                                distributionSchedule,
+                                macManager,
+                                ledgerId,
+                                getCurrentEnsemble(),
+                                ledgerKey,
+                                innercb).initiate();
     }
 
     /**

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

@@ -93,7 +93,12 @@ LedgerRecoveryOp setEntryListener(ReadEntryListener entryListener) {
     }
 
     public CompletableFuture<LedgerHandle> initiate() {
-        ReadLastConfirmedOp rlcop = new ReadLastConfirmedOp(lh, clientCtx.getBookieClient(), lh.getCurrentEnsemble(),
+        ReadLastConfirmedOp rlcop = new ReadLastConfirmedOp(clientCtx.getBookieClient(),
+                                                            lh.distributionSchedule,
+                                                            lh.macManager,
+                                                            lh.ledgerId,
+                                                            lh.getCurrentEnsemble(),
+                                                            lh.ledgerKey,
                 new ReadLastConfirmedOp.LastConfirmedDataCallback() {
                     public void readLastConfirmedDataComplete(int rc, RecoveryData data) {
                         if (rc == BKException.Code.OK) {

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

@@ -17,6 +17,8 @@
  */
 package org.apache.bookkeeper.client;
 
+import com.google.common.annotations.VisibleForTesting;
+
 import io.netty.buffer.ByteBuf;
 import java.util.List;
 
@@ -25,6 +27,7 @@
 import org.apache.bookkeeper.proto.BookieClient;
 import org.apache.bookkeeper.proto.BookieProtocol;
 import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.ReadEntryCallback;
+import org.apache.bookkeeper.proto.checksum.DigestManager;
 import org.apache.bookkeeper.proto.checksum.DigestManager.RecoveryData;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -35,17 +38,18 @@
  */
 class ReadLastConfirmedOp implements ReadEntryCallback {
     static final Logger LOG = LoggerFactory.getLogger(ReadLastConfirmedOp.class);
-    LedgerHandle lh;
-    BookieClient bookieClient;
-    int numResponsesPending;
-    int numSuccessfulResponse;
-    RecoveryData maxRecoveredData;
-    volatile boolean completed = false;
-    int lastSeenError = BKException.Code.ReadException;
-
-    LastConfirmedDataCallback cb;
-    final DistributionSchedule.QuorumCoverageSet coverageSet;
-    final List<BookieSocketAddress> currentEnsemble;
+    private final long ledgerId;
+    private final byte[] ledgerKey;
+    private final BookieClient bookieClient;
+    private final DigestManager digestManager;
+    private int numResponsesPending;
+    private RecoveryData maxRecoveredData;
+    private volatile boolean completed = false;
+    private int lastSeenError = BKException.Code.ReadException;
+
+    private final LastConfirmedDataCallback cb;
+    private final DistributionSchedule.QuorumCoverageSet coverageSet;
+    private final List<BookieSocketAddress> currentEnsemble;
 
     /**
      * Wrapper to get all recovered data from the request.
@@ -54,22 +58,28 @@ interface LastConfirmedDataCallback {
         void readLastConfirmedDataComplete(int rc, RecoveryData data);
     }
 
-    public ReadLastConfirmedOp(LedgerHandle lh, BookieClient bookieClient,
-                               List<BookieSocketAddress> ensemble, LastConfirmedDataCallback cb) {
+    public ReadLastConfirmedOp(BookieClient bookieClient,
+                               DistributionSchedule schedule,
+                               DigestManager digestManager,
+                               long ledgerId,
+                               List<BookieSocketAddress> ensemble,
+                               byte[] ledgerKey,
+                               LastConfirmedDataCallback cb) {
         this.cb = cb;
         this.bookieClient = bookieClient;
         this.maxRecoveredData = new RecoveryData(LedgerHandle.INVALID_ENTRY_ID, 0);
-        this.lh = lh;
-        this.numSuccessfulResponse = 0;
-        this.numResponsesPending = lh.getLedgerMetadata().getEnsembleSize();
-        this.coverageSet = lh.distributionSchedule.getCoverageSet();
+        this.numResponsesPending = ensemble.size();
+        this.coverageSet = schedule.getCoverageSet();
         this.currentEnsemble = ensemble;
+        this.ledgerId = ledgerId;
+        this.ledgerKey = ledgerKey;
+        this.digestManager = digestManager;
     }
 
     public void initiate() {
         for (int i = 0; i < currentEnsemble.size(); i++) {
             bookieClient.readEntry(currentEnsemble.get(i),
-                                   lh.ledgerId,
+                                   ledgerId,
                                    BookieProtocol.LAST_ADD_CONFIRMED,
                                    this, i, BookieProtocol.FLAG_NONE);
         }
@@ -78,10 +88,10 @@ public void initiate() {
     public void initiateWithFencing() {
         for (int i = 0; i < currentEnsemble.size(); i++) {
             bookieClient.readEntry(currentEnsemble.get(i),
-                                   lh.ledgerId,
+                                   ledgerId,
                                    BookieProtocol.LAST_ADD_CONFIRMED,
                                    this, i, BookieProtocol.FLAG_DO_FENCING,
-                                   lh.ledgerKey);
+                                   ledgerKey);
         }
     }
 
@@ -96,12 +106,11 @@ public synchronized void readEntryComplete(final int rc, final long ledgerId, fi
         boolean heardValidResponse = false;
         if (rc == BKException.Code.OK) {
             try {
-                RecoveryData recoveryData = lh.macManager.verifyDigestAndReturnLastConfirmed(buffer);
+                RecoveryData recoveryData = digestManager.verifyDigestAndReturnLastConfirmed(buffer);
                 if (recoveryData.getLastAddConfirmed() > maxRecoveredData.getLastAddConfirmed()) {
                     maxRecoveredData = recoveryData;
                 }
                 heardValidResponse = true;
-                numSuccessfulResponse++;
             } catch (BKDigestMatchException e) {
                 // Too bad, this bookie didn't give us a valid answer, we
                 // still might be able to recover though so continue
@@ -140,17 +149,15 @@ public synchronized void readEntryComplete(final int rc, final long ledgerId, fi
         }
 
         if (numResponsesPending == 0 && !completed) {
-            int totalExepctedResponse = lh.getLedgerMetadata().getWriteQuorumSize()
-                    - lh.getLedgerMetadata().getAckQuorumSize() + 1;
-            if (numSuccessfulResponse >= totalExepctedResponse) {
-                cb.readLastConfirmedDataComplete(BKException.Code.OK, maxRecoveredData);
-                return;
-            }
-            // Have got all responses back but was still not enough, just fail the operation
-            LOG.error("While readLastConfirmed ledger: {} did not hear success responses from all quorums {}", ledgerId,
-                    lastSeenError);
+            LOG.error("While readLastConfirmed ledger: {} did not hear success responses from all quorums, {}",
+                      ledgerId, coverageSet);
             cb.readLastConfirmedDataComplete(lastSeenError, maxRecoveredData);
         }
 
     }
+
+    @VisibleForTesting
+    synchronized int getNumResponsesPending() {
+        return numResponsesPending;
+    }
 }

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

@@ -373,29 +373,43 @@ public synchronized void addBookie(int bookieIndexHeardFrom, int rc) {
         public synchronized boolean checkCovered() {
             // now check if there are any write quorums, with |ackQuorum| nodes available
             for (int i = 0; i < ensembleSize; i++) {
-                int nodesNotCovered = 0;
-                int nodesOkay = 0;
-                int nodesUninitialized = 0;
+                /* Nodes which have either responded with an error other than NoSuch{Entry,Ledger},
+                   or have not responded at all. We cannot know if these nodes ever accepted a entry. */
+                int nodesUnknown = 0;
+
                 for (int j = 0; j < writeQuorumSize; j++) {
                     int nodeIndex = (i + j) % ensembleSize;
-                    if (covered[nodeIndex] == BKException.Code.OK) {
-                        nodesOkay++;
-                    } else if (covered[nodeIndex] != BKException.Code.NoSuchEntryException
-                            && covered[nodeIndex] != BKException.Code.NoSuchLedgerExistsException) {
-                        nodesNotCovered++;
-                        if (covered[nodeIndex] == BKException.Code.UNINITIALIZED) {
-                            nodesUninitialized++;
-                        }
+                    if (covered[nodeIndex] != BKException.Code.OK
+                        && covered[nodeIndex] != BKException.Code.NoSuchEntryException
+                        && covered[nodeIndex] != BKException.Code.NoSuchLedgerExistsException) {
+                        nodesUnknown++;
                     }
                 }
-                // if we haven't seen any OK responses and there are still nodes not heard from,
-                // let's wait until
-                if (nodesNotCovered >= ackQuorumSize || (nodesOkay == 0 && nodesUninitialized > 0)) {
+
+                /* If nodesUnknown is greater than the ack quorum size, then
+                   it is possible those two unknown nodes accepted an entry which
+                   we do not know about */
+                if (nodesUnknown >= ackQuorumSize) {
                     return false;
                 }
             }
             return true;
         }
+
+        @Override
+        public String toString() {
+            StringBuilder buffer = new StringBuilder();
+            buffer.append("QuorumCoverage(e:").append(ensembleSize)
+                .append(",w:").append(writeQuorumSize)
+                .append(",a:").append(ackQuorumSize)
+                .append(") = [");
+            int i = 0;
+            for (; i < covered.length - 1; i++) {
+                buffer.append(covered[i]).append(", ");
+            }
+            buffer.append(covered[i]).append("]");
+            return buffer.toString();
+        }
     }
 
     @Override

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

@@ -675,8 +675,12 @@ public void operationComplete(int rc, Void result) {
         final AtomicInteger rcHolder = new AtomicInteger(-1234);
         final CountDownLatch doneLatch = new CountDownLatch(1);
 
-        new ReadLastConfirmedOp(readLh, bkc.getBookieClient(),
+        new ReadLastConfirmedOp(bkc.getBookieClient(),
+                                readLh.distributionSchedule,
+                                readLh.macManager,
+                                readLh.ledgerId,
                                 readLh.getLedgerMetadata().getAllEnsembles().lastEntry().getValue(),
+                                readLh.ledgerKey,
                 new ReadLastConfirmedOp.LastConfirmedDataCallback() {
                     @Override
                     public void readLastConfirmedDataComplete(int rc, DigestManager.RecoveryData data) {
@@ -760,8 +764,13 @@ public void testRecoveryWithUnavailableBookie() throws Exception {
     private int readLACFromQuorum(LedgerHandle ledger, int... bookieLACResponse) throws Exception {
         MutableInt responseCode = new MutableInt(100);
         CountDownLatch responseLatch = new CountDownLatch(1);
-        ReadLastConfirmedOp readLCOp = new ReadLastConfirmedOp(ledger, bkc.getBookieClient(),
+        ReadLastConfirmedOp readLCOp = new ReadLastConfirmedOp(
+                bkc.getBookieClient(),
+                ledger.getDistributionSchedule(),
+                ledger.getDigestManager(),
+                ledger.getId(),
                 ledger.getLedgerMetadata().getAllEnsembles().lastEntry().getValue(),
+                ledger.getLedgerKey(),
                 new ReadLastConfirmedOp.LastConfirmedDataCallback() {
                     @Override
                     public void readLastConfirmedDataComplete(int rc, DigestManager.RecoveryData data) {

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

@@ -0,0 +1,111 @@
+/*
+ *
+ * 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.client;
+
+import com.google.common.collect.Lists;
+import io.netty.buffer.UnpooledByteBufAllocator;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import org.apache.bookkeeper.common.util.OrderedExecutor;
+import org.apache.bookkeeper.net.BookieSocketAddress;
+import org.apache.bookkeeper.proto.DataFormats.LedgerMetadataFormat.DigestType;
+import org.apache.bookkeeper.proto.MockBookieClient;
+import org.apache.bookkeeper.proto.checksum.DigestManager;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Tests about ReadLastConfirmedOp.
+ */
+public class ReadLastConfirmedOpTest {
+    private static final Logger log = LoggerFactory.getLogger(ReadLastConfirmedOpTest.class);
+    private final BookieSocketAddress bookie1 = new BookieSocketAddress("bookie1", 3181);
+    private final BookieSocketAddress bookie2 = new BookieSocketAddress("bookie2", 3181);
+
+    OrderedExecutor executor = null;
+
+    @Before
+    public void setup() throws Exception {
+        executor = OrderedExecutor.newBuilder()
+                .name("BookKeeperClientWorker")
+                .numThreads(1)
+                .build();
+    }
+
+    @After
+    public void teardown() throws Exception {
+        if (executor != null) {
+            executor.shutdown();
+        }
+    }
+
+    /**
+     * Test for specific bug that was introduced with dcdd1e88.
+     */
+    @Test
+    public void testBookieFailsAfterLedgerMissingOnFirst() throws Exception {
+        long ledgerId = 0xf00b;
+        List<BookieSocketAddress> ensemble = Lists.newArrayList(bookie1, bookie2);
+        byte[] ledgerKey = new byte[0];
+
+        MockBookieClient bookieClient = new MockBookieClient(executor);
+        DistributionSchedule schedule = new RoundRobinDistributionSchedule(2, 2, 2);
+        DigestManager digestManager = DigestManager.instantiate(ledgerId, ledgerKey,
+                                                                DigestType.CRC32C,
+                                                                UnpooledByteBufAllocator.DEFAULT,
+                                                                true /* useV2 */);
+
+        CompletableFuture<Void> blocker = new CompletableFuture<>();
+        bookieClient.setPreReadHook((bookie, lId, entryId) -> {
+                if (bookie.equals(bookie1)) {
+                    return CompletableFuture.completedFuture(null);
+                } else {
+                    return blocker;
+                }
+            });
+        CompletableFuture<DigestManager.RecoveryData> promise = new CompletableFuture<>();
+        ReadLastConfirmedOp op = new ReadLastConfirmedOp(
+                bookieClient, schedule,
+                digestManager, ledgerId, ensemble,
+                ledgerKey,
+                (rc, data) -> {
+                    if (rc != BKException.Code.OK) {
+                        promise.completeExceptionally(
+                                BKException.create(rc));
+                    } else {
+                        promise.complete(data);
+                    }
+                });
+        op.initiateWithFencing();
+
+        while (op.getNumResponsesPending() > 1) {
+            Thread.sleep(100);
+        }
+        blocker.completeExceptionally(
+                new BKException.BKBookieHandleNotAvailableException());
+        promise.get();
+    }
+}

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

@@ -74,16 +74,6 @@ public void testCoverageSets() {
             }
         }
         assertEquals("Should be no errors", 0, errors);
-
-        RoundRobinDistributionSchedule schedule = new RoundRobinDistributionSchedule(
-            5, 3, 5);
-        DistributionSchedule.QuorumCoverageSet covSet = schedule.getCoverageSet();
-        covSet.addBookie(0, BKException.Code.NoSuchLedgerExistsException);
-        covSet.addBookie(1, BKException.Code.NoSuchEntryException);
-        covSet.addBookie(2, BKException.Code.NoSuchLedgerExistsException);
-        covSet.addBookie(3, BKException.Code.UNINITIALIZED);
-        covSet.addBookie(4, BKException.Code.UNINITIALIZED);
-        assertFalse(covSet.checkCovered());
     }
 
     /**