@Override
    public void commit(Collection<CommitRequest<FileSinkCommittable>> requests)
            throws IOException, InterruptedException {
        for (CommitRequest<FileSinkCommittable> request : requests) {
            FileSinkCommittable committable = request.getCommittable();
            if (committable.hasPendingFile()) {
                // We should always use commitAfterRecovery which contains additional checks.
                bucketWriter.recoverPendingFile(committable.getPendingFile()).commitAfterRecovery();
            }

            if (committable.hasInProgressFileToCleanup()) {
                bucketWriter.cleanupInProgressFileRecoverable(
                        committable.getInProgressFileToCleanup());
            }

            if (committable.hasCompactedFileToCleanup()) {
                Path committedFileToCleanup = committable.getCompactedFileToCleanup();
                try {
                    committedFileToCleanup.getFileSystem().delete(committedFileToCleanup, false);
                } catch (Exception e) {
                    // Try best to cleanup compacting files, skip if failed.
                    if (LOG.isDebugEnabled()) {
                        LOG.debug(
                                "Failed to cleanup a compacted file, the file will be remained and should not be visible: {}",
                                committedFileToCleanup,
                                e);
                    }
                }
            }
        }
    }
@Test
    void testCommitPendingFile() throws Exception {
        StubBucketWriter stubBucketWriter = new StubBucketWriter();
        FileCommitter fileCommitter = new FileCommitter(stubBucketWriter);

        MockCommitRequest<FileSinkCommittable> fileSinkCommittable =
                new MockCommitRequest<>(
                        new FileSinkCommittable(
                                "0", new FileSinkTestUtils.TestPendingFileRecoverable()));
        fileCommitter.commit(Collections.singletonList(fileSinkCommittable));

        assertThat(stubBucketWriter.getRecoveredPendingFiles()).hasSize(1);
        assertThat(stubBucketWriter.getNumCleanUp()).isEqualTo(0);
        assertThat(stubBucketWriter.getRecoveredPendingFiles().get(0).isCommitted()).isTrue();
        assertThat(fileSinkCommittable.getNumberOfRetries()).isEqualTo(0);
    }