private void testValidBlobs(short blobVersion, BlobType blobType, short headerVersionToUse) throws Exception {
    // MessageSievingInputStream contains put records for 3 valid blobs
    // id1(put record for valid blob), id2(put record for valid blob) and id3(put record for valid blob)

    MessageFormatRecord.headerVersionToUse = headerVersionToUse;
    byte[] encryptionKey = new byte[100];
    RANDOM.nextBytes(encryptionKey);
    // create message stream for blob 1
    StoreKey key1 = new MockId("id1");
    short accountId1 = Utils.getRandomShort(RANDOM);
    short containerId1 = Utils.getRandomShort(RANDOM);
    BlobProperties prop1 = new BlobProperties(10, "servid1", accountId1, containerId1, false);
    byte[] usermetadata1 = new byte[1000];
    RANDOM.nextBytes(usermetadata1);
    int blobContentSize = 2000;
    byte[] data1 = new byte[blobContentSize];
    RANDOM.nextBytes(data1);
    if (blobType != BlobType.DataBlob) {
      ByteBuffer byteBufferBlob = MessageFormatTestUtils.getBlobContentForMetadataBlob(blobContentSize);
      data1 = byteBufferBlob.array();
      blobContentSize = data1.length;
    }
    long blobSize = (blobVersion == Blob_Version_V3) ? (int) Blob_Format_V3.getBlobRecordSize(blobContentSize) :
        (int) Blob_Format_V2.getBlobRecordSize(blobContentSize);

    ByteBufferInputStream stream1 = new ByteBufferInputStream(ByteBuffer.wrap(data1));

    // For Blob_Version_V2, encryption key is null.
    MessageFormatInputStream messageFormatStream1 =
        (blobVersion >= Blob_Version_V2) ? new PutMessageFormatInputStream(key1, null, prop1,
            ByteBuffer.wrap(usermetadata1), stream1, blobContentSize, blobType)
            : new PutMessageFormatBlobV1InputStream(key1, prop1, ByteBuffer.wrap(usermetadata1), stream1,
                blobContentSize, blobType);

    MessageInfo msgInfo1 =
        new MessageInfo(key1, messageFormatStream1.getSize(), accountId1, containerId1, prop1.getCreationTimeInMs());

    // create message stream for blob 2
    StoreKey key2 = new MockId("id2");
    short accountId2 = Utils.getRandomShort(RANDOM);
    short containerId2 = Utils.getRandomShort(RANDOM);
    BlobProperties prop2 = new BlobProperties(10, "servid2", accountId2, containerId2, false);
    byte[] usermetadata2 = new byte[1000];
    RANDOM.nextBytes(usermetadata2);
    blobContentSize = 2000;
    byte[] data2 = new byte[blobContentSize];
    RANDOM.nextBytes(data2);
    if (blobType == BlobType.MetadataBlob) {
      ByteBuffer byteBufferBlob = MessageFormatTestUtils.getBlobContentForMetadataBlob(blobContentSize);
      data2 = byteBufferBlob.array();
      blobContentSize = data2.length;
    }
    ByteBufferInputStream stream2 = new ByteBufferInputStream(ByteBuffer.wrap(data2));

    // For Blob_Version_V2, encryption key is non-null.
    MessageFormatInputStream messageFormatStream2 =
        (blobVersion >= Blob_Version_V2) ? new PutMessageFormatInputStream(key2, ByteBuffer.wrap(encryptionKey), prop2,
            ByteBuffer.wrap(usermetadata2), stream2, blobContentSize, blobType)
            : new PutMessageFormatBlobV1InputStream(key2, prop2, ByteBuffer.wrap(usermetadata2), stream2,
                blobContentSize, blobType);

    MessageInfo msgInfo2 =
        new MessageInfo(key2, messageFormatStream2.getSize(), accountId2, containerId2, prop2.getCreationTimeInMs());

    // create message stream for blob 3
    StoreKey key3 = new MockId("id3");
    short accountId3 = Utils.getRandomShort(RANDOM);
    short containerId3 = Utils.getRandomShort(RANDOM);
    BlobProperties prop3 = new BlobProperties(10, "servid3", accountId3, containerId3, false);
    byte[] usermetadata3 = new byte[1000];
    RANDOM.nextBytes(usermetadata3);
    blobContentSize = 2000;
    byte[] data3 = new byte[blobContentSize];
    RANDOM.nextBytes(data3);
    if (blobType == BlobType.MetadataBlob) {
      ByteBuffer byteBufferBlob = MessageFormatTestUtils.getBlobContentForMetadataBlob(blobContentSize);
      data3 = byteBufferBlob.array();
      blobContentSize = data3.length;
    }
    ByteBufferInputStream stream3 = new ByteBufferInputStream(ByteBuffer.wrap(data3));

    // For Blob_Version_V2, encryption key is null.
    MessageFormatInputStream messageFormatStream3 =
        (blobVersion >= Blob_Version_V2) ? new PutMessageFormatInputStream(key3, null, prop3,
            ByteBuffer.wrap(usermetadata3), stream3, blobContentSize, blobType)
            : new PutMessageFormatBlobV1InputStream(key3, prop3, ByteBuffer.wrap(usermetadata3), stream3,
                blobContentSize, blobType);

    MessageInfo msgInfo3 =
        new MessageInfo(key3, messageFormatStream3.getSize(), accountId3, containerId3, prop3.getCreationTimeInMs());

    MessageInfo msgInfo4 = null;
    MessageFormatInputStream messageFormatStream4 = null;
    MessageInfo msgInfo5 = null;
    MessageFormatInputStream messageFormatStream5 = null;
    // create message stream for blob 4. Header version 2, with encryption key.
    StoreKey key4 = new MockId("id4");
    short accountId4 = Utils.getRandomShort(RANDOM);
    short containerId4 = Utils.getRandomShort(RANDOM);
    BlobProperties prop4 = new BlobProperties(10, "servid4", accountId4, containerId4, false);
    byte[] usermetadata4 = new byte[1000];
    RANDOM.nextBytes(usermetadata4);
    blobContentSize = 2000;
    byte[] data4 = new byte[blobContentSize];
    RANDOM.nextBytes(data4);
    if (blobType == BlobType.MetadataBlob) {
      ByteBuffer byteBufferBlob = MessageFormatTestUtils.getBlobContentForMetadataBlob(blobContentSize);
      data4 = byteBufferBlob.array();
      blobContentSize = data4.length;
    }

    if (blobVersion >= Blob_Version_V2) {
      ByteBufferInputStream stream4 = new ByteBufferInputStream(ByteBuffer.wrap(data4));
      MessageFormatRecord.headerVersionToUse = Message_Header_Version_V2;
      // encryption key is non-null.
      messageFormatStream4 =
          new PutMessageFormatInputStream(key4, ByteBuffer.wrap(encryptionKey), prop4, ByteBuffer.wrap(usermetadata4),
              stream4, blobContentSize, blobType);

      msgInfo4 =
          new MessageInfo(key4, messageFormatStream4.getSize(), accountId4, containerId4, prop4.getCreationTimeInMs());
    }

    // create message stream for blob 5. Header version 2, without encryption key.
    StoreKey key5 = new MockId("id5");
    short accountId5 = Utils.getRandomShort(RANDOM);
    short containerId5 = Utils.getRandomShort(RANDOM);
    BlobProperties prop5 = new BlobProperties(10, "servid5", accountId5, containerId5, false);
    byte[] usermetadata5 = new byte[1000];
    RANDOM.nextBytes(usermetadata5);
    blobContentSize = 2000;
    byte[] data5 = new byte[blobContentSize];
    RANDOM.nextBytes(data5);
    if (blobType == BlobType.MetadataBlob) {
      ByteBuffer byteBufferBlob = MessageFormatTestUtils.getBlobContentForMetadataBlob(blobContentSize);
      data5 = byteBufferBlob.array();
      blobContentSize = data5.length;
    }
    if (blobVersion >= Blob_Version_V2) {
      ByteBufferInputStream stream5 = new ByteBufferInputStream(ByteBuffer.wrap(data5));
      MessageFormatRecord.headerVersionToUse = Message_Header_Version_V2;
      // encryption key is null.
      messageFormatStream5 =
          new PutMessageFormatInputStream(key5, null, prop5, ByteBuffer.wrap(usermetadata5), stream5, blobContentSize,
              blobType);

      msgInfo5 =
          new MessageInfo(key5, messageFormatStream5.getSize(), accountId5, containerId5, prop5.getCreationTimeInMs());
    }

    //create input stream for all blob messages together
    byte[] totalMessageStreamContent =
        new byte[(int) (messageFormatStream1.getSize() + messageFormatStream2.getSize() + messageFormatStream3.getSize()
            + (blobVersion >= Blob_Version_V2 ? messageFormatStream4.getSize() + messageFormatStream5.getSize() : 0))];
    messageFormatStream1.read(totalMessageStreamContent, 0, (int) messageFormatStream1.getSize());
    messageFormatStream2.read(totalMessageStreamContent, (int) messageFormatStream1.getSize(),
        (int) messageFormatStream2.getSize());
    messageFormatStream3.read(totalMessageStreamContent,
        (int) messageFormatStream1.getSize() + (int) messageFormatStream2.getSize(),
        (int) messageFormatStream3.getSize());

    if (blobVersion >= Blob_Version_V2) {
      messageFormatStream4.read(totalMessageStreamContent,
          (int) messageFormatStream1.getSize() + (int) messageFormatStream2.getSize()
              + (int) messageFormatStream3.getSize(), (int) messageFormatStream4.getSize());
      messageFormatStream5.read(totalMessageStreamContent,
          (int) messageFormatStream1.getSize() + (int) messageFormatStream2.getSize()
              + (int) messageFormatStream3.getSize() + (int) messageFormatStream4.getSize(),
          (int) messageFormatStream5.getSize());
    }

    InputStream inputStream = new ByteBufferInputStream(ByteBuffer.wrap(totalMessageStreamContent));

    List<MessageInfo> msgInfoList = new ArrayList<MessageInfo>();
    msgInfoList.add(msgInfo1);
    msgInfoList.add(msgInfo2);
    msgInfoList.add(msgInfo3);
    if (blobVersion >= Blob_Version_V2) {
      msgInfoList.add(msgInfo4);
      msgInfoList.add(msgInfo5);
    }

    MessageSievingInputStream sievedStream =
        new MessageSievingInputStream(inputStream, msgInfoList, transformers, new MetricRegistry());

    Map<StoreKey, StoreKey> convertedMap = randomKeyConverter.convert(Arrays.asList(key1, key2, key3));
    Map<StoreKey, StoreKey> convertedMapExtra = randomKeyConverter.convert(Arrays.asList(key4, key5));
    int headerSizeV1 = MessageHeader_Format_V1.getHeaderSize();
    int headerSizeV2 = MessageHeader_Format_V2.getHeaderSize();
    int blobPropertiesRecordSize = BlobProperties_Format_V1.getBlobPropertiesRecordSize(prop1);
    int userMetadataSize = UserMetadata_Format_V1.getUserMetadataSize(ByteBuffer.wrap(usermetadata1));

    int totalHeadSize =
        3 * (headerVersionToUse == MessageFormatRecord.Message_Header_Version_V1 ? headerSizeV1 : headerSizeV2);
    int totalEncryptionRecordSize = 0;
    int totalBlobPropertiesSize = 3 * blobPropertiesRecordSize;
    int totalUserMetadataSize = 3 * userMetadataSize;
    int totalBlobSize = 3 * (int) blobSize;
    int totalKeySize = (options.contains(TransformerOptions.KeyConvert) ? convertedMap.values() : convertedMap.keySet())
        .stream()
        .mapToInt(StoreKey::sizeInBytes)
        .sum();
    int encryptionRecordSize =
        BlobEncryptionKey_Format_V1.getBlobEncryptionKeyRecordSize(ByteBuffer.wrap(encryptionKey));

    if (blobVersion >= Blob_Version_V2) {
      totalHeadSize += 2 * headerSizeV2;
      // stream 2 and stream 4 have encryption keys.
      totalEncryptionRecordSize += 2 * encryptionRecordSize;
      totalBlobPropertiesSize += 2 * blobPropertiesRecordSize;
      totalUserMetadataSize += 2 * userMetadataSize;
      totalBlobSize += 2 * (int) blobSize;
      totalKeySize += (options.contains(TransformerOptions.KeyConvert) ? convertedMapExtra.values()
          : convertedMapExtra.keySet()).stream().mapToInt(StoreKey::sizeInBytes).sum();
    }

    Assert.assertFalse(sievedStream.hasInvalidMessages());

    if (!options.isEmpty()) {
      Assert.assertEquals(options.isEmpty() ? totalMessageStreamContent.length
          : totalHeadSize + totalEncryptionRecordSize + totalBlobPropertiesSize + totalUserMetadataSize + totalBlobSize
              + totalKeySize, sievedStream.getSize());

      Assert.assertEquals((int) sievedStream.getValidMessageInfoList().stream().mapToLong(MessageInfo::getSize).sum(),
          sievedStream.getSize());

      verifySievedTransformedMessage(sievedStream,
          options.contains(TransformerOptions.KeyConvert) ? convertedMap.get(key1) : key1, "servid1", accountId1,
          containerId1, null, usermetadata1, data1, blobVersion, blobType);

      verifySievedTransformedMessage(sievedStream,
          options.contains(TransformerOptions.KeyConvert) ? convertedMap.get(key2) : key2, "servid2", accountId2,
          containerId2, blobVersion >= Blob_Version_V2 ? encryptionKey : null, usermetadata2, data2, blobVersion,
          blobType);

      verifySievedTransformedMessage(sievedStream,
          options.contains(TransformerOptions.KeyConvert) ? convertedMap.get(key3) : key3, "servid3", accountId3,
          containerId3, null, usermetadata3, data3, blobVersion, blobType);

      if (blobVersion >= Blob_Version_V2) {
        verifySievedTransformedMessage(sievedStream,
            options.contains(TransformerOptions.KeyConvert) ? convertedMapExtra.get(key4) : key4, "servid4", accountId4,
            containerId4, encryptionKey, usermetadata4, data4, blobVersion, blobType);

        verifySievedTransformedMessage(sievedStream,
            options.contains(TransformerOptions.KeyConvert) ? convertedMapExtra.get(key5) : key5, "servid5", accountId5,
            containerId5, null, usermetadata5, data5, blobVersion, blobType);
      }
    } else {
      Assert.assertEquals(totalMessageStreamContent.length, sievedStream.getSize());
      byte[] sievedBytes = Utils.readBytesFromStream(sievedStream, sievedStream.getSize());
      Assert.assertArrayEquals(totalMessageStreamContent, sievedBytes);
    }
    Assert.assertEquals(-1, sievedStream.read());
  }
Test the case where all messages are valid.
ambry-messageformat/src/test/java/com/github/ambry/messageformat/MessageSievingInputStreamTest.java