@Override
    public void processElement(StreamRecord<CoordinatorOutput> element) throws Exception {
        CoordinatorOutput value = element.getValue();
        if (value instanceof CompactionUnit) {
            CompactionUnit unit = (CompactionUnit) value;
            if (unit.isTaskMessage(
                    getRuntimeContext().getTaskInfo().getNumberOfParallelSubtasks(),
                    getRuntimeContext().getTaskInfo().getIndexOfThisSubtask())) {
                String partition = unit.getPartition();
                List<Path> paths = unit.getPaths();
                // create a target file to compact to
                Path targetPath = createCompactedFile(paths);
                // do compaction
                CompactFileUtils.doCompact(
                        fileSystem,
                        partition,
                        paths,
                        targetPath,
                        getContainingTask()
                                .getEnvironment()
                                .getTaskManagerInfo()
                                .getConfiguration(),
                        readerFactory,
                        writerFactory);

                this.partitions.add(partition);

                // Only after the current checkpoint is successfully executed can delete
                // the expired files, so as to ensure the existence of the files.
                this.currentExpiredFiles.addAll(paths);
            }
        } else if (value instanceof EndCompaction) {
            endCompaction(((EndCompaction) value).getCheckpointId());
        }
    }
@Test
    void testUnitSelection() throws Exception {
        OneInputStreamOperatorTestHarness<CoordinatorOutput, PartitionCommitInfo> harness0 =
                create(2, 0);
        harness0.setup();
        harness0.open();

        OneInputStreamOperatorTestHarness<CoordinatorOutput, PartitionCommitInfo> harness1 =
                create(2, 1);
        harness1.setup();
        harness1.open();

        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();

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

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

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

        // f3 and f6 are in the charge of another task
        assertThat(fs.exists(new Path(folder, "compacted-f3"))).isFalse();
        assertThat(fs.exists(new Path(folder, "compacted-f6"))).isFalse();

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

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

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

        harness0.close();
        harness1.close();
    }