@Test
  public void partialWriteTest() throws IOException, StoreException {
    LogSegmentName prevLogSegmentName = LogSegmentName.fromPositionAndGeneration(0, 0);
    LogSegmentName logSegmentName = prevLogSegmentName.getNextPositionName();
    Offset startOffset = new Offset(logSegmentName, 0);
    MockId id1 = new MockId("0" + TestUtils.getRandomString(CUSTOM_ID_SIZE - 1));
    MockId id2 = new MockId("1" + TestUtils.getRandomString(CUSTOM_ID_SIZE - 1));
    MockId id3 = new MockId("2" + TestUtils.getRandomString(CUSTOM_ID_SIZE - 1));
    short accountId = Utils.getRandomShort(TestUtils.RANDOM);
    short containerId = Utils.getRandomShort(TestUtils.RANDOM);
    short lifeVersion = 0;
    IndexValue value1 =
        IndexValueTest.getIndexValue(1000, new Offset(logSegmentName, 0), Utils.Infinite_Time, time.milliseconds(),
            accountId, containerId, lifeVersion, formatVersion);
    IndexValue value2 = IndexValueTest.getIndexValue(1000, new Offset(logSegmentName, 1000), time.milliseconds() + 1,
        time.milliseconds(), accountId, containerId, lifeVersion, formatVersion);
    IndexValue value3 =
        IndexValueTest.getIndexValue(1000, new Offset(logSegmentName, 2000), Utils.Infinite_Time, time.milliseconds(),
            accountId, containerId, lifeVersion, formatVersion);
    time.sleep(TimeUnit.SECONDS.toMillis(1));
    // generate a TTL Update
    IndexValue ttlUpValue2 =
        IndexValueTest.getIndexValue(value2.getSize(), value2.getOffset(), Utils.Infinite_Time, time.milliseconds(),
            value2.getAccountId(), value2.getContainerId(), lifeVersion, formatVersion);
    ttlUpValue2.setNewOffset(new Offset(logSegmentName, 3000));
    ttlUpValue2.setNewSize(50);
    ttlUpValue2.setFlag(IndexValue.Flags.Ttl_Update_Index);
    time.sleep(TimeUnit.SECONDS.toMillis(1));
    // generate a DELETE
    IndexValue delValue2 = IndexValueTest.getIndexValue(value2.getSize(), value2.getOffset(), ttlUpValue2.getFlags(),
        value2.getExpiresAtMs(), value2.getOffset().getOffset(), time.milliseconds(), value2.getAccountId(),
        value2.getContainerId(), lifeVersion, formatVersion);
    delValue2.setNewOffset(new Offset(logSegmentName, 3050));
    delValue2.setNewSize(100);
    delValue2.setFlag(IndexValue.Flags.Delete_Index);
    IndexSegment indexSegment = generateIndexSegment(startOffset, STORE_KEY_FACTORY);
    // inserting in the opposite order by design to ensure that writes are based on offset ordering and not key ordering
    indexSegment.addEntry(new IndexEntry(id3, value1), new Offset(logSegmentName, 1000));
    indexSegment.addEntry(new IndexEntry(id2, value2), new Offset(logSegmentName, 2000));
    indexSegment.addEntry(new IndexEntry(id1, value3), new Offset(logSegmentName, 3000));
    indexSegment.addEntry(new IndexEntry(id2, ttlUpValue2), new Offset(logSegmentName, 3050));
    indexSegment.addEntry(new IndexEntry(id2, delValue2), new Offset(logSegmentName, 3150));

    // provide end offsets such that nothing is written
    checkNonCreationOfIndexSegmentFile(indexSegment, new Offset(prevLogSegmentName, 0));
    checkNonCreationOfIndexSegmentFile(indexSegment,
        new Offset(prevLogSegmentName, indexSegment.getStartOffset().getOffset()));
    checkNonCreationOfIndexSegmentFile(indexSegment, new Offset(logSegmentName, 0));
    List<MockId> shouldBeFound = new ArrayList<>();
    List<MockId> shouldNotBeFound = new ArrayList<>(Arrays.asList(id3, id2, id1));
    for (int safeEndPoint = 1000; safeEndPoint <= 3000; safeEndPoint += 1000) {
      shouldBeFound.add(shouldNotBeFound.remove(0));
      // repeat twice
      for (int i = 0; i < 2; i++) {
        indexSegment.writeIndexSegmentToFile(new Offset(logSegmentName, safeEndPoint));
        Journal journal = new Journal(tempDir.getAbsolutePath(), 3, 3);
        IndexSegment fromDisk =
            new IndexSegment(indexSegment.getFile(), false, STORE_KEY_FACTORY, config, metrics, journal, time);
        assertEquals("End offset not as expected", new Offset(logSegmentName, safeEndPoint), fromDisk.getEndOffset());
        assertEquals("Number of items incorrect", shouldBeFound.size(), fromDisk.getNumberOfItems());
        for (MockId id : shouldBeFound) {
          verifyValues(fromDisk, id, 1, EnumSet.noneOf(IndexValue.Flags.class));
        }
        for (MockId id : shouldNotBeFound) {
          assertNull("Values for key should not have been found", fromDisk.find(id));
        }
      }
    }
    // now persist the ttl update only
    indexSegment.writeIndexSegmentToFile(new Offset(logSegmentName, 3050));
    Journal journal = new Journal(tempDir.getAbsolutePath(), 3, 3);
    IndexSegment fromDisk =
        new IndexSegment(indexSegment.getFile(), false, STORE_KEY_FACTORY, config, metrics, journal, time);
    assertEquals("Number of items incorrect", 4, fromDisk.getNumberOfItems());
    for (MockId id : new MockId[]{id1, id2, id3}) {
      int valueCount = id.equals(id2) ? 2 : 1;
      EnumSet<IndexValue.Flags> flags =
          id.equals(id2) ? EnumSet.of(IndexValue.Flags.Ttl_Update_Index) : EnumSet.noneOf(IndexValue.Flags.class);
      verifyValues(fromDisk, id, valueCount, flags);
    }
    // now persist the delete too
    indexSegment.writeIndexSegmentToFile(new Offset(logSegmentName, 3150));
    journal = new Journal(tempDir.getAbsolutePath(), 3, 3);
    fromDisk = new IndexSegment(indexSegment.getFile(), false, STORE_KEY_FACTORY, config, metrics, journal, time);
    assertEquals("Number of items incorrect", 5, fromDisk.getNumberOfItems());
    for (MockId id : new MockId[]{id1, id2, id3}) {
      int valueCount = id.equals(id2) ? 3 : 1;
      EnumSet<IndexValue.Flags> flags =
          id.equals(id2) ? EnumSet.of(IndexValue.Flags.Delete_Index, IndexValue.Flags.Ttl_Update_Index)
              : EnumSet.noneOf(IndexValue.Flags.class);
      verifyValues(fromDisk, id, valueCount, flags);
    }
    // verify reset key
    if (formatVersion >= PersistentIndex.VERSION_1) {
      verifyResetKeyInfo(id3, value1, fromDisk.getResetKey(), fromDisk.getResetKeyType(),
          fromDisk.getResetKeyLifeVersion());
    }
  }
Tests the case when {@link IndexSegment#writeIndexSegmentToFile(Offset)} is provided with different offsets <=
{@link IndexSegment#getEndOffset()} and makes sure that only the relevant parts of the segment are written to disk.
@throws IOException
@throws StoreException
ambry-store/src/test/java/com/github/ambry/store/IndexSegmentTest.java