public CompactOperator(
            SupplierWithException<FileSystem, IOException> fsFactory,
            CompactReader.Factory<T> readerFactory,
            CompactWriter.Factory<T> writerFactory) {
        this.fsFactory = fsFactory;
        this.readerFactory = readerFactory;
        this.writerFactory = writerFactory;
    }
@Test
    void testCompactOperator() throws Exception {
        AtomicReference<OperatorSubtaskState> state = new AtomicReference<>();
        Path f0 = newFile(".uncompacted-f0", 3);
        Path f1 = newFile(".uncompacted-f1", 2);
        Path f2 = newFile(".uncompacted-f2", 2);
        Path f3 = newFile(".uncompacted-f3", 5);
        Path f4 = newFile(".uncompacted-f4", 1);
        Path f5 = newFile(".uncompacted-f5", 5);
        Path f6 = newFile(".uncompacted-f6", 4);
        FileSystem fs = f0.getFileSystem();
        runCompact(
                harness -> {
                    harness.setup();
                    harness.open();

                    harness.processElement(
                            new CompactionUnit(0, "p0", Arrays.asList(f0, f1, f4)), 0);
                    harness.processElement(
                            new CompactionUnit(1, "p0", Collections.singletonList(f3)), 0);
                    harness.processElement(new CompactionUnit(2, "p1", Arrays.asList(f2, f5)), 0);
                    harness.processElement(
                            new CompactionUnit(3, "p0", Collections.singletonList(f6)), 0);

                    harness.processElement(new EndCompaction(1), 0);

                    state.set(harness.snapshot(2, 0));

                    // check output commit info
                    List<PartitionCommitInfo> outputs = harness.extractOutputValues();
                    assertThat(outputs).hasSize(1);
                    assertThat(outputs.get(0).getCheckpointId()).isEqualTo(1);
                    assertThat(outputs.get(0).getPartitions()).isEqualTo(new String[] {"p0", "p1"});

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

                    // check one compacted file
                    byte[] bytes =
                            FileUtils.readAllBytes(
                                    new File(folder.getPath(), "compacted-f0").toPath());
                    Arrays.sort(bytes);
                    assertThat(bytes).isEqualTo(new byte[] {0, 0, 0, 1, 1, 2});
                });

        runCompact(
                harness -> {
                    harness.setup();
                    harness.initializeState(state.get());
                    harness.open();

                    harness.notifyOfCompletedCheckpoint(2);

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