@Test
  public void testBatchDeleteRequestResponseWithMultiplePartitions() throws IOException {
    final MockClusterMap clusterMap = new MockClusterMap();
    final short accountId = Utils.getRandomShort(TestUtils.RANDOM);
    final short containerId = Utils.getRandomShort(TestUtils.RANDOM);
    final String clientId = "client";
    //Partition 0
    final PartitionId partitionId0 = clusterMap.getWritablePartitionIds(MockClusterMap.DEFAULT_PARTITION_CLASS).get(0);
    final BlobId blobId1 = new BlobId(CommonTestUtils.getCurrentBlobIdVersion(), BlobId.BlobIdType.NATIVE,
        ClusterMap.UNKNOWN_DATACENTER_ID, accountId, containerId, partitionId0, false, BlobId.BlobDataType.DATACHUNK);
    final BlobId blobId2 = new BlobId(CommonTestUtils.getCurrentBlobIdVersion(), BlobId.BlobIdType.NATIVE,
        ClusterMap.UNKNOWN_DATACENTER_ID, accountId, containerId, partitionId0, false, BlobId.BlobDataType.DATACHUNK);

    //Partition 1
    final PartitionId partitionId1 = clusterMap.getWritablePartitionIds(MockClusterMap.DEFAULT_PARTITION_CLASS).get(1);
    final BlobId blobId3 = new BlobId(CommonTestUtils.getCurrentBlobIdVersion(), BlobId.BlobIdType.NATIVE,
        ClusterMap.UNKNOWN_DATACENTER_ID, accountId, containerId, partitionId1, false, BlobId.BlobDataType.DATACHUNK);
    final BlobId blobId4 = new BlobId(CommonTestUtils.getCurrentBlobIdVersion(), BlobId.BlobIdType.NATIVE,
        ClusterMap.UNKNOWN_DATACENTER_ID, accountId, containerId, partitionId1, false, BlobId.BlobDataType.DATACHUNK);

    final int correlationId = TestUtils.RANDOM.nextInt();
    final RequestOrResponseType operationType =
        RequestOrResponseType.values()[Utils.getRandomShort(TestUtils.RANDOM) % RequestOrResponseType.values().length];
    final long deletionTimeInMs = 1L;

    //Delete Partition Request Info for Partition 0
    final BatchDeletePartitionRequestInfo batchDeletePartitionRequestInfo0 =
        new BatchDeletePartitionRequestInfo(partitionId0, Arrays.asList(blobId1, blobId2));
    //Delete Partition Request Info for Partition 1
    final BatchDeletePartitionRequestInfo batchDeletePartitionRequestInfo1 =
        new BatchDeletePartitionRequestInfo(partitionId1, Arrays.asList(blobId3, blobId4));

    //Batch Delete Request
    final BatchDeleteRequest batchDeleteRequest = new BatchDeleteRequest(correlationId, clientId,
        Arrays.asList(batchDeletePartitionRequestInfo0, batchDeletePartitionRequestInfo1), deletionTimeInMs);

    DataInputStream requestStream = serAndPrepForRead(batchDeleteRequest, -1, true);
    final BatchDeleteRequest deserializedBatchDeleteRequest = BatchDeleteRequest.readFrom(requestStream, clusterMap);
    //verify the request
    verifyBatchDeleteRequest(batchDeleteRequest, deserializedBatchDeleteRequest);
    batchDeleteRequest.release();

    //Delete Partition Response Info for Partition 0
    final BatchDeletePartitionResponseInfo batchDeletePartitionResponseInfo0 =
        new BatchDeletePartitionResponseInfo(partitionId0,
            Arrays.asList(new BlobDeleteStatus(blobId1, ServerErrorCode.NoError),
                new BlobDeleteStatus(blobId2, ServerErrorCode.BlobNotFound)));
    //Delete Partition Response Info for Partition 1
    final BatchDeletePartitionResponseInfo batchDeletePartitionResponseInfo1 =
        new BatchDeletePartitionResponseInfo(partitionId1,
            Arrays.asList(new BlobDeleteStatus(blobId3, ServerErrorCode.BlobDeleted),
                new BlobDeleteStatus(blobId4, ServerErrorCode.BlobNotDeleted)));
    //Batch Delete Response
    final BatchDeleteResponse batchDeleteResponse = new BatchDeleteResponse((short) 1, correlationId, clientId,
        Arrays.asList(batchDeletePartitionResponseInfo0, batchDeletePartitionResponseInfo1), ServerErrorCode.NoError);
    requestStream = serAndPrepForRead(batchDeleteResponse, -1, false);
    final BatchDeleteResponse deserializedBatchDeleteResponse = BatchDeleteResponse.readFrom(requestStream, clusterMap);
    //verify the response
    verifyBatchDeleteResponse(batchDeleteResponse, deserializedBatchDeleteResponse);
    batchDeleteResponse.release();
  }
Tests for {@link BatchDeleteRequest} and {@link BatchDeleteResponse} with multiple partitions.
@throws IOException
ambry-protocol/src/test/java/com/github/ambry/protocol/RequestResponseTest.java