@Override
    public void commit() throws IOException {
        LOGGER.trace("Committing recoverable with options {}: {}", options, recoverable);

        // see discussion: https://github.com/apache/flink/pull/15599#discussion_r623127365
        // first, make sure the final blob doesn't already exist
        Optional<GSBlobStorage.BlobMetadata> blobMetadata =
                storage.getMetadata(recoverable.finalBlobIdentifier);
        if (blobMetadata.isPresent()) {
            throw new IOException(
                    String.format(
                            "Blob %s already exists during attempted commit",
                            recoverable.finalBlobIdentifier));
        }

        // write the final blob
        writeFinalBlob();

        // clean up after successful commit
        cleanupTemporaryBlobs();
    }
@Test(expected = IOException.class)
    public void commitOverwriteShouldFailTest() throws IOException {
        blobStorage.createBlob(blobIdentifier);
        GSRecoverableWriterCommitter committer = commitTestInternal();
        committer.commit();
    }