@Override
  public void run() {
    try {
      for (int i = 0; i < blobIds.size(); i++) {
        PutRequest putRequest =
            new PutRequest(1, "client1", blobIds.get(i), blobProperties, ByteBuffer.wrap(usermetadata),
                Unpooled.wrappedBuffer(data), blobProperties.getBlobSize(), BlobType.DataBlob,
                encryptionKey != null ? ByteBuffer.wrap(encryptionKey) : null);

        DataInputStream putResponseStream = channel.sendAndReceive(putRequest).getInputStream();
        PutResponse response = PutResponse.readFrom(putResponseStream);
        if (putResponseStream instanceof NettyByteBufDataInputStream) {
          // Release the underlying netty buf if the stream is NettyByteBuf based.
          // The ByteBuf should also be released if exception happens, but we ignore it since this is test and exception is not expected.
          ((NettyByteBufDataInputStream) putResponseStream).getBuffer().release();
        }
        Assert.assertEquals(response.getError(), ServerErrorCode.NoError);
      }
    } catch (Exception e) {
      Assert.assertTrue(false);
    } finally {
      endLatch.countDown();
    }
  }
Constructor for {@link DirectSender} object.
@param channel {@link BlockingChannel} object to connect to destination.
@param blobIds {@link List} of {@link BlobId}s to send.
@param data data data to put in each blob.
@param usermetadata usermetadata user metadata to put in each blob.
@param blobProperties {@link BlobProperties} object to add to each blob.
@param encryptionKey encryption key to encrypt blobs.
@param endLatch {@link CountDownLatch} object to signal all blobs have been uploaded.
ambry-test-utils/src/main/java/com/github/ambry/server/DirectSender.java