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

        // see discussion: https://github.com/apache/flink/pull/15599#discussion_r623127365
        // only write the final blob if it doesn't already exist
        Optional<GSBlobStorage.BlobMetadata> blobMetadata =
                storage.getMetadata(recoverable.finalBlobIdentifier);
        if (!blobMetadata.isPresent()) {
            writeFinalBlob();
        }

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