private void multipleSegmentRecoveryTest() throws StoreException {
    Offset indexEndOffsetBeforeRecovery = state.index.getCurrentEndOffset();
    LogSegment activeSegment = state.log.getSegment(indexEndOffsetBeforeRecovery.getName());
    // recover a few messages across segments
    final List<MessageInfo> activeSegmentInfos = new ArrayList<>();
    // 1 PUT record that will be deleted in the next log segment
    MockId idToCreateAndDeleteAcrossSegments = state.getUniqueId();
    short pdAccountId = idToCreateAndDeleteAcrossSegments.getAccountId();
    short pdContainerId = idToCreateAndDeleteAcrossSegments.getContainerId();
    state.appendToLog(CuratedLogIndexState.PUT_RECORD_SIZE);
    activeSegmentInfos.add(
        new MessageInfo(idToCreateAndDeleteAcrossSegments, CuratedLogIndexState.PUT_RECORD_SIZE, pdAccountId,
            pdContainerId, state.time.milliseconds()));
    // 1 PUT record that will be ttl updated in the next log segment
    MockId idToCreateAndUpdateAcrossSegments = state.getUniqueId();
    short puAccountId = idToCreateAndUpdateAcrossSegments.getAccountId();
    short puContainerId = idToCreateAndUpdateAcrossSegments.getContainerId();
    state.appendToLog(CuratedLogIndexState.PUT_RECORD_SIZE);
    activeSegmentInfos.add(
        new MessageInfo(idToCreateAndUpdateAcrossSegments, CuratedLogIndexState.PUT_RECORD_SIZE, puAccountId,
            puContainerId, state.time.milliseconds()));
    // 1 TTL update record that will be deleted in the next log segment
    MockId idToUpdateAndDeleteAcrossSegments = state.getIdToDeleteFromLogSegment(state.log.getFirstSegment(), false);
    short udAccountId = idToUpdateAndDeleteAcrossSegments.getAccountId();
    short udContainerId = idToUpdateAndDeleteAcrossSegments.getContainerId();
    state.appendToLog(CuratedLogIndexState.TTL_UPDATE_RECORD_SIZE);
    activeSegmentInfos.add(
        new MessageInfo(idToUpdateAndDeleteAcrossSegments, CuratedLogIndexState.TTL_UPDATE_RECORD_SIZE, false, true,
            udAccountId, udContainerId, state.time.milliseconds()));
    state.ttlUpdatedKeys.add(idToUpdateAndDeleteAcrossSegments);
    // 1 PUT record that will remain and covers almost the rest of the active segment.
    long size =
        activeSegment.getCapacityInBytes() - activeSegment.getEndOffset() - (CuratedLogIndexState.DELETE_RECORD_SIZE
            - 1);
    state.appendToLog(size);
    activeSegmentInfos.add(new MessageInfo(state.getUniqueId(), size, Utils.getRandomShort(TestUtils.RANDOM),
        Utils.getRandomShort(TestUtils.RANDOM), Utils.Infinite_Time));
    final List<MessageInfo> nextSegmentInfos = getCuratedSingleSegmentRecoveryInfos();
    // 1 DELETE record for the PUT in the previous segment
    state.appendToLog(CuratedLogIndexState.DELETE_RECORD_SIZE);
    nextSegmentInfos.add(
        new MessageInfo(idToCreateAndDeleteAcrossSegments, CuratedLogIndexState.DELETE_RECORD_SIZE, true, false, false,
            Utils.Infinite_Time, null, pdAccountId, pdContainerId, state.time.milliseconds(), (short) 5));
    // 1 TTL update for the PUT in the previous segment
    state.appendToLog(CuratedLogIndexState.TTL_UPDATE_RECORD_SIZE);
    nextSegmentInfos.add(
        new MessageInfo(idToCreateAndUpdateAcrossSegments, CuratedLogIndexState.TTL_UPDATE_RECORD_SIZE, false, true,
            false, Utils.Infinite_Time, null, puAccountId, puContainerId, state.time.milliseconds(), (short) 6));
    state.ttlUpdatedKeys.add(idToUpdateAndDeleteAcrossSegments);
    // 1 DELETE for the TTL update in the previous segment
    state.appendToLog(CuratedLogIndexState.DELETE_RECORD_SIZE);
    nextSegmentInfos.add(
        new MessageInfo(idToUpdateAndDeleteAcrossSegments, CuratedLogIndexState.DELETE_RECORD_SIZE, true, false, false,
            Utils.Infinite_Time, null, udAccountId, udContainerId, state.time.milliseconds(), (short) 7));
    final AtomicInteger returnTracker = new AtomicInteger(0);
    state.recovery = (read, startOffset, endOffset, factory) -> {
      switch (returnTracker.getAndIncrement()) {
        case 0:
          return new MessageStoreRecovery.RecoveryResult(activeSegmentInfos, null, endOffset);
        case 1:
          return new MessageStoreRecovery.RecoveryResult(nextSegmentInfos, null, endOffset);
        default:
          throw new IllegalStateException("This function should not have been called more than two times");
      }
    };
    long activeSegmentExpectedEndOffset = activeSegment.getEndOffset();
    long nextSegmentExpectedEndOffset = state.log.getNextSegment(activeSegment).getEndOffset();
    // write a little "extra" data
    state.appendToLog(2 * CuratedLogIndexState.PUT_RECORD_SIZE);
    state.reloadIndex(true, false);
    assertEquals("End offset of former active segment not as expected", activeSegmentExpectedEndOffset,
        activeSegment.getEndOffset());
    assertEquals("Journal should only include entry for last log segment",
        state.index.journal.getFirstOffset().getName(), state.index.log.getLastSegment().getName());
    activeSegment = state.log.getNextSegment(activeSegment);
    assertNotNull("A new segment has not been created", activeSegment);
    assertEquals("End offset active segment not as expected", nextSegmentExpectedEndOffset,
        activeSegment.getEndOffset());
    List<MessageInfo> infos = new ArrayList<>(activeSegmentInfos);
    infos.addAll(nextSegmentInfos);
    infos.forEach(this::checkRecoveryInfoEquivalence);
  }
Tests recovery of more than one segment.
@throws StoreException
ambry-store/src/test/java/com/github/ambry/store/IndexTest.java