@Test
  public void recoveryFailureTest() {
    // recovery info contains a PUT for a key that already exists
    MessageInfo info = new MessageInfo(state.liveKeys.iterator().next(), CuratedLogIndexState.PUT_RECORD_SIZE,
        Utils.getRandomShort(TestUtils.RANDOM), Utils.getRandomShort(TestUtils.RANDOM), Utils.Infinite_Time);
    doRecoveryFailureTest(info, StoreErrorCodes.IndexRecoveryError);
    // recovery info contains a PUT for a key that has been deleted
    info = new MessageInfo(state.deletedKeys.iterator().next(), CuratedLogIndexState.PUT_RECORD_SIZE,
        Utils.getRandomShort(TestUtils.RANDOM), Utils.getRandomShort(TestUtils.RANDOM), Utils.Infinite_Time);
    doRecoveryFailureTest(info, StoreErrorCodes.IndexRecoveryError);
    // recovery info contains a Ttl Update for a key that does not exist and there is no delete info that follows
    MockId nonExistentId = state.getUniqueId();
    info = new MessageInfo(nonExistentId, CuratedLogIndexState.TTL_UPDATE_RECORD_SIZE, false, true,
        nonExistentId.getAccountId(), nonExistentId.getContainerId(), state.time.milliseconds());
    doRecoveryFailureTest(info, StoreErrorCodes.IndexRecoveryError);
    // recovery info contains a Ttl Update for a key that is already Ttl updated
    MockId updatedId = null;
    for (MockId id : state.ttlUpdatedKeys) {
      if (state.liveKeys.contains(id)) {
        updatedId = id;
        break;
      }
    }
    assertNotNull("There is no key that is ttl updated but not deleted", updatedId);
    info =
        new MessageInfo(updatedId, CuratedLogIndexState.TTL_UPDATE_RECORD_SIZE, false, true, updatedId.getAccountId(),
            updatedId.getContainerId(), state.time.milliseconds());
    doRecoveryFailureTest(info, StoreErrorCodes.AlreadyUpdated);
    // recovery info contains a Ttl Update for a key that is already deleted
    MockId deletedId = state.deletedKeys.iterator().next();
    info =
        new MessageInfo(deletedId, CuratedLogIndexState.TTL_UPDATE_RECORD_SIZE, false, true, deletedId.getAccountId(),
            deletedId.getContainerId(), state.time.milliseconds());
    doRecoveryFailureTest(info, StoreErrorCodes.IDDeleted);
    // recovery info contains a DELETE for a key that has been deleted, with the same lifeVersion
    info = new MessageInfo(state.deletedKeys.iterator().next(), CuratedLogIndexState.DELETE_RECORD_SIZE, true, false,
        Utils.getRandomShort(TestUtils.RANDOM), Utils.getRandomShort(TestUtils.RANDOM), state.time.milliseconds());
    doRecoveryFailureTest(info, StoreErrorCodes.LifeVersionConflict);
    // recovery info that contains a PUT beyond the end offset of the log segment
    info = new MessageInfo(state.getUniqueId(), CuratedLogIndexState.PUT_RECORD_SIZE,
        Utils.getRandomShort(TestUtils.RANDOM), Utils.getRandomShort(TestUtils.RANDOM), Utils.Infinite_Time);
    doRecoveryFailureTest(info, StoreErrorCodes.LogEndOffsetError);
  }
Tests recovery failure cases.
Cases
1. Recovery info contains a PUT for a key that already exists
2. Recovery info contains a PUT for a key that has been deleted
3. Recovery info contains a DELETE for a key that has been deleted
4. Recovery info that contains a DELETE for a key that has no PUT record
5. Recovery info that contains a PUT beyond the end offset of the log segment
ambry-store/src/test/java/com/github/ambry/store/IndexTest.java