@Override
  public int write(ByteBuffer sourceBuffer) throws IOException {
    if (!isOpen()) {
      throw new ClosedChannelException();
    }

    int totalBytesWritten = 0;
    while (sourceBuffer.hasRemaining()) {
      int position = sourceBuffer.position();
      int bytesWritten = Math.min(sourceBuffer.remaining(), uploadBuffer.remaining());
      totalBytesWritten += bytesWritten;

      if (sourceBuffer.hasArray()) {
        // If the underlying array is accessible, direct access is the most efficient approach.
        int start = sourceBuffer.arrayOffset() + position;
        uploadBuffer.put(sourceBuffer.array(), start, bytesWritten);
        md5.update(sourceBuffer.array(), start, bytesWritten);
      } else {
        // Otherwise, use a readonly copy with an appropriate mark to read the current range of the
        // buffer twice.
        ByteBuffer copyBuffer = sourceBuffer.asReadOnlyBuffer();
        copyBuffer.mark().limit(position + bytesWritten);
        uploadBuffer.put(copyBuffer);
        copyBuffer.reset();
        md5.update(copyBuffer);
      }
      sourceBuffer.position(position + bytesWritten); // move position forward by the bytes written

      if (!uploadBuffer.hasRemaining() || sourceBuffer.hasRemaining()) {
        flush();
      }
    }

    return totalBytesWritten;
  }
@Test
  public void write() throws IOException {
    writeFromConfig(s3Config("s3"), false);
    writeFromConfig(s3Config("s3"), true);
    writeFromConfig(s3ConfigWithSSEAlgorithm("s3"), false);
    writeFromConfig(s3ConfigWithSSECustomerKey("s3"), false);
    writeFromConfig(s3ConfigWithSSEKMSKeyId("s3"), false);
    assertThrows(
        IllegalArgumentException.class,
        () -> writeFromConfig(s3ConfigWithMultipleSSEOptions("s3"), false));
  }