@Test
  public void messageFormatDeleteRecordTest() throws IOException, MessageFormatException {
    short[] versions = {MessageFormatRecord.Update_Version_V1, MessageFormatRecord.Update_Version_V2,
        MessageFormatRecord.Update_Version_V3};
    for (short version : versions) {
      StoreKey key = new MockId("id1");
      short accountId = Utils.getRandomShort(TestUtils.RANDOM);
      short containerId = Utils.getRandomShort(TestUtils.RANDOM);
      long deletionTimeMs = SystemTime.getInstance().milliseconds() + TestUtils.RANDOM.nextInt();
      short lifeVersion = 1;
      MessageFormatInputStream messageFormatStream;
      short messageHeaderVersionInUse;
      int deleteRecordSize;
      if (version == MessageFormatRecord.Update_Version_V1) {
        messageFormatStream = new DeleteMessageFormatV1InputStream(key, accountId, containerId, deletionTimeMs);
        deleteRecordSize = MessageFormatRecord.Update_Format_V1.getRecordSize();
        messageHeaderVersionInUse = MessageFormatRecord.Message_Header_Version_V1;
        // reset account, container ids and time
        accountId = Account.UNKNOWN_ACCOUNT_ID;
        containerId = Container.UNKNOWN_CONTAINER_ID;
        deletionTimeMs = Utils.Infinite_Time;
      } else if (version == MessageFormatRecord.Update_Version_V2) {
        messageFormatStream =
            new DeleteMessageFormatV2InputStream(key, accountId, containerId, deletionTimeMs, lifeVersion);
        deleteRecordSize = MessageFormatRecord.Update_Format_V2.getRecordSize();
        messageHeaderVersionInUse = MessageFormatRecord.headerVersionToUse;
      } else {
        messageFormatStream =
            new DeleteMessageFormatInputStream(key, accountId, containerId, deletionTimeMs, lifeVersion);
        deleteRecordSize = MessageFormatRecord.Update_Format_V3.getRecordSize(SubRecord.Type.DELETE);
        messageHeaderVersionInUse = MessageFormatRecord.headerVersionToUse;
      }
      int headerSize = MessageFormatRecord.getHeaderSizeForVersion(messageHeaderVersionInUse);
      Assert.assertEquals("Unexpected size for version " + version, headerSize + deleteRecordSize + key.sizeInBytes(),
          messageFormatStream.getSize());

      // check header
      byte[] headerOutput = new byte[headerSize];
      messageFormatStream.read(headerOutput);
      ByteBuffer headerBuf = ByteBuffer.wrap(headerOutput);
      Assert.assertEquals(messageHeaderVersionInUse, headerBuf.getShort());
      if (messageHeaderVersionInUse == MessageFormatRecord.Message_Header_Version_V3) {
        Assert.assertEquals(lifeVersion, headerBuf.getShort());
      }
      Assert.assertEquals(deleteRecordSize, headerBuf.getLong());
      // read encryption key relative offset
      if (messageHeaderVersionInUse >= MessageFormatRecord.Message_Header_Version_V2) {
        Assert.assertEquals(MessageFormatRecord.Message_Header_Invalid_Relative_Offset, headerBuf.getInt());
      }
      // blob properties relative offset
      Assert.assertEquals(MessageFormatRecord.Message_Header_Invalid_Relative_Offset, headerBuf.getInt());
      // delete record relative offset. This is the only relative offset with a valid value.
      Assert.assertEquals(headerSize + key.sizeInBytes(), headerBuf.getInt());
      // user metadata relative offset
      Assert.assertEquals(MessageFormatRecord.Message_Header_Invalid_Relative_Offset, headerBuf.getInt());
      // blob relative offset
      Assert.assertEquals(MessageFormatRecord.Message_Header_Invalid_Relative_Offset, headerBuf.getInt());
      CRC32 crc = new CRC32();
      crc.update(headerOutput, 0, headerSize - MessageFormatRecord.Crc_Size);
      Assert.assertEquals(crc.getValue(), headerBuf.getLong());

      // verify handle
      byte[] handleOutput = new byte[key.sizeInBytes()];
      messageFormatStream.read(handleOutput);
      Assert.assertArrayEquals(handleOutput, key.toBytes());

      // check delete record
      UpdateRecord updateRecord = MessageFormatRecord.deserializeUpdateRecord(messageFormatStream);
      Assert.assertEquals("Type of update record not DELETE", SubRecord.Type.DELETE, updateRecord.getType());
      Assert.assertNotNull("DeleteSubRecord should not be null", updateRecord.getDeleteSubRecord());
      Assert.assertEquals("AccountId mismatch", accountId, updateRecord.getAccountId());
      Assert.assertEquals("ContainerId mismatch", containerId, updateRecord.getContainerId());
      Assert.assertEquals("DeletionTime mismatch", deletionTimeMs, updateRecord.getUpdateTimeInMs());
    }
  }
Tests for {@link DeleteMessageFormatInputStream} in different versions.
ambry-messageformat/src/test/java/com/github/ambry/messageformat/MessageFormatInputStreamTest.java