content: /*
   * 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 static org.apache.bookkeeper.client.LedgerHandle.INVALID_ENTRY_ID;
  import static org.apache.bookkeeper.replication.ReplicationStats.NUM_BYTES_READ;
  import static org.apache.bookkeeper.replication.ReplicationStats.NUM_BYTES_WRITTEN;
  import static org.apache.bookkeeper.replication.ReplicationStats.NUM_ENTRIES_READ;
  import static org.apache.bookkeeper.replication.ReplicationStats.NUM_ENTRIES_WRITTEN;
  import static org.apache.bookkeeper.replication.ReplicationStats.READ_DATA_LATENCY;
  import static org.apache.bookkeeper.replication.ReplicationStats.REPLICATION_WORKER_SCOPE;
  import static org.apache.bookkeeper.replication.ReplicationStats.WRITE_DATA_LATENCY;
  
  import com.google.common.util.concurrent.RateLimiter;
  import io.netty.buffer.ByteBuf;
  import io.netty.buffer.Unpooled;
  import io.netty.util.ReferenceCounted;
  import java.util.Enumeration;
  import java.util.HashSet;
  import java.util.Iterator;
  import java.util.LinkedList;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.AtomicBoolean;
  import java.util.concurrent.atomic.AtomicInteger;
  import java.util.function.BiConsumer;
  import java.util.stream.Collectors;
  import org.apache.bookkeeper.client.AsyncCallback.ReadCallback;
  import org.apache.bookkeeper.client.api.WriteFlag;
  import org.apache.bookkeeper.common.util.MathUtils;
  import org.apache.bookkeeper.conf.ClientConfiguration;
  import org.apache.bookkeeper.meta.LedgerManager;
  import org.apache.bookkeeper.net.BookieId;
  import org.apache.bookkeeper.proto.BookieProtocol;
  import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.MultiCallback;
  import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.WriteCallback;
  import org.apache.bookkeeper.stats.Counter;
  import org.apache.bookkeeper.stats.NullStatsLogger;
  import org.apache.bookkeeper.stats.OpStatsLogger;
  import org.apache.bookkeeper.stats.StatsLogger;
  import org.apache.bookkeeper.stats.annotations.StatsDoc;
  import org.apache.bookkeeper.util.ByteBufList;
  import org.apache.zookeeper.AsyncCallback;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  /**
   * This is the helper class for replicating the fragments from one bookie to
   * another.
   */
  @StatsDoc(
      name = REPLICATION_WORKER_SCOPE,
      help = "Ledger fragment replicator related stats"
  )
  public class LedgerFragmentReplicator {
  
      // BookKeeper instance
      private BookKeeper bkc;
      private StatsLogger statsLogger;
      @StatsDoc(
          name = NUM_ENTRIES_READ,
          help = "Number of entries read by the replicator"
      )
      private final Counter numEntriesRead;
      @StatsDoc(
          name = NUM_BYTES_READ,
          help = "The distribution of size of entries read by the replicator"
      )
      private final OpStatsLogger numBytesRead;
      @StatsDoc(
          name = NUM_ENTRIES_WRITTEN,
          help = "Number of entries written by the replicator"
      )
      private final Counter numEntriesWritten;
      @StatsDoc(
          name = NUM_BYTES_WRITTEN,
          help = "The distribution of size of entries written by the replicator"
      )
      private final OpStatsLogger numBytesWritten;
      @StatsDoc(
              name = READ_DATA_LATENCY,
              help = "The distribution of latency of read entries by the replicator"
      )
      private final OpStatsLogger readDataLatency;
      @StatsDoc(
              name = WRITE_DATA_LATENCY,
              help = "The distribution of latency of write entries by the replicator"
      )
      private final OpStatsLogger writeDataLatency;
  
      protected Throttler replicationThrottle = null;
  
      private AtomicInteger averageEntrySize;
  
      private static final int INITIAL_AVERAGE_ENTRY_SIZE = 1024;
      private static final double AVERAGE_ENTRY_SIZE_RATIO = 0.8;
      private ClientConfiguration conf;
  
      public LedgerFragmentReplicator(BookKeeper bkc, StatsLogger statsLogger, ClientConfiguration conf) {
          this.bkc = bkc;
          this.statsLogger = statsLogger;
          numEntriesRead = this.statsLogger.getCounter(NUM_ENTRIES_READ);
          numBytesRead = this.statsLogger.getOpStatsLogger(NUM_BYTES_READ);
          numEntriesWritten = this.statsLogger.getCounter(NUM_ENTRIES_WRITTEN);
          numBytesWritten = this.statsLogger.getOpStatsLogger(NUM_BYTES_WRITTEN);
          readDataLatency = this.statsLogger.getOpStatsLogger(READ_DATA_LATENCY);
          writeDataLatency = this.statsLogger.getOpStatsLogger(WRITE_DATA_LATENCY);
          if (conf.getReplicationRateByBytes() > 0) {
              this.replicationThrottle = new Throttler(conf.getReplicationRateByBytes());
          }
          averageEntrySize = new AtomicInteger(INITIAL_AVERAGE_ENTRY_SIZE);
          this.conf = conf;
      }
  
      public LedgerFragmentReplicator(BookKeeper bkc, ClientConfiguration conf) {
          this(bkc, NullStatsLogger.INSTANCE, conf);
      }
  
      private static final Logger LOG = LoggerFactory
              .getLogger(LedgerFragmentReplicator.class);
  
      private void replicateFragmentInternal(final LedgerHandle lh,
              final LedgerFragment lf,
              final AsyncCallback.VoidCallback ledgerFragmentMcb,
              final Set<BookieId> newBookies,
              final BiConsumer<Long, Long> onReadEntryFailureCallback) throws InterruptedException {
          if (!lf.isClosed()) {
              LOG.error("Trying to replicate an unclosed fragment;"
                        + " This is not safe {}", lf);
              ledgerFragmentMcb.processResult(BKException.Code.UnclosedFragmentException,
                                              null, null);
              return;
          }
          Long startEntryId = lf.getFirstStoredEntryId();
          Long endEntryId = lf.getLastStoredEntryId();
  
          /*
           * if startEntryId is INVALID_ENTRY_ID then endEntryId should be
           * INVALID_ENTRY_ID and viceversa.
           */
          if (startEntryId == INVALID_ENTRY_ID ^ endEntryId == INVALID_ENTRY_ID) {
              LOG.error("For LedgerFragment: {}, seeing inconsistent firstStoredEntryId: {} and lastStoredEntryId: {}",
                      lf, startEntryId, endEntryId);
              assert false;
          }
  
          if (startEntryId > endEntryId || endEntryId <= INVALID_ENTRY_ID) {
              // for open ledger which there is no entry, the start entry id is 0,
              // the end entry id is -1.
              // we can return immediately to trigger forward read
              ledgerFragmentMcb.processResult(BKException.Code.OK, null, null);
              return;
          }
  
          /*
           * Now asynchronously replicate all of the entries for the ledger
           * fragment that were on the dead bookie.
           */
          int entriesToReplicateCnt = (int) (endEntryId - startEntryId + 1);
          MultiCallback ledgerFragmentEntryMcb = new MultiCallback(
                  entriesToReplicateCnt, ledgerFragmentMcb, null, BKException.Code.OK,
                  BKException.Code.LedgerRecoveryException);
          if (this.replicationThrottle != null) {
              this.replicationThrottle.resetRate(this.conf.getReplicationRateByBytes());
          }
  
          if (conf.isRecoveryBatchReadEnabled()
                  && conf.getUseV2WireProtocol()
                  && conf.isBatchReadEnabled()
                  && lh.getLedgerMetadata().getEnsembleSize() == lh.getLedgerMetadata().getWriteQuorumSize()) {
              batchRecoverLedgerFragmentEntry(startEntryId, endEntryId, lh, ledgerFragmentEntryMcb,
                      newBookies, onReadEntryFailureCallback);
  
          } else {
              /*
               * Add all the entries to entriesToReplicate list from
               * firstStoredEntryId to lastStoredEntryID.
               */
              List<Long> entriesToReplicate = new LinkedList<Long>();
              long lastStoredEntryId = lf.getLastStoredEntryId();
              for (long i = lf.getFirstStoredEntryId(); i <= lastStoredEntryId; i++) {
                  entriesToReplicate.add(i);
              }
              for (final Long entryId : entriesToReplicate) {
                  recoverLedgerFragmentEntry(entryId, lh, ledgerFragmentEntryMcb,
                          newBookies, onReadEntryFailureCallback);
              }
          }
  
      }
  
      /**
       * This method replicate a ledger fragment which is a contiguous portion of
       * a ledger that was stored in an ensemble that included the failed bookie.
       * It will Splits the fragment into multiple sub fragments by keeping the
       * max entries up to the configured value of rereplicationEntryBatchSize and
       * then it re-replicates that batched entry fragments one by one. After
       * re-replication of all batched entry fragments, it will update the
       * ensemble info with new Bookie once
       *
       * @param lh
       *            LedgerHandle for the ledger
       * @param lf
       *            LedgerFragment to replicate
       * @param ledgerFragmentMcb
       *            MultiCallback to invoke once we've recovered the current
       *            ledger fragment.
       * @param targetBookieAddresses
       *            New bookies we want to use to recover and replicate the ledger
       *            entries that were stored on the failed bookie.
       */
      void replicate(final LedgerHandle lh, final LedgerFragment lf,
              final AsyncCallback.VoidCallback ledgerFragmentMcb,
              final Set<BookieId> targetBookieAddresses,
              final BiConsumer<Long, Long> onReadEntryFailureCallback)
              throws InterruptedException {
          Set<LedgerFragment> partitionedFragments = splitIntoSubFragments(lh, lf,
                  bkc.getConf().getRereplicationEntryBatchSize());
          LOG.info("Replicating fragment {} in {} sub fragments.",
                  lf, partitionedFragments.size());
          replicateNextBatch(lh, partitionedFragments.iterator(),
                  ledgerFragmentMcb, targetBookieAddresses, onReadEntryFailureCallback);
      }
  
      /**
       * Replicate the batched entry fragments one after other.
       */
      private void replicateNextBatch(final LedgerHandle lh,
              final Iterator<LedgerFragment> fragments,
              final AsyncCallback.VoidCallback ledgerFragmentMcb,
              final Set<BookieId> targetBookieAddresses,
              final BiConsumer<Long, Long> onReadEntryFailureCallback) {
          if (fragments.hasNext()) {
              try {
                  replicateFragmentInternal(lh, fragments.next(),
                          new AsyncCallback.VoidCallback() {
                              @Override
                              public void processResult(int rc, String v, Object ctx) {
                                  if (rc != BKException.Code.OK) {
                                      ledgerFragmentMcb.processResult(rc, null,
                                              null);
                                  } else {
                                      replicateNextBatch(lh, fragments,
                                              ledgerFragmentMcb,
                                              targetBookieAddresses,
                                              onReadEntryFailureCallback);
                                  }
                              }
  
                          }, targetBookieAddresses, onReadEntryFailureCallback);
              } catch (InterruptedException e) {
                  ledgerFragmentMcb.processResult(
                          BKException.Code.InterruptedException, null, null);
                  Thread.currentThread().interrupt();
              }
          } else {
              ledgerFragmentMcb.processResult(BKException.Code.OK, null, null);
          }
      }
  
      /**
       * Split the full fragment into batched entry fragments by keeping
       * rereplicationEntryBatchSize of entries in each one and can treat them as
       * sub fragments.
       */
      static Set<LedgerFragment> splitIntoSubFragments(LedgerHandle lh,
              LedgerFragment ledgerFragment, long rereplicationEntryBatchSize) {
          Set<LedgerFragment> fragments = new HashSet<LedgerFragment>();
          if (rereplicationEntryBatchSize <= 0) {
              // rereplicationEntryBatchSize can not be 0 or less than 0,
              // returning with the current fragment
              fragments.add(ledgerFragment);
              return fragments;
          }
  
          long firstEntryId = ledgerFragment.getFirstStoredEntryId();
          long lastEntryId = ledgerFragment.getLastStoredEntryId();
  
          /*
           * if firstEntryId is INVALID_ENTRY_ID then lastEntryId should be
           * INVALID_ENTRY_ID and viceversa.
           */
          if (firstEntryId == INVALID_ENTRY_ID ^ lastEntryId == INVALID_ENTRY_ID) {
              LOG.error("For LedgerFragment: {}, seeing inconsistent firstStoredEntryId: {} and lastStoredEntryId: {}",
                      ledgerFragment, firstEntryId, lastEntryId);
              assert false;
          }
  
          long numberOfEntriesToReplicate = firstEntryId == INVALID_ENTRY_ID ? 0 : (lastEntryId - firstEntryId) + 1;
          long splitsWithFullEntries = numberOfEntriesToReplicate
                  / rereplicationEntryBatchSize;
  
          if (splitsWithFullEntries == 0) {// only one fragment
              fragments.add(ledgerFragment);
              return fragments;
          }
  
          long fragmentSplitLastEntry = 0;
          for (int i = 0; i < splitsWithFullEntries; i++) {
              fragmentSplitLastEntry = (firstEntryId + rereplicationEntryBatchSize) - 1;
              fragments.add(new LedgerFragment(lh, firstEntryId,
                      fragmentSplitLastEntry, ledgerFragment.getBookiesIndexes()));
              firstEntryId = fragmentSplitLastEntry + 1;
          }
  
          long lastSplitWithPartialEntries = numberOfEntriesToReplicate
                  % rereplicationEntryBatchSize;
          if (lastSplitWithPartialEntries > 0) {
              fragments.add(new LedgerFragment(lh, firstEntryId, firstEntryId
                      + lastSplitWithPartialEntries - 1, ledgerFragment
                      .getBookiesIndexes()));
          }
          return fragments;
      }
  
      /**
       * This method asynchronously recovers a specific ledger entry by reading
       * the values via the BookKeeper Client (which would read it from the other
       * replicas) and then writing it to the chosen new bookie.
       *
       * @param entryId
       *            Ledger Entry ID to recover.
       * @param lh
       *            LedgerHandle for the ledger
       * @param ledgerFragmentEntryMcb
       *            MultiCallback to invoke once we've recovered the current
       *            ledger entry.
       * @param newBookies
       *            New bookies we want to use to recover and replicate the ledger
       *            entries that were stored on the failed bookie.
       */
      void recoverLedgerFragmentEntry(final Long entryId,
              final LedgerHandle lh,
              final AsyncCallback.VoidCallback ledgerFragmentEntryMcb,
              final Set<BookieId> newBookies,
              final BiConsumer<Long, Long> onReadEntryFailureCallback) throws InterruptedException {
          final long ledgerId = lh.getId();
          final AtomicInteger numCompleted = new AtomicInteger(0);
          final AtomicBoolean completed = new AtomicBoolean(false);
  
          if (replicationThrottle != null) {
              replicationThrottle.acquire(averageEntrySize.get());
          }
  
          final WriteCallback multiWriteCallback = new WriteCallback() {
              @Override
              public void writeComplete(int rc, long ledgerId, long entryId, BookieId addr, Object ctx) {
                  if (rc != BKException.Code.OK) {
                      LOG.error("BK error writing entry for ledgerId: {}, entryId: {}, bookie: {}",
                              ledgerId, entryId, addr, BKException.create(rc));
                      if (completed.compareAndSet(false, true)) {
                          ledgerFragmentEntryMcb.processResult(rc, null, null);
                      }
                  } else {
                      numEntriesWritten.inc();
                      if (ctx instanceof Long) {
                          numBytesWritten.registerSuccessfulValue((Long) ctx);
                      }
                      if (LOG.isDebugEnabled()) {
                          LOG.debug("Success writing ledger id {}, entry id {} to a new bookie {}!",
                                  ledgerId, entryId, addr);
                      }
                      if (numCompleted.incrementAndGet() == newBookies.size() && completed.compareAndSet(false, true)) {
                          ledgerFragmentEntryMcb.processResult(rc, null, null);
                      }
                  }
              }
          };
  
          long startReadEntryTime = MathUtils.nowInNano();
          /*
           * Read the ledger entry using the LedgerHandle. This will allow us to
           * read the entry from one of the other replicated bookies other than
           * the dead one.
           */
          lh.asyncReadEntries(entryId, entryId, new ReadCallback() {
              @Override
              public void readComplete(int rc, LedgerHandle lh,
                      Enumeration<LedgerEntry> seq, Object ctx) {
                  if (rc != BKException.Code.OK) {
                      LOG.error("BK error reading ledger entry: " + entryId,
                              BKException.create(rc));
                      onReadEntryFailureCallback.accept(ledgerId, entryId);
                      ledgerFragmentEntryMcb.processResult(rc, null, null);
                      return;
                  }
  
                  readDataLatency.registerSuccessfulEvent(MathUtils.elapsedNanos(startReadEntryTime),
                          TimeUnit.NANOSECONDS);
  
                  /*
                   * Now that we've read the ledger entry, write it to the new
                   * bookie we've selected.
                   */
                  LedgerEntry entry = seq.nextElement();
                  byte[] data = entry.getEntry();
                  final long dataLength = data.length;
                  numEntriesRead.inc();
                  numBytesRead.registerSuccessfulValue(dataLength);
  
                  ReferenceCounted toSend = lh.getDigestManager()
                          .computeDigestAndPackageForSending(entryId,
                                  lh.getLastAddConfirmed(), entry.getLength(),
                                  Unpooled.wrappedBuffer(data, 0, data.length),
                                  lh.getLedgerKey(),
                                  BookieProtocol.FLAG_RECOVERY_ADD
                                  );
                  if (replicationThrottle != null) {
                      if (toSend instanceof ByteBuf) {
                          updateAverageEntrySize(((ByteBuf) toSend).readableBytes());
                      } else if (toSend instanceof ByteBufList) {
                          updateAverageEntrySize(((ByteBufList) toSend).readableBytes());
                      }
                  }
                  for (BookieId newBookie : newBookies) {
                      long startWriteEntryTime = MathUtils.nowInNano();
                      bkc.getBookieClient().addEntry(newBookie, lh.getId(),
                              lh.getLedgerKey(), entryId, toSend,
                              multiWriteCallback, dataLength, BookieProtocol.FLAG_RECOVERY_ADD,
                              false, WriteFlag.NONE);
                      writeDataLatency.registerSuccessfulEvent(
                             MathUtils.elapsedNanos(startWriteEntryTime), TimeUnit.NANOSECONDS);
                  }
                  toSend.release();
              }
          }, null);
      }
  
      void batchRecoverLedgerFragmentEntry(final long startEntryId,
                                           final long endEntryId,
                                           final LedgerHandle lh,
                                           final AsyncCallback.VoidCallback ledgerFragmentMcb,
                                           final Set<BookieId> newBookies,
                                           final BiConsumer<Long, Long> onReadEntryFailureCallback)
              throws InterruptedException {
          int entriesToReplicateCnt = (int) (endEntryId - startEntryId + 1);
          int maxBytesToReplicate = conf.getReplicationRateByBytes();
          if (replicationThrottle != null) {
              if (maxBytesToReplicate != -1 && maxBytesToReplicate > averageEntrySize.get() * entriesToReplicateCnt) {
                  maxBytesToReplicate = averageEntrySize.get() * entriesToReplicateCnt;
              }
              replicationThrottle.acquire(maxBytesToReplicate);
          }
  
          lh.asyncBatchReadEntries(startEntryId, entriesToReplicateCnt, maxBytesToReplicate,
              new ReadCallback() {
                  @Override
                  public void readComplete(int rc, LedgerHandle lh, Enumeration<LedgerEntry> seq, Object ctx) {
                      if (rc != BKException.Code.OK) {
                          LOG.error("BK error reading ledger entries: {} - {}",
                                  startEntryId, endEntryId, BKException.create(rc));
                          onReadEntryFailureCallback.accept(lh.getId(), startEntryId);
                          for (int i = 0; i < entriesToReplicateCnt; i++) {
                              ledgerFragmentMcb.processResult(rc, null, null);
                          }
                          return;
                      }
                      long lastEntryId = startEntryId;
                      while (seq.hasMoreElements()) {
                          LedgerEntry entry = seq.nextElement();
                          lastEntryId = entry.getEntryId();
                          byte[] data = entry.getEntry();
                          final long dataLength = data.length;
                          numEntriesRead.inc();
                          numBytesRead.registerSuccessfulValue(dataLength);
  
                          ReferenceCounted toSend = lh.getDigestManager()
                                  .computeDigestAndPackageForSending(entry.getEntryId(),
                                          lh.getLastAddConfirmed(), entry.getLength(),
                                          Unpooled.wrappedBuffer(data, 0, data.length),
                                          lh.getLedgerKey(),
                                          BookieProtocol.FLAG_RECOVERY_ADD);
                          if (replicationThrottle != null) {
                              if (toSend instanceof ByteBuf) {
                                  updateAverageEntrySize(((ByteBuf) toSend).readableBytes());
                              } else if (toSend instanceof ByteBufList) {
                                  updateAverageEntrySize(((ByteBufList) toSend).readableBytes());
                              }
                          }
                          AtomicInteger numCompleted = new AtomicInteger(0);
                          AtomicBoolean completed = new AtomicBoolean(false);
  
                          WriteCallback multiWriteCallback = new WriteCallback() {
                              @Override
                              public void writeComplete(int rc, long ledgerId, long entryId, BookieId addr, Object ctx) {
                                  if (rc != BKException.Code.OK) {
                                      LOG.error("BK error writing entry for ledgerId: {}, entryId: {}, bookie: {}",
                                              ledgerId, entryId, addr, BKException.create(rc));
                                      if (completed.compareAndSet(false, true)) {
                                          ledgerFragmentMcb.processResult(rc, null, null);
                                      }
                                  } else {
                                      numEntriesWritten.inc();
                                      if (ctx instanceof Long) {
                                          numBytesWritten.registerSuccessfulValue((Long) ctx);
                                      }
                                      if (LOG.isDebugEnabled()) {
                                          LOG.debug("Success writing ledger id {}, entry id {} to a new bookie {}!",
                                                  ledgerId, entryId, addr);
                                      }
                                      if (numCompleted.incrementAndGet() == newBookies.size()
                                              && completed.compareAndSet(false, true)) {
                                          ledgerFragmentMcb.processResult(rc, null, null);
                                      }
                                  }
                              }
                          };
  
                          for (BookieId newBookie : newBookies) {
                              long startWriteEntryTime = MathUtils.nowInNano();
                              bkc.getBookieClient().addEntry(newBookie, lh.getId(),
                                      lh.getLedgerKey(), entry.getEntryId(), toSend,
                                      multiWriteCallback, dataLength, BookieProtocol.FLAG_RECOVERY_ADD,
                                      false, WriteFlag.NONE);
                              writeDataLatency.registerSuccessfulEvent(
                                      MathUtils.elapsedNanos(startWriteEntryTime), TimeUnit.NANOSECONDS);
                          }
                          toSend.release();
                      }
                      if (lastEntryId != endEntryId) {
                          try {
                              batchRecoverLedgerFragmentEntry(lastEntryId + 1, endEntryId, lh,
                                      ledgerFragmentMcb, newBookies, onReadEntryFailureCallback);
                          } catch (InterruptedException e) {
                              int remainingEntries = (int) (endEntryId - lastEntryId);
                              for (int i = 0; i < remainingEntries; i++) {
                                  ledgerFragmentMcb.processResult(BKException.Code.InterruptedException, null, null);
                              }
                          }
                      }
                  }
              }, null);
      }
  
      private void updateAverageEntrySize(int toSendSize) {
          averageEntrySize.updateAndGet(value -> (int) (value * AVERAGE_ENTRY_SIZE_RATIO
                  + (1 - AVERAGE_ENTRY_SIZE_RATIO) * toSendSize));
      }
  
      /**
       * Callback for recovery of a single ledger fragment. Once the fragment has
       * had all entries replicated, update the ensemble in zookeeper. Once
       * finished propagate callback up to ledgerFragmentsMcb which should be a
       * multicallback responsible for all fragments in a single ledger
       */
      static class SingleFragmentCallback implements AsyncCallback.VoidCallback {
          final AsyncCallback.VoidCallback ledgerFragmentsMcb;
          final LedgerHandle lh;
          final LedgerManager ledgerManager;
          final long fragmentStartId;
          final Map<BookieId, BookieId> oldBookie2NewBookie;
  
          SingleFragmentCallback(AsyncCallback.VoidCallback ledgerFragmentsMcb,
                                 LedgerHandle lh, LedgerManager ledgerManager, long fragmentStartId,
                                 Map<BookieId, BookieId> oldBookie2NewBookie) {
              this.ledgerFragmentsMcb = ledgerFragmentsMcb;
              this.lh = lh;
              this.ledgerManager = ledgerManager;
              this.fragmentStartId = fragmentStartId;
              this.oldBookie2NewBookie = oldBookie2NewBookie;
          }
  
          @Override
          public void processResult(int rc, String path, Object ctx) {
              if (rc != BKException.Code.OK) {
                  LOG.error("BK error replicating ledger fragments for ledger: "
                          + lh.getId(), BKException.create(rc));
                  ledgerFragmentsMcb.processResult(rc, null, null);
                  return;
              }
              updateEnsembleInfo(ledgerManager, ledgerFragmentsMcb, fragmentStartId, lh, oldBookie2NewBookie);
          }
      }
  
      /**
       * Updates the ensemble with newBookie and notify the ensembleUpdatedCb.
       */
      private static void updateEnsembleInfo(
              LedgerManager ledgerManager, AsyncCallback.VoidCallback ensembleUpdatedCb, long fragmentStartId,
              LedgerHandle lh, Map<BookieId, BookieId> oldBookie2NewBookie) {
  
          MetadataUpdateLoop updateLoop = new MetadataUpdateLoop(
                  ledgerManager,
                  lh.getId(),
                  lh::getVersionedLedgerMetadata,
                  (metadata) -> {
                      // returns true if any of old bookies exist in ensemble
                      List<BookieId> ensemble = metadata.getAllEnsembles().get(fragmentStartId);
                      return oldBookie2NewBookie.keySet().stream().anyMatch(ensemble::contains);
                  },
                  (currentMetadata) -> {
                      // replace all old bookies with new bookies in ensemble
                      List<BookieId> newEnsemble = currentMetadata.getAllEnsembles().get(fragmentStartId)
                          .stream().map((bookie) -> oldBookie2NewBookie.getOrDefault(bookie, bookie))
                          .collect(Collectors.toList());
                      return LedgerMetadataBuilder.from(currentMetadata)
                          .replaceEnsembleEntry(fragmentStartId, newEnsemble).build();
                  },
                  lh::setLedgerMetadata);
  
          updateLoop.run().whenComplete((result, ex) -> {
                  if (ex == null) {
                      LOG.info("Updated ZK to point ledger fragments"
                               + " from old bookies to new bookies: {}", oldBookie2NewBookie);
  
                      ensembleUpdatedCb.processResult(BKException.Code.OK, null, null);
                  } else {
                      LOG.error("Error updating ledger config metadata for ledgerId {}", lh.getId(), ex);
  
                      ensembleUpdatedCb.processResult(
                              BKException.getExceptionCode(ex, BKException.Code.UnexpectedConditionException),
                              null, null);
                  }
              });
      }
  
      static class Throttler {
          private final RateLimiter rateLimiter;
  
          Throttler(int throttleBytes) {
              this.rateLimiter = RateLimiter.create(throttleBytes);
          }
  
          // reset rate of limiter before compact one entry log file
          void resetRate(int throttleBytes) {
              this.rateLimiter.setRate(throttleBytes);
          }
  
          // get rate of limiter for unit test
          double getRate() {
              return this.rateLimiter.getRate();
          }
  
          // acquire. if bybytes: bytes of this entry; if byentries: 1.
          void acquire(int permits) {
              rateLimiter.acquire(permits);
          }
      }
  }
  