@Test
  public void setEndOffsetsTest() throws StoreException {
    // check that current end offsets set are correct
    LogSegment segment = state.log.getFirstSegment();
    while (segment != null) {
      Offset lastRecordStartOffset = getLastRecordOffset(segment);
      long size = state.logOrder.get(lastRecordStartOffset).getSecond().indexValue.getSize();
      Offset expectedEndOffset = state.log.getFileSpanForMessage(lastRecordStartOffset, size).getEndOffset();
      assertEquals("End offset of segment not as expected", expectedEndOffset.getOffset(), segment.getEndOffset());
      segment = state.log.getNextSegment(segment);
    }

    // write some data to the log but not the index, check that end offset of the segment has changed
    // reload the index and check the end offset has been reset
    LogSegment activeSegment = state.log.getSegment(state.index.getCurrentEndOffset().getName());
    long offsetBeforeAppend = activeSegment.getEndOffset();
    state.appendToLog(CuratedLogIndexState.PUT_RECORD_SIZE);
    assertEquals("End offset of active segment did not change",
        offsetBeforeAppend + CuratedLogIndexState.PUT_RECORD_SIZE, activeSegment.getEndOffset());
    state.reloadIndex(true, false);
    assertEquals("End offset of active segment should have been reset", offsetBeforeAppend,
        activeSegment.getEndOffset());

    if (isLogSegmented) {
      // this test works under the assumption that log segments are not allocated until they are required
      // this is a fair assumption because the PersistentIndex works under the same assumption and would break if it
      // were not true (which this test failing would indicate).

      // write some data to the log but not the index such that new segment is created, check that end offset of the
      // segment has changed and a new segment created, reload the index and check the end offset has been reset and
      // the new segment does not exist.
      activeSegment = state.log.getSegment(state.index.getCurrentEndOffset().getName());
      offsetBeforeAppend = activeSegment.getEndOffset();
      // fill up this segment
      state.appendToLog(activeSegment.getCapacityInBytes() - activeSegment.getEndOffset());
      assertEquals("End offset of active segment did not change", activeSegment.getCapacityInBytes(),
          activeSegment.getEndOffset());
      // write a little more so that a new segment is created
      state.appendToLog(CuratedLogIndexState.PUT_RECORD_SIZE);
      LogSegment nextActiveSegment = state.log.getNextSegment(activeSegment);
      assertNotNull("New segment has not been created", nextActiveSegment);
      assertEquals("Unexpected end offset for new segment", CuratedLogIndexState.PUT_RECORD_SIZE,
          nextActiveSegment.getEndOffset() - nextActiveSegment.getStartOffset());
      state.reloadIndex(true, false);
      // there should no longer be a "next" segment to the old active segment
      assertNull("There should have been no more segments", state.log.getNextSegment(activeSegment));
      assertEquals("End offset of active segment should have been reset", offsetBeforeAppend,
          activeSegment.getEndOffset());
    }
  }
Tests that end offsets are set correctly in log segments when the index is created.
Cases
1. Current end offsets have been set correctly
2. Add data to log but not index, restart and check that end offsets have been reset.
3. Add data to log such that a new log segment is created but not to index, restart and check that the new log
segment is gone.
@throws StoreException
ambry-store/src/test/java/com/github/ambry/store/IndexTest.java