public Pair<FindToken, Boolean> copy(FindToken startToken) throws Exception {
    boolean sourceHasProblems = false;
    FindToken lastToken;
    FindToken token = startToken;
    do {
      lastToken = token;
      FindInfo findInfo = src.findEntriesSince(lastToken, fetchSizeInBytes, null, null);
      List<MessageInfo> messageInfos = findInfo.getMessageEntries();
      for (Transformer transformer : transformers) {
        transformer.warmup(messageInfos, false);
      }
      for (MessageInfo messageInfo : messageInfos) {
        logger.trace("Processing {} - isDeleted: {}, isExpired {}", messageInfo.getStoreKey(), messageInfo.isDeleted(),
            messageInfo.isExpired());
        if (!messageInfo.isExpired() && !messageInfo.isDeleted()) {
          if (tgt.findMissingKeys(Collections.singletonList(messageInfo.getStoreKey())).size() == 1) {
            StoreInfo storeInfo =
                src.get(Collections.singletonList(messageInfo.getStoreKey()), EnumSet.allOf(StoreGetOptions.class));
            MessageReadSet readSet = storeInfo.getMessageReadSet();
            if (readSet.sizeInBytes(0) > Integer.MAX_VALUE) {
              throw new IllegalStateException("Cannot copy blobs whose size > Integer.MAX_VALUE");
            }
            int size = (int) readSet.sizeInBytes(0);
            byte[] buf = new byte[size];
            readSet.writeTo(0, new ByteBufferChannel(ByteBuffer.wrap(buf)), 0, size);
            Message message = new Message(storeInfo.getMessageReadSetInfo().get(0), new ByteArrayInputStream(buf));
            for (Transformer transformer : transformers) {
              TransformationOutput tfmOutput = transformer.transform(message);
              if (tfmOutput.getException() != null) {
                throw tfmOutput.getException();
              } else {
                message = tfmOutput.getMsg();
              }
              if (message == null) {
                break;
              }
            }
            if (message == null) {
              logger.trace("Dropping {} because the transformers did not return a message", messageInfo.getStoreKey());
              continue;
            }
            MessageFormatWriteSet writeSet =
                new MessageFormatWriteSet(message.getStream(), Collections.singletonList(message.getMessageInfo()),
                    false);
            tgt.put(writeSet);
            MessageInfo tgtMsgInfo = message.getMessageInfo();
            if (tgtMsgInfo.isTtlUpdated()) {
              MessageInfo updateMsgInfo =
                  new MessageInfo(tgtMsgInfo.getStoreKey(), 0, false, true, tgtMsgInfo.getExpirationTimeInMs(),
                      tgtMsgInfo.getAccountId(), tgtMsgInfo.getContainerId(), tgtMsgInfo.getOperationTimeMs());
              tgt.updateTtl(Collections.singletonList(updateMsgInfo));
            }
            logger.trace("Copied {} as {}", messageInfo.getStoreKey(), tgtMsgInfo.getStoreKey());
          } else if (!messageInfo.isTtlUpdated()) {
            logger.warn("Found a duplicate entry for {} while copying data", messageInfo.getStoreKey());
            sourceHasProblems = true;
          }
        }
      }
      token = findInfo.getFindToken();
      double percentBytesRead = src.isEmpty() ? 100.0 : token.getBytesRead() * 100.0 / src.getSizeInBytes();
      logger.info("[{}] [{}] {}% copied", Thread.currentThread().getName(), storeId, df.format(percentBytesRead));
    } while (!token.equals(lastToken));
    return new Pair<>(token, sourceHasProblems);
  }
Copies data starting from {@code startToken} until all the data is copied.
@param startToken the {@link FindToken} to start copying from. It is expected that start token does not cause
                  the copier to attempt to copy blobs that have already been copied. If that happens, the boolean
                  in the return value will be {@code true}.
@return a {@link Pair} of the {@link FindToken} until which data has been copied and a {@link Boolean} indicating
whether the source had problems that were skipped over - like duplicates ({@code true} indicates that there were).
@throws Exception if there is any exception during processing
ambry-tools/src/main/java/com/github/ambry/store/StoreCopier.java