private void doPutRequestTest(short testVersion, MockClusterMap clusterMap, int correlationId, String clientId,
      BlobId blobId, BlobProperties blobProperties, byte[] userMetadata, BlobType blobType, byte[] blob, int blobSize,
      byte[] blobKey, byte[] expectedKey) throws IOException {
    // This PutRequest is created just to get the size.
    PutRequest putRequest =
        new PutRequest(correlationId, clientId, blobId, blobProperties, ByteBuffer.wrap(userMetadata),
            Unpooled.wrappedBuffer(blob), blobSize, blobType, blobKey == null ? null : ByteBuffer.wrap(blobKey));
    int sizeInBytes = (int) putRequest.sizeInBytes();
    putRequest.release();
    // Initialize channel write limits in such a way that writeTo() may or may not be able to write out all the
    // data at once.
    int channelWriteLimits[] =
        {sizeInBytes, 2 * sizeInBytes, sizeInBytes / 2, sizeInBytes / (TestUtils.RANDOM.nextInt(sizeInBytes - 1) + 1)};
    int sizeInBlobProperties = (int) blobProperties.getBlobSize();
    DataInputStream requestStream;
    for (int allocationSize : channelWriteLimits) {
      for (boolean useCompositeBlob : new boolean[]{true, false}) {
        PutRequest request = null;
        switch (testVersion) {
          case InvalidVersionPutRequest.Put_Request_Invalid_version:
            request = new InvalidVersionPutRequest(correlationId, clientId, blobId, blobProperties,
                ByteBuffer.wrap(userMetadata), Unpooled.wrappedBuffer(blob), sizeInBlobProperties, BlobType.DataBlob);
            requestStream = serAndPrepForRead(request, -1, true);
            try {
              PutRequest.readFrom(requestStream, clusterMap);
              Assert.fail("Deserialization of PutRequest with invalid version should have thrown an exception.");
            } catch (IllegalStateException e) {
            }
            request.release();
            break;
          default:
            if (request == null) {
              ByteBuf blobBuf = Unpooled.wrappedBuffer(blob);
              if (useCompositeBlob) {
                // break this into three ByteBuf and make a composite blob;
                int start = 0, end = blob.length / 3;
                ByteBuf blob1 = PooledByteBufAllocator.DEFAULT.heapBuffer(end - start);
                blob1.writeBytes(blob, start, end - start);
                start = end;
                end += blob.length / 3;
                ByteBuf blob2 = PooledByteBufAllocator.DEFAULT.heapBuffer(end - start);
                blob2.writeBytes(blob, start, end - start);
                start = end;
                end = blob.length;
                ByteBuf blob3 = PooledByteBufAllocator.DEFAULT.heapBuffer(end - start);
                blob3.writeBytes(blob, start, end - start);
                blobBuf = PooledByteBufAllocator.DEFAULT.compositeHeapBuffer(3);
                ((CompositeByteBuf) blobBuf).addComponent(true, blob1);
                ((CompositeByteBuf) blobBuf).addComponent(true, blob2);
                ((CompositeByteBuf) blobBuf).addComponent(true, blob3);
              }
              request = new PutRequest(correlationId, clientId, blobId, blobProperties, ByteBuffer.wrap(userMetadata),
                  blobBuf, blobSize, blobType, blobKey == null ? null : ByteBuffer.wrap(blobKey));
            }
            requestStream = serAndPrepForRead(request, allocationSize, true);
            PutRequest deserializedPutRequest = PutRequest.readFrom(requestStream, clusterMap);
            Assert.assertEquals(blobId, deserializedPutRequest.getBlobId());
            Assert.assertEquals(sizeInBlobProperties, deserializedPutRequest.getBlobProperties().getBlobSize());
            Assert.assertEquals(userMetadata.length, deserializedPutRequest.getUsermetadata().remaining());
            byte[] deserializedUserMetadata = new byte[userMetadata.length];
            deserializedPutRequest.getUsermetadata().get(deserializedUserMetadata);
            Assert.assertArrayEquals(userMetadata, deserializedUserMetadata);
            Assert.assertEquals(blobSize, deserializedPutRequest.getBlobSize());
            Assert.assertEquals(blobType, deserializedPutRequest.getBlobType());
            if (expectedKey == null) {
              Assert.assertNull(deserializedPutRequest.getBlobEncryptionKey());
            } else {
              Assert.assertArrayEquals(expectedKey, deserializedPutRequest.getBlobEncryptionKey().array());
            }
            byte[] blobRead = new byte[blobSize];
            deserializedPutRequest.getBlobStream().read(blobRead);
            Assert.assertArrayEquals(blob, blobRead);
            request.release();
            break;
        }
      }
    }
  }
Test Put requests in a specific version.
@param testVersion the version to use for the Put requests. If -1, uses the default version.
@param clusterMap the cluster map to use.
@param correlationId the correlation id associated with the request.
@param clientId the client id associated with the request.
@param blobId the blob id in the request.
@param blobProperties the {@link BlobProperties} associated with the request.
@param userMetadata the user metadata associated with the request.
@param blobType the {@link BlobType} associated with the request.
@param blob the blob content.
@param blobSize the size of the blob.
@param blobKey the encryption key of the blob.
@param expectedKey the expected encryption key from the deserialized Put request.
ambry-protocol/src/test/java/com/github/ambry/protocol/RequestResponseTest.java