@Test
  public void testS3MultiPartUpload() throws Exception {
    int chunkNumber = 9;
    // composite blob with "chunkNumber" of data chunks
    maxPutChunkSize = PUT_CONTENT_SIZE / chunkNumber;
    if (PUT_CONTENT_SIZE % maxPutChunkSize != 0) {
      maxPutChunkSize++;
    }
    setRouter();
    setOperationParams();

    int numParts = 2;
    Part parts[] = new Part[numParts];
    for (int part = 0; part < numParts; part++) {
      // set skipCompositeChunk to true
      RestRequest request = createRestRequestForPutOperation();
      request.setArg(RestUtils.InternalKeys.REQUEST_PATH, RequestPath.parse(request, null, CLUSTER_NAME));
      PutBlobOptions putBlobOptions = new PutBlobOptionsBuilder().restRequest(request).skipCompositeChunk(true).build();
      ReadableStreamChannel putChannel = new ByteBufferReadableStreamChannel(ByteBuffer.wrap(putContent));
      String compositeBlobInfo = router.putBlob(putBlobProperties, putUserMetadata, putChannel, putBlobOptions)
          .get(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS);
      PutBlobMetaInfo metaInfo = PutBlobMetaInfo.deserialize(compositeBlobInfo);

      // Serialize and Deserialize, the PutBlobMetaInfo should be the same
      String metaInfoStr = metaInfo.toString();
      PutBlobMetaInfo deserializedMetaInfo = PutBlobMetaInfo.deserialize(metaInfoStr);

      // verify the deserialized PutBlobMetaInfo
      Assert.assertEquals(chunkNumber, deserializedMetaInfo.getNumChunks());
      // verify the content, one data chunk after another
      List<Pair<String, Long>> chunks = deserializedMetaInfo.getOrderedChunkIdSizeList();
      // reserved metadata chunk id should be null
      Assert.assertNull(deserializedMetaInfo.getReservedMetadataChunkId());
      int chunkSize = maxPutChunkSize;
      for (int i = 0; i < chunkNumber; i++) {
        Pair<String, Long> chunk = chunks.get(i);
        GetBlobResult chunkResult =
            router.getBlob(chunk.getFirst(), new GetBlobOptionsBuilder().build(), null, null).get();
        // last chunk
        if (i == chunkNumber - 1) {
          chunkSize = PUT_CONTENT_SIZE - maxPutChunkSize * i;
        }
        Assert.assertEquals(chunkSize, chunk.getSecond().longValue());

        // verify the reserved metadata chunk id of each data chunk is null
        BlobProperties blobProperties = chunkResult.getBlobInfo().getBlobProperties();
        Assert.assertEquals(blobProperties.getReservedMetadataBlobId(), null);

        // verify data content
        RetainingAsyncWritableChannel retainingAsyncWritableChannel = new RetainingAsyncWritableChannel();
        chunkResult.getBlobDataChannel().readInto(retainingAsyncWritableChannel, null).get();
        InputStream input = retainingAsyncWritableChannel.consumeContentAsInputStream();
        retainingAsyncWritableChannel.close();
        Assert.assertEquals(chunkSize, input.available());
        byte[] readContent = Utils.readBytesFromStream(input, chunkSize);
        input.close();

        byte[] originSlice = Arrays.copyOfRange(putContent, i * maxPutChunkSize, i * maxPutChunkSize + chunkSize);
        Assert.assertArrayEquals(originSlice, readContent);
      }

      parts[part] = new Part(Integer.toString(part + 1), compositeBlobInfo);
    }

    // Complete Multipart Upload with the stitch command
    CompleteMultipartUpload completeMultipartUpload = new CompleteMultipartUpload(parts);

    RestRequest request = createRestRequestForPutOperation();
    PutBlobOptions options = new PutBlobOptionsBuilder().restRequest(request).build();
    List<ChunkInfo> chunksToStitch = getChunksToStitch(completeMultipartUpload);
    request.setArg(RestUtils.InternalKeys.REQUEST_PATH, RequestPath.parse(request, null, CLUSTER_NAME));
    String blobId = router.stitchBlob(putBlobProperties, putUserMetadata, chunksToStitch, options).get();

    router.getBlob(blobId, new GetBlobOptionsBuilder().build()).get();
  }
Test the S3 Multipart upload
1. Part Upload: Test the PutBlob with skipCompositeChunk option. We don't generate composite chunk for it.
Instead, we return PutBlobMetaInfo which includes the data chunk list.
No matter if router enables "reserved metadata chunk" feature, S3 part upload doesn't reserve metadata chunk.
2. Stitch Blob: It's called in completeMultipartUpload. It stitches all data chunks.
@throws Exception
ambry-router/src/test/java/com/github/ambry/router/S3NonBlockingRouterTest.java