@Test
  public void storeMessageReadSetTest() throws IOException, StoreException {
    int logCapacity = 2000;
    int segCapacity = 1000;
    Log log = new Log(tempDir.getAbsolutePath(), logCapacity, StoreTestUtils.DEFAULT_DISK_SPACE_ALLOCATOR,
        createStoreConfig(segCapacity, setFilePermissionEnabled), metrics, null);
    try {
      LogSegment firstSegment = log.getFirstSegment();
      int availableSegCapacity = (int) (segCapacity - firstSegment.getStartOffset());
      byte[] srcOfTruth = TestUtils.getRandomBytes(2 * availableSegCapacity);
      ReadableByteChannel dataChannel = Channels.newChannel(new ByteBufferInputStream(ByteBuffer.wrap(srcOfTruth)));
      log.appendFrom(dataChannel, availableSegCapacity);
      log.appendFrom(dataChannel, availableSegCapacity);
      LogSegment secondSegment = log.getNextSegment(firstSegment);

      Offset firstSegOffset1 = new Offset(firstSegment.getName(), firstSegment.getStartOffset());
      Offset firstSegOffset2 =
          new Offset(firstSegment.getName(), firstSegment.getStartOffset() + availableSegCapacity / 2);
      Offset secondSegOffset1 = new Offset(secondSegment.getName(), secondSegment.getStartOffset());
      Offset secondSegOffset2 =
          new Offset(secondSegment.getName(), secondSegment.getStartOffset() + availableSegCapacity / 2);
      List<MockId> mockIdList = new ArrayList<>();
      MockId mockId = new MockId("id1");
      mockIdList.add(mockId);
      BlobReadOptions ro1 = new BlobReadOptions(log, firstSegOffset2,
          new MessageInfo(mockId, availableSegCapacity / 3, 1, Utils.getRandomShort(TestUtils.RANDOM),
              Utils.getRandomShort(TestUtils.RANDOM), System.currentTimeMillis() + TestUtils.RANDOM.nextInt(10000)));
      mockId = new MockId("id2");
      mockIdList.add(mockId);
      BlobReadOptions ro2 = new BlobReadOptions(log, secondSegOffset1,
          new MessageInfo(mockId, availableSegCapacity / 4, 1, Utils.getRandomShort(TestUtils.RANDOM),
              Utils.getRandomShort(TestUtils.RANDOM), System.currentTimeMillis() + TestUtils.RANDOM.nextInt(10000)));
      mockId = new MockId("id3");
      mockIdList.add(mockId);
      BlobReadOptions ro3 = new BlobReadOptions(log, secondSegOffset2,
          new MessageInfo(mockId, availableSegCapacity / 2, 1, Utils.getRandomShort(TestUtils.RANDOM),
              Utils.getRandomShort(TestUtils.RANDOM), System.currentTimeMillis() + TestUtils.RANDOM.nextInt(10000)));
      mockId = new MockId("id4");
      mockIdList.add(mockId);
      BlobReadOptions ro4 = new BlobReadOptions(log, firstSegOffset1,
          new MessageInfo(mockId, availableSegCapacity / 5, 1, Utils.getRandomShort(TestUtils.RANDOM),
              Utils.getRandomShort(TestUtils.RANDOM), System.currentTimeMillis() + TestUtils.RANDOM.nextInt(10000)));
      // to test equality in the compareTo() of BlobReadOptions
      mockId = new MockId("id5");
      mockIdList.add(mockId);
      BlobReadOptions ro5 = new BlobReadOptions(log, firstSegOffset2,
          new MessageInfo(mockId, availableSegCapacity / 6, 1, Utils.getRandomShort(TestUtils.RANDOM),
              Utils.getRandomShort(TestUtils.RANDOM), System.currentTimeMillis() + TestUtils.RANDOM.nextInt(10000)));
      List<BlobReadOptions> options = new ArrayList<>(Arrays.asList(ro1, ro2, ro3, ro4, ro5));
      MessageReadSet readSet = new StoreMessageReadSet(options);

      assertEquals(readSet.count(), options.size());
      // options should get sorted by offsets in the constructor
      assertEquals(readSet.getKeyAt(0), mockIdList.get(3));
      assertEquals(readSet.sizeInBytes(0), availableSegCapacity / 5);
      assertEquals(readSet.getKeyAt(1), mockIdList.get(0));
      assertEquals(readSet.sizeInBytes(1), availableSegCapacity / 3);
      assertEquals(readSet.getKeyAt(2), mockIdList.get(4));
      assertEquals(readSet.sizeInBytes(2), availableSegCapacity / 6);
      assertEquals(readSet.getKeyAt(3), mockIdList.get(1));
      assertEquals(readSet.sizeInBytes(3), availableSegCapacity / 4);
      assertEquals(readSet.getKeyAt(4), mockIdList.get(2));
      assertEquals(readSet.sizeInBytes(4), availableSegCapacity / 2);

      ByteBuffer readBuf = ByteBuffer.allocate(availableSegCapacity / 5);
      ByteBufferOutputStream stream = new ByteBufferOutputStream(readBuf);

      // read the first one all at once
      if (doDataPrefetch) {
        readSet.doPrefetch(0, 0, Long.MAX_VALUE);
      }
      long written = readSet.writeTo(0, Channels.newChannel(stream), 0, Long.MAX_VALUE);
      assertEquals("Return value from writeTo() is incorrect", availableSegCapacity / 5, written);
      assertArrayEquals(readBuf.array(), Arrays.copyOfRange(srcOfTruth, 0, availableSegCapacity / 5));

      // read the second one byte by byte
      readBuf = ByteBuffer.allocate(availableSegCapacity / 3);
      stream = new ByteBufferOutputStream(readBuf);
      WritableByteChannel channel = Channels.newChannel(stream);
      long currentReadOffset = 0;
      if (doDataPrefetch) {
        readSet.doPrefetch(1, currentReadOffset, availableSegCapacity / 3);
      }
      while (currentReadOffset < availableSegCapacity / 3) {
        written = readSet.writeTo(1, channel, currentReadOffset, 1);
        assertEquals("Return value from writeTo() is incorrect", 1, written);
        currentReadOffset++;
      }
      long startOffset = availableSegCapacity / 2;
      long endOffset = availableSegCapacity / 2 + availableSegCapacity / 3;
      assertArrayEquals(readBuf.array(), Arrays.copyOfRange(srcOfTruth, (int) startOffset, (int) endOffset));

      // read the last one in multiple stages
      readBuf = ByteBuffer.allocate(availableSegCapacity / 2);
      stream = new ByteBufferOutputStream(readBuf);
      channel = Channels.newChannel(stream);
      currentReadOffset = 0;
      if (doDataPrefetch) {
        readSet.doPrefetch(4, currentReadOffset, availableSegCapacity / 2);
      }
      while (currentReadOffset < availableSegCapacity / 2) {
        written = readSet.writeTo(4, channel, currentReadOffset, availableSegCapacity / 6);
        long expectedWritten = Math.min(availableSegCapacity / 2 - currentReadOffset, availableSegCapacity / 6);
        assertEquals("Return value from writeTo() is incorrect", expectedWritten, written);
        currentReadOffset += availableSegCapacity / 6;
      }
      startOffset = availableSegCapacity + availableSegCapacity / 2;
      endOffset = startOffset + availableSegCapacity / 2;
      assertArrayEquals(readBuf.array(), Arrays.copyOfRange(srcOfTruth, (int) startOffset, (int) endOffset));

      // should not write anything if relative offset is at the size
      readBuf = ByteBuffer.allocate(1);
      stream = new ByteBufferOutputStream(readBuf);
      channel = Channels.newChannel(stream);
      written = readSet.writeTo(0, channel, readSet.sizeInBytes(0), 1);
      assertEquals("No data should have been written", 0, written);

      try {
        readSet.sizeInBytes(options.size());
        fail("Reading should have failed because index is out of bounds");
      } catch (IndexOutOfBoundsException e) {
        // expected. Nothing to do.
      }

      try {
        readSet.writeTo(options.size(), channel, 100, 100);
        fail("Reading should have failed because index is out of bounds");
      } catch (IndexOutOfBoundsException e) {
        // expected. Nothing to do.
      }

      try {
        readSet.getKeyAt(options.size());
        fail("Getting key should have failed because index is out of bounds");
      } catch (IndexOutOfBoundsException e) {
        // expected. Nothing to do.
      }
    } finally {
      log.close(false);
    }
  }
Primarily tests {@link StoreMessageReadSet} and its APIs but also checks the the {@link Comparable} APIs of
{@link BlobReadOptions}.
@throws IOException
ambry-store/src/test/java/com/github/ambry/store/StoreMessageReadSetTest.java