Merge branch 'feat/replicator_worker' into '4.14.4_csig' (merge request !24)

feat: support retry in replicator worker

Author: zhiyuanlei

bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/ZkLedgerUnderreplicationManager.java

@@ -636,6 +636,10 @@ public void releaseUnderreplicatedLedger(long ledgerId) throws ReplicationExcept
             }
         } catch (KeeperException.NoNodeException nne) {
             // this is ok
+        } catch (KeeperException.ConnectionLossException | KeeperException.SessionExpiredException ke) {
+            LOG.error("Error deleting underreplicated ledger lock,we need to retry", ke);
+            waitBackOffTime(1000);
+            releaseUnderreplicatedLedger(ledgerId);
         } catch (KeeperException ke) {
             LOG.error("Error deleting underreplicated ledger lock", ke);
             throw new ReplicationException.UnavailableException("Error contacting zookeeper", ke);
@@ -646,6 +650,14 @@ public void releaseUnderreplicatedLedger(long ledgerId) throws ReplicationExcept
         heldLocks.remove(ledgerId);
     }
 
+    private static void waitBackOffTime(long backoffMs) {
+        try {
+            Thread.sleep(backoffMs);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+    }
+
     @Override
     public void close() throws ReplicationException.UnavailableException {
         if (LOG.isDebugEnabled()) {