@Override
    public void endInput() throws Exception {
        endCompaction(Long.MAX_VALUE);
        snapshotState(Long.MAX_VALUE);
        clearExpiredFiles(Long.MAX_VALUE);
    }
@Test
    void testEndInput() throws Exception {
        Path f0 = newFile(".uncompacted-f0", 3);
        Path f1 = newFile(".uncompacted-f1", 4);
        Path f2 = newFile(".uncompacted-f2", 2);

        FileSystem fs = f0.getFileSystem();

        runCompact(
                harness -> {
                    harness.setup();
                    harness.open();

                    harness.processElement(new CompactionUnit(0, "p0", Arrays.asList(f0, f1)), 0);
                    harness.processElement(
                            new CompactionUnit(1, "p0", Collections.singletonList(f2)), 0);

                    // test without snapshot
                    harness.endInput();

                    // check all compacted file generated
                    assertThat(fs.exists(new Path(folder, "compacted-f0"))).isTrue();
                    assertThat(fs.exists(new Path(folder, "compacted-f2"))).isTrue();

                    // check all temp files have been deleted
                    assertThat(fs.exists(f0)).isFalse();
                    assertThat(fs.exists(f1)).isFalse();
                    assertThat(fs.exists(f2)).isFalse();
                });
    }