private void verifyCompositeBlob(BlobProperties properties, byte[] originalPutContent, byte[] originalUserMetadata,
      List<StoreKey> dataBlobIds, PutRequest request, HashMap<String, ByteBuffer> serializedRequests) throws Exception {
    StoreKey lastKey = dataBlobIds.get(dataBlobIds.size() - 1);
    byte[] content = new byte[(int) request.getBlobProperties().getBlobSize()];
    AtomicInteger offset = new AtomicInteger(0);
    for (StoreKey key : dataBlobIds) {
      PutRequest dataBlobPutRequest = deserializePutRequest(serializedRequests.get(key.getID()));
      AtomicInteger dataBlobLength = new AtomicInteger((int) dataBlobPutRequest.getBlobSize());
      InputStream dataBlobStream = dataBlobPutRequest.getBlobStream();
      if (!properties.isEncrypted()) {
        Utils.readBytesFromStream(dataBlobStream, content, offset.get(), dataBlobLength.get());
        Assert.assertArrayEquals("UserMetadata mismatch", originalUserMetadata,
            dataBlobPutRequest.getUsermetadata().array());
      } else {
        byte[] dataBlobContent = Utils.readBytesFromStream(dataBlobStream, dataBlobLength.get());
        // reason to directly call run() instead of spinning up a thread instead of calling start() is that, any exceptions or
        // assertion failures in non main thread will not fail the test.
        new DecryptJob(dataBlobPutRequest.getBlobId(), dataBlobPutRequest.getBlobEncryptionKey().duplicate(),
            Unpooled.wrappedBuffer(dataBlobContent), dataBlobPutRequest.getUsermetadata().duplicate(), cryptoService,
            kms, null, new CryptoJobMetricsTracker(metrics.decryptJobMetrics), (result, exception) -> {
          Assert.assertNull("Exception should not be thrown", exception);
          assertEquals("BlobId mismatch", dataBlobPutRequest.getBlobId(), result.getBlobId());
          Assert.assertArrayEquals("UserMetadata mismatch", originalUserMetadata,
              result.getDecryptedUserMetadata().array());
          ByteBuf decryptedBlobContent = result.getDecryptedBlobContent();
          dataBlobLength.set(decryptedBlobContent.readableBytes());
          decryptedBlobContent.readBytes(content, offset.get(), dataBlobLength.get());
          decryptedBlobContent.release();
        }).run();
      }
      if (metadataContentVersion <= MessageFormatRecord.Metadata_Content_Version_V2 && key != lastKey) {
        assertEquals("all chunks except last should be fully filled", chunkSize, dataBlobLength.get());
      } else if (key == lastKey) {
        assertEquals("Last chunk should be of non-zero length and equal to the length of the remaining bytes",
            (originalPutContent.length - 1) % chunkSize + 1, dataBlobLength.get());
      }
      offset.addAndGet(dataBlobLength.get());
      assertEquals("dataBlobStream should have no more data", -1, dataBlobStream.read());
      notificationSystem.verifyNotification(key.getID(), NotificationBlobType.DataChunk,
          dataBlobPutRequest.getBlobProperties());
    }
    Assert.assertArrayEquals("Input blob and written blob should be the same", originalPutContent, content);
  }
Verify Composite blob for content, userMetadata and
@param properties {@link BlobProperties} of the blob
@param originalPutContent original out content
@param originalUserMetadata original user-metadata
@param dataBlobIds {@link List} of {@link StoreKey}s of the composite blob in context
@param request {@link com.github.ambry.protocol.PutRequest} to fetch info from
@param serializedRequests the mapping from blob ids to their corresponding serialized {@link PutRequest}.
@throws Exception
ambry-router/src/test/java/com/github/ambry/router/PutManagerTest.java