(@bug W-3315975@) Provide an option to delay auto recovery of lost bookies for planned downtimes or intermittent

Fixing a bug in the earlier implementation where, on a rolling upgrade, we were not honoring 'lostBookieRecoveryDelay'.
I.e. if the first bookie goes down, we honor 'lostBookieRecoveryDelay'. But if that bookie comes up and a different
bookie goes down we would treat it like multiple bookie failure and start the audit. This change fixes this problem
such that when the second bookie goes down it realizes that there is only one bookie down and hence doesn't treat
it like a multiple bookie failure scenario.

Author: rithin-shetty

bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/Auditor.java

@@ -21,6 +21,7 @@
 package org.apache.bookkeeper.replication;
 
 import com.google.common.base.Stopwatch;
+import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
 
 import org.apache.bookkeeper.client.BKException;
@@ -104,6 +105,7 @@ public class Auditor implements BookiesListener {
     private final long auditorPeriodicCheckInterval;
     private final long bookieCheckInterval;
     private volatile boolean isBookieCheckFirstTime;
+    private Set<String> bookiesToBeAudited = Sets.newHashSet();
 
     public Auditor(final String bookieIdentifier, ServerConfiguration conf,
                    ZooKeeper zkc, StatsLogger statsLogger) throws UnavailableException {
@@ -195,46 +197,61 @@ public void run() {
                         waitIfLedgerReplicationDisabled();
 
                         List<String> availableBookies = getAvailableBookies();
-
                         // casting to String, as knownBookies and availableBookies
                         // contains only String values
                         // find new bookies(if any) and update the known bookie list
                         Collection<String> newBookies = CollectionUtils.subtract(
                                 availableBookies, knownBookies);
                         knownBookies.addAll(newBookies);
+                        if (!bookiesToBeAudited.isEmpty() && knownBookies.containsAll(bookiesToBeAudited)) {
+                            // the bookie, which went down earlier and had an audit scheduled for,
+                            // has come up. So let us stop tracking it and cancel the audit. Since
+                            // we allow delaying of audit when there is only one failed bookie,
+                            // bookiesToBeAudited should just have 1 element and hence containsAll
+                            // check should be ok
+                            if (auditTask != null && auditTask.cancel(false)) {
+                                auditTask = null;
+                                numDelayedBookieAuditsCancelled.inc();
+                            }
+                            bookiesToBeAudited.clear();
+                        }
 
                         // find lost bookies(if any)
-                        Collection<String> lostBookies = CollectionUtils.subtract(
-                                knownBookies, availableBookies);
-
-                        if (lostBookies.size() > 0) {
-                            knownBookies.removeAll(lostBookies);
-                            if (conf.getLostBookieRecoveryDelay() == 0 ||
-                                lostBookies.size() > 1 ||
-                                auditTask != null) {
-                                // 1) if more than one bookie is down, start the audit immediately;
-                                // 2) if we had scheduled an audit earlier for a lost bookie and in
-                                // the meantime another bookie goes down, let us not delay recovery
-                                // we cancel the previously scheduled audit before starting the new
-                                // one
-                                if (auditTask != null) {
-                                    if (auditTask.cancel(false)) {
-                                        auditTask = null;
-                                        numDelayedBookieAuditsCancelled.inc();
-                                    }
-                                }
-                                startAudit(false);
-                            } else {
-                                auditTask = executor.schedule( new Runnable() {
-                                    public void run() {
-                                        startAudit(false);
-                                        auditTask = null;
-                                    }
-                                }, conf.getLostBookieRecoveryDelay(), TimeUnit.SECONDS);
-                                numBookieAuditsDelayed.inc();
-                                LOG.info("Delaying the start of bookie audit by " + conf.getLostBookieRecoveryDelay() +
-                                         " seconds");
+                        bookiesToBeAudited.addAll(CollectionUtils.subtract(knownBookies, availableBookies));
+                        if (bookiesToBeAudited.size() == 0) {
+                            return;
+                        }
+                        knownBookies.removeAll(bookiesToBeAudited);
+                        if (conf.getLostBookieRecoveryDelay() == 0) {
+                            startAudit(false);
+                            bookiesToBeAudited.clear();
+                            return;
+                        }
+                        if (bookiesToBeAudited.size() > 1) {
+                            // if more than one bookie is down, start the audit immediately;
+                            LOG.info("Multiple bookie failure; not delaying bookie audit. Bookies lost now: "
+                                     + CollectionUtils.subtract(knownBookies, availableBookies)
+                                     +"; All lost bookies: " + bookiesToBeAudited.toString());
+                            if (auditTask != null && auditTask.cancel(false)) {
+                                auditTask = null;
+                                numDelayedBookieAuditsCancelled.inc();
                             }
+                            startAudit(false);
+                            bookiesToBeAudited.clear();
+                            return;
+                        }
+                        if (auditTask == null) {
+                            // if there is no scheduled audit, schedule one
+                            auditTask = executor.schedule( new Runnable() {
+                                public void run() {
+                                    startAudit(false);
+                                    auditTask = null;
+                                    bookiesToBeAudited.clear();
+                                }
+                            }, conf.getLostBookieRecoveryDelay(), TimeUnit.SECONDS);
+                            numBookieAuditsDelayed.inc();
+                            LOG.info("Delaying bookie audit by " + conf.getLostBookieRecoveryDelay()
+                                     + "secs for " + bookiesToBeAudited.toString());
                         }
                     } catch (BKException bke) {
                         LOG.error("Exception getting bookie list", bke);

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

@@ -436,7 +436,7 @@ public void testDelayedAuditWithMultipleBookieFailures() throws Exception {
         // 2 second grace period for the ledgers to get reported as under replicated
         Thread.sleep(2000);
 
-        // If the following checks pass, it means that auditing happened
+        // If the following checks pass, it means that audit happened
         // within 2 seconds of second bookie going down and it didn't
         // wait for 7 more seconds. Hence the second bookie failure doesn't
         // delay the audit
@@ -449,6 +449,70 @@ public void testDelayedAuditWithMultipleBookieFailures() throws Exception {
                 data.contains(shutdownBookie1) && data.contains(shutdownBookie2));
     }
 
+    /**
+     * Test audit of bookies is delayed during rolling upgrade scenario:
+     * a bookies goes down and comes up, the next bookie go down and up and so on.
+     * At any time only one bookie is down.
+     */
+    @Test(timeout=60000)
+    public void testDelayedAuditWithRollingUpgrade() throws Exception {
+        // wait for the periodic bookie check to finish
+        Thread.sleep(1000);
+
+        // create a ledger with a bunch of entries
+        LedgerHandle lh1 = createAndAddEntriesToLedger();
+        Long ledgerId = lh1.getId();
+        LOG.debug("Created ledger : " + ledgerId);
+        ledgerList.add(ledgerId);
+        lh1.close();
+
+        CountDownLatch underReplicaLatch = registerUrLedgerWatcher(ledgerList.size());
+
+        // wait for 5 seconds before starting the recovery work when a bookie fails
+        baseConf.setLostBookieRecoveryDelay(5);
+
+        // shutdown a non auditor bookie to avoid an election
+        int idx1 = getShutDownNonAuditorBookieIdx("");
+        ServerConfiguration conf1 = bsConfs.get(idx1);
+        String shutdownBookie1 = shutdownBookie(idx1);
+
+        // wait for 2 seconds and there shouldn't be any under replicated ledgers
+        // because we have delayed the start of audit by 5 seconds
+        assertFalse("audit of lost bookie isn't delayed", underReplicaLatch.await(2, TimeUnit.SECONDS));
+        assertEquals("under replicated ledgers identified when it was not expected", 0,
+                urLedgerList.size());
+
+        // restart the bookie we shut down above
+        bs.add(startBookie(conf1));
+
+        // Now to simulate the rolling upgrade, bring down a bookie different from
+        // the one we brought down/up above.
+        String shutdownBookie2 = shutDownNonAuditorBookie(shutdownBookie1);
+
+        // since the first bookie that was brought down/up has come up, there is only
+        // one bookie down at this time. Hence the lost bookie check shouldn't start
+        // immediately; it will start 5 seconds after the second bookie went down
+        assertFalse("audit of lost bookie isn't delayed", underReplicaLatch.await(2, TimeUnit.SECONDS));
+        assertEquals("under replicated ledgers identified when it was not expected", 0,
+                urLedgerList.size());
+
+        // wait for a total of 6 seconds(2+4) for the ledgers to get reported as under replicated
+        Thread.sleep(4000);
+
+        // If the following checks pass, it means that auditing happened
+        // after lostBookieRecoveryDelay during rolling upgrade as expected
+        assertTrue("Ledger is not marked as underreplicated:" + ledgerId,
+                urLedgerList.contains(ledgerId));
+        Map<Long, String> urLedgerData = getUrLedgerData(urLedgerList);
+        String data = urLedgerData.get(ledgerId);
+        assertTrue("Bookie " + shutdownBookie1 + "wrongly listed as missing the ledger: " + data,
+                   !data.contains(shutdownBookie1));
+        assertTrue("Bookie " + shutdownBookie2
+                   + " is not listed in the ledger as missing replicas :" + data,
+                   data.contains(shutdownBookie2));
+        LOG.info("*****************Test Complete");
+    }
+
     /**
      * Wait for ledger to be underreplicated, and to be missing all replicas specified
      */
@@ -596,4 +660,22 @@ private String  shutDownNonAuditorBookie() throws Exception {
         }
         return shutdownBookie(bkIndexDownBookie);
     }
+
+    private int getShutDownNonAuditorBookieIdx(String exclude) throws Exception {
+        // shutdown bookie which is not an auditor
+        int indexOf = bs.indexOf(getAuditorBookie());
+        int bkIndexDownBookie = 0;
+        for (int i = 0; i < bs.size(); i++) {
+            if (i == indexOf || bs.get(i).getLocalAddress().toString().equals(exclude)) {
+                continue;
+            }
+            bkIndexDownBookie = i;
+            break;
+        }
+        return bkIndexDownBookie;
+    }
+
+    private String shutDownNonAuditorBookie(String exclude) throws Exception {
+        return shutdownBookie(getShutDownNonAuditorBookieIdx(exclude));
+    }
 }