@Override
    public void uploadPart(RefCountedFSOutputStream file) throws IOException {
        // this is to guarantee that nobody is
        // writing to the file we are uploading.
        checkState(file.isClosed());

        final CompletableFuture<PartETag> future = new CompletableFuture<>();
        uploadsInProgress.add(future);

        final long partLength = file.getPos();
        currentUploadInfo.registerNewPart(partLength);

        file.retain(); // keep the file while the async upload still runs
        uploadThreadPool.execute(new UploadTask(s3AccessHelper, currentUploadInfo, file, future));
    }
@Test
    public void multiplePartAndObjectUploadsShouldBeIncluded() throws IOException {
        final byte[] firstCompletePart = bytesOf("hello world");
        final byte[] secondCompletePart = bytesOf("hello again");
        final byte[] thirdIncompletePart = bytesOf("!!!");

        uploadPart(firstCompletePart);
        uploadPart(secondCompletePart);
        uploadObject(thirdIncompletePart);

        assertThat(
                stubMultiPartUploader,
                allOf(
                        hasMultiPartUploadWithPart(1, firstCompletePart),
                        hasMultiPartUploadWithPart(2, secondCompletePart),
                        hasUploadedObject(thirdIncompletePart)));
    }