private boolean copyRecords(LogSegment logSegmentToCopy, List<IndexEntry> srcIndexEntries, long lastModifiedTimeSecs)
      throws IOException, StoreException {
    boolean copiedAll = true;
    long totalCapacity = tgtLog.getCapacityInBytes();
    long writtenLastTime = 0;
    try (FileChannel fileChannel = Utils.openChannel(logSegmentToCopy.getView().getFirst(), false)) {
      // byte[] for both general IO and direct IO.
      byte[] byteArrayToUse;
      // ByteBuffer for general IO
      ByteBuffer bufferToUse = null;
      // the index number of the list to start/end with.
      int start = 0, end;
      // the size (in bytes) from start to end for read.
      int readSize;
      while (start < srcIndexEntries.size()) {
        // try to do a bundle of read to reduce disk IO
        long startOffset = srcIndexEntries.get(start).getValue().getOffset().getOffset();
        if (bundleReadBuffer == null || srcIndexEntries.get(start).getValue().getSize() > bundleReadBuffer.length) {
          end = start;
          readSize = (int) srcIndexEntries.get(start).getValue().getSize();
          byteArrayToUse = new byte[readSize];
          srcMetrics.compactionBundleReadBufferNotFitIn.inc();
          logger.trace("Record size greater than bundleReadBuffer capacity, key: {} size: {}",
              srcIndexEntries.get(start).getKey(), srcIndexEntries.get(start).getValue().getSize());
        } else {
          end = getBundleReadEndIndex(srcIndexEntries, start);
          readSize = (int) (srcIndexEntries.get(end).getValue().getOffset().getOffset() + srcIndexEntries.get(end)
              .getValue()
              .getSize() - startOffset);
          byteArrayToUse = bundleReadBuffer;
          srcMetrics.compactionBundleReadBufferUsed.inc();
        }
        if (useDirectIO) {
          // do direct IO read
          logSegmentToCopy.readIntoDirectly(byteArrayToUse, startOffset, readSize);
        } else {
          // do general IO read
          bufferToUse = ByteBuffer.wrap(byteArrayToUse);
          bufferToUse.position(0);
          bufferToUse.limit(readSize);
          int ioCount = Utils.readFileToByteBuffer(fileChannel, startOffset, bufferToUse);
          srcMetrics.compactionBundleReadBufferIoCount.inc(ioCount);
        }
        srcMetrics.compactionBufferReadSize.update(readSize);

        // copy from buffer to tgtLog
        int effectiveBytes = 0;
        for (int i = start; i <= end; i++) {
          IndexEntry srcIndexEntry = srcIndexEntries.get(i);
          IndexValue srcValue = srcIndexEntry.getValue();
          long usedCapacity = tgtIndex.getLogUsedCapacity();
          effectiveBytes += (int) srcValue.getSize();
          if (isActive && (tgtLog.getCapacityInBytes() - usedCapacity >= srcValue.getSize())) {
            Offset endOffsetOfLastMessage = tgtLog.getEndOffset();
            if (config.storeCompactionMinOperationsBytesPerSec < config.storeCompactionOperationsBytesPerSec
                && diskMetrics != null) {
              // Enable dynamic desired rate based on disk latency.
              double currentReadTimePerMbInMs = diskMetrics.diskReadTimePerMbInMs.getSnapshot().get95thPercentile();
              int desiredReadPerSecond = getDesiredSpeedPerSecond(currentReadTimePerMbInMs,
                  config.storeCompactionIoPerMbReadLatencyThresholdMs);
              double currentWriteTimePerMbInMs = diskMetrics.diskWriteTimePerMbInMs.getSnapshot().get95thPercentile();
              int desiredWritePerSecond = getDesiredSpeedPerSecond(currentWriteTimePerMbInMs,
                  config.storeCompactionIoPerMbWriteLatencyThresholdMs);
              if (currentReadTimePerMbInMs > config.storeCompactionIoPerMbReadLatencyThresholdMs
                  || currentWriteTimePerMbInMs > config.storeCompactionIoPerMbWriteLatencyThresholdMs) {
                // Only log when compaction copy rate is impacted.
                logger.debug(
                    "Current disk read per MB: {}ms, current disk write per MB: {} ms, Desired compaction copy rate(bytes/seconds): read: {} write: {}",
                    currentReadTimePerMbInMs, currentWriteTimePerMbInMs, desiredReadPerSecond, desiredWritePerSecond);
              }
              int desiredRate = Math.min(desiredReadPerSecond, desiredWritePerSecond);
              double currentRate = diskMetrics.diskCompactionCopyRateInBytes.getOneMinuteRate();
              if (currentRate > desiredRate) {
                // If last one minute rate is too high, need to slow down(by sleep) to desired rate.
                // 100 -> 50 : sleep 2-1 time unit
                // 100 -> 25 : sleep 4-1 time unit
                // 100 -> 90: sleep (100/90 - 1) time unit
                long sleepInMs = (long) ((currentRate / desiredRate - 1) * Time.MsPerSec);
                long actualSleepTimeInMs = Math.min(sleepInMs, 10 * Time.MsPerSec);
                logger.debug(
                    "Current rate: {}, desired rate: {}. Calculated sleep time: {} ms. Actual sleep time: {} ms",
                    currentRate, desiredRate, sleepInMs, actualSleepTimeInMs);
                try {
                  Thread.sleep(actualSleepTimeInMs);
                } catch (InterruptedException e) {
                  logger.warn("Compaction throttling sleep is interrupted", e);
                }
              }
            } else {
              logger.debug(
                  "Adaptive compaction is not enabled: storeCompactionMinOperationsBytesPerSec: {}, storeCompactionOperationsBytesPerSec: {}",
                  config.storeCompactionMinOperationsBytesPerSec, config.storeCompactionOperationsBytesPerSec);
              // call into diskIOScheduler to make sure we can proceed.
              diskIOScheduler.getSlice(COMPACTION_CLEANUP_JOB_NAME, COMPACTION_CLEANUP_JOB_NAME, writtenLastTime);
            }

            if (useDirectIO) {
              // do direct IO write
              tgtLog.appendFromDirectly(byteArrayToUse, (int) (srcValue.getOffset().getOffset() - startOffset),
                  (int) srcIndexEntry.getValue().getSize());
            } else {
              // do general write
              long bufferPosition = srcValue.getOffset().getOffset() - startOffset;
              bufferToUse.limit((int) (bufferPosition + srcIndexEntry.getValue().getSize()));
              bufferToUse.position((int) (bufferPosition));
              tgtLog.appendFrom(bufferToUse);
            }
            FileSpan fileSpan = tgtLog.getFileSpanForMessage(endOffsetOfLastMessage, srcValue.getSize());
            IndexValue valueFromTgtIdx = tgtIndex.findKey(srcIndexEntry.getKey());
            if (srcValue.isDelete()) {
              if (valueFromTgtIdx != null) {
                tgtIndex.markAsDeleted(srcIndexEntry.getKey(), fileSpan, null, srcValue.getOperationTimeInMs(),
                    srcValue.getLifeVersion());
              } else {
                IndexValue tgtValue = new IndexValue(srcValue.getSize(), fileSpan.getStartOffset(), srcValue.getFlags(),
                    srcValue.getExpiresAtMs(), srcValue.getOperationTimeInMs(), srcValue.getAccountId(),
                    srcValue.getContainerId(), srcValue.getLifeVersion());
                tgtValue.setFlag(IndexValue.Flags.Delete_Index);
                tgtValue.clearOriginalMessageOffset();
                tgtIndex.addToIndex(new IndexEntry(srcIndexEntry.getKey(), tgtValue), fileSpan);
              }
            } else if (srcValue.isUndelete()) {
              if (valueFromTgtIdx != null) {
                // Here don't use markAsUndelete method. In markAsUndelete, multiple sanity check would be applied. But
                // target index might only contain incomplete blob history, which would fail sanity check.
                // Here we do a different check to make sure we don't insert undelete with lower lifeVersion.
                if (valueFromTgtIdx.isUndelete()) {
                  throw new StoreException(
                      "Id " + srcIndexEntry.getKey() + " already has undelete " + valueFromTgtIdx + " in index "
                          + dataDir, StoreErrorCodes.IDUndeleted);
                }
                // Undelete would increase the life version
                if (valueFromTgtIdx.getLifeVersion() >= srcValue.getLifeVersion()) {
                  throw new StoreException(
                      "Id " + srcIndexEntry.getKey() + " has bad lifeversion " + valueFromTgtIdx + " in index "
                          + dataDir, StoreErrorCodes.LifeVersionConflict);
                }
              }
              IndexValue tgtValue = new IndexValue(srcValue.getSize(), fileSpan.getStartOffset(), srcValue.getFlags(),
                  srcValue.getExpiresAtMs(), srcValue.getOperationTimeInMs(), srcValue.getAccountId(),
                  srcValue.getContainerId(), srcValue.getLifeVersion());
              tgtValue.setFlag(IndexValue.Flags.Undelete_Index);
              tgtValue.clearOriginalMessageOffset();
              tgtIndex.addToIndex(new IndexEntry(srcIndexEntry.getKey(), tgtValue), fileSpan);
            } else if (srcValue.isTtlUpdate()) {
              if (valueFromTgtIdx != null) {
                tgtIndex.markAsPermanent(srcIndexEntry.getKey(), fileSpan, null, srcValue.getOperationTimeInMs(),
                    srcValue.getLifeVersion());
              } else {
                IndexValue tgtValue = new IndexValue(srcValue.getSize(), fileSpan.getStartOffset(), srcValue.getFlags(),
                    srcValue.getExpiresAtMs(), srcValue.getOperationTimeInMs(), srcValue.getAccountId(),
                    srcValue.getContainerId(), srcValue.getLifeVersion());
                tgtValue.setFlag(IndexValue.Flags.Ttl_Update_Index);
                tgtValue.clearOriginalMessageOffset();
                tgtIndex.addToIndex(new IndexEntry(srcIndexEntry.getKey(), tgtValue), fileSpan);
              }
            } else if (valueFromTgtIdx != null) {
              throw new StoreException("Cannot insert duplicate PUT entry for " + srcIndexEntry.getKey(),
                  StoreErrorCodes.UnknownError);
            } else {
              // Add a PUT IndexValue with the same info (especially same lifeVersion).
              IndexValue tgtValue =
                  new IndexValue(srcValue.getSize(), fileSpan.getStartOffset(), IndexValue.FLAGS_DEFAULT_VALUE,
                      srcValue.getExpiresAtMs(), fileSpan.getStartOffset().getOffset(), srcValue.getOperationTimeInMs(),
                      srcValue.getAccountId(), srcValue.getContainerId(), srcValue.getLifeVersion());
              tgtIndex.addToIndex(new IndexEntry(srcIndexEntry.getKey(), tgtValue), fileSpan);
            }
            long lastModifiedTimeSecsToSet =
                srcValue.getOperationTimeInMs() != Utils.Infinite_Time ? srcValue.getOperationTimeInMs() / Time.MsPerSec
                    : lastModifiedTimeSecs;
            tgtIndex.getIndexSegments().lastEntry().getValue().setLastModifiedTimeSecs(lastModifiedTimeSecsToSet);
            writtenLastTime = srcValue.getSize();
            srcMetrics.compactionCopyRateInBytes.mark(srcValue.getSize());
            if (diskMetrics != null) {
              diskMetrics.diskCompactionCopyRateInBytes.mark(srcValue.getSize());
            }
          } else if (!isActive) {
            logger.info("Stopping copying in {} because shutdown is in progress", storeId);
            copiedAll = false;
            break;
          } else {
            // this is the extra segment, so it is ok to run out of space.
            logger.info(
                "There is no more capacity in the destination log in {}. Total capacity is {}. Used capacity is {}."
                    + " Segment that was being copied is {}", storeId, totalCapacity, usedCapacity,
                logSegmentToCopy.getName());
            copiedAll = false;
            break;
          }
        }
        if (readSize != 0) {
          srcMetrics.compactionBufferReadUtilizationRate.update((int) (effectiveBytes * 1.0 / readSize * 100));
        }
        if (!copiedAll) {
          // break outer while loop
          break;
        }
        tgtLog.flush();
        start = end + 1;
      }
    } finally {
      logSegmentToCopy.closeView();
    }
    return copiedAll;
  }
Copies the given {@code srcIndexEntries} from the given log segment into the swap spaces.
@param logSegmentToCopy the {@link LogSegment} to copy from.
@param srcIndexEntries the {@link IndexEntry}s to copy.
@param lastModifiedTimeSecs the last modified time of the source index segment.
@return @code true} if all the records  were copied. {@code false} if some records were not copied either because
there was no more capacity or because a shutdown was initiated.
@throws IOException if there were I/O errors during copying.
@throws StoreException if there are any problems reading or writing to store components.
ambry-store/src/main/java/com/github/ambry/store/BlobStoreCompactor.java