@Test
  public void partialRecoveryWithLastLogSegment() throws Exception {
    assumeTrue(isLogSegmented);
    // Make sure that we started with a new index segment
    // Write everything to disk
    IndexSegment lastIndexSegment = state.index.getIndexSegments().get(state.index.getActiveIndexSegmentOffset());
    while (true) {
      state.addPutEntries(1, PUT_RECORD_SIZE, -1);
      if (state.index.getIndexSegments().get(state.index.getActiveIndexSegmentOffset()) != lastIndexSegment) {
        // if this is the case, we already have 1 put entry
        // Add another 3 entries
        state.addPutEntries(3, PUT_RECORD_SIZE, -1);
        break;
      }
    }
    Map.Entry<Offset, TreeMap<MockId, TreeSet<IndexValue>>> lastIndexSegmentEntry = state.referenceIndex.lastEntry();
    List<MessageInfo> activeSegmentInfos = getMessagesFromIndexSegmentEntryInOrder(lastIndexSegmentEntry);

    Offset endOffset = state.index.getCurrentEndOffset();
    lastIndexSegment = state.index.getIndexSegments().get(state.index.getActiveIndexSegmentOffset());
    // Write everything to disk
    state.index.close(false);

    // Now remove the last index segment file, this would force a recovery
    File indexFile = lastIndexSegment.getFile();
    indexFile.delete();

    // Return RecoveryResult with an exception
    final List<MessageInfo> recoveryMessages =
        new ArrayList<>(activeSegmentInfos.subList(0, activeSegmentInfos.size() - 1));
    MessageInfo lastMessage = activeSegmentInfos.get(activeSegmentInfos.size() - 1);
    long lastMessageSize = lastMessage.getSize();
    long recoveryOffset = endOffset.getOffset() - lastMessageSize;

    state.recovery = (read, startOffset, eoffset, factory) -> new MessageStoreRecovery.RecoveryResult(recoveryMessages,
        new StoreException(new IndexOutOfBoundsException(), StoreErrorCodes.LogFileFormatError), recoveryOffset);

    // Case 1, we haven't enabled the partial log segment recovery, it should fail
    try {
      state.reloadLog(true);
      fail("Should fail due to exception in recovery");
    } catch (StoreException ex) {
      Assert.assertEquals(StoreErrorCodes.LogFileFormatError, ex.getErrorCode());
      Assert.assertTrue(ex.getCause() instanceof IndexOutOfBoundsException);
      Assert.assertNull(state.index);
    }

    // Case 2, enable partial log segment recovery, but the threshold is not smaller than the last message size
    state.properties.setProperty(StoreConfig.storeEnablePartialLogSegmentRecoveryName, "true");
    state.properties.setProperty(StoreConfig.storePartialLogSegmentRecoveryRemainingDataSizeThresholdName,
        String.valueOf(lastMessageSize / 2));
    try {
      state.reloadLog(true);
      fail("Should fail due to threshold being too low for recovery");
    } catch (StoreException ex) {
      Assert.assertEquals(StoreErrorCodes.LogFileFormatError, ex.getErrorCode());
      Assert.assertTrue(ex.getCause() instanceof IndexOutOfBoundsException);
      Assert.assertNull(state.index);
    }

    // Case 3. enable partial log segment recovery and the threshold is high enough for last message
    state.properties.setProperty(StoreConfig.storePartialLogSegmentRecoveryRemainingDataSizeThresholdName,
        String.valueOf(lastMessageSize * 2));
    state.reloadLog(true);
    Assert.assertTrue(state.index.hasPartialRecovery());

    // Make sure all the recovered messages are valid in the state
    recoveryMessages.forEach(this::checkRecoveryInfoEquivalence);

    IndexSegment recoveredIndexSegment = state.index.getIndexSegments().get(state.index.getActiveIndexSegmentOffset());
    assertEquals("End offset doesn't match", recoveryOffset, recoveredIndexSegment.getEndOffset().getOffset());
    assertEquals("Number of index entries doesn't match", recoveryMessages.size(),
        recoveredIndexSegment.getNumberOfItems());
    for (Map.Entry<MockId, TreeSet<IndexValue>> keyAndValues : lastIndexSegmentEntry.getValue().entrySet()) {
      MockId key = keyAndValues.getKey();
      TreeSet<IndexValue> indexValues = keyAndValues.getValue();
      NavigableSet<IndexValue> indexValuesFromSegment = recoveredIndexSegment.find(key);
      if (key.equals(lastMessage.getStoreKey())) {
        assertNull(indexValuesFromSegment);
      } else {
        assertNotNull(indexValuesFromSegment);
        assertEquals("Number of index values doesn't match for key " + key, indexValues.size(),
            indexValuesFromSegment.size());
        assertEquals("IndexValues don't match", indexValues, indexValuesFromSegment);
      }
    }

    File logSegmentFile = state.log.getLastSegment().getView().getFirst();
    Assert.assertEquals(recoveryOffset, logSegmentFile.length());
  }
Tests success case for recovery when the last index segment file is corrupted.
@throws Exception
ambry-store/src/test/java/com/github/ambry/store/IndexTest.java