@Override
    public void initializeState(StateInitializationContext context) throws Exception {
        if (isPartitionCommitTriggerEnabled()) {
            partitionCommitPredicate =
                    PartitionCommitPredicate.create(conf, getUserCodeClassloader(), partitionKeys);
        }

        currentNewPartitions = new HashSet<>();
        newPartitions = new TreeMap<>();
        committablePartitions = new HashSet<>();
        inProgressPartitions = new HashMap<>();
        super.initializeState(context);
    }
@Test
    void testCommitFileWhenPartitionIsCommittableByProcessTime() throws Exception {
        // the rolling policy is not to roll file by filesize and roll file after one day,
        // it can ensure the file can be closed only when the partition is committable in this test.
        FileSystemTableSink.TableRollingPolicy tableRollingPolicy =
                new FileSystemTableSink.TableRollingPolicy(
                        false,
                        Long.MAX_VALUE,
                        Duration.ofDays(1).toMillis(),
                        Duration.ofDays(1).toMillis());
        List<String> partitionKeys = Collections.singletonList("d");
        // commit delay is 1 second with process-time trigger
        Configuration conf = getProcTimeCommitTriggerConf(Duration.ofSeconds(1).toMillis());
        OperatorSubtaskState state;
        long currentTimeMillis = System.currentTimeMillis();
        try (OneInputStreamOperatorTestHarness<RowData, PartitionCommitInfo> harness =
                create(tableRollingPolicy, partitionKeys, conf)) {
            harness.setup();
            harness.initializeEmptyState();
            harness.open();
            harness.setProcessingTime(currentTimeMillis);
            harness.processElement(row("1"), 0);
            harness.processElement(row("2"), 0);
            state = harness.snapshot(1, 1);
            harness.processElement(row("3"), 0);
            harness.notifyOfCompletedCheckpoint(1);
            // assert files aren't committed in {1, 2} partitions
            assertThat(isPartitionFileCommitted("1", 0, 0)).isFalse();
            assertThat(isPartitionFileCommitted("2", 0, 1)).isFalse();
        }

        // first retry
        try (OneInputStreamOperatorTestHarness<RowData, PartitionCommitInfo> harness =
                create(tableRollingPolicy, partitionKeys, conf)) {
            harness.setup();
            harness.initializeState(state);
            harness.open();
            harness.processElement(row("3"), 0);

            // simulate waiting for 2 seconds, now partition {3} is committable
            currentTimeMillis += Duration.ofSeconds(2).toMillis();
            harness.setProcessingTime(currentTimeMillis);
            harness.processElement(row("4"), 0);
            harness.snapshot(2, 2);
            harness.notifyOfCompletedCheckpoint(2);
            // only file in partition {3} should be committed
            // assert files are committed
            assertThat(isPartitionFileCommitted("3", 0, 2)).isTrue();
            assertThat(isPartitionFileCommitted("4", 0, 3)).isFalse();

            // simulate waiting for 2 seconds again, now partition {1} is committable
            currentTimeMillis += Duration.ofSeconds(2).toMillis();
            harness.setProcessingTime(currentTimeMillis);
            state = harness.snapshot(3, 3);
            harness.notifyOfCompletedCheckpoint(3);
            assertThat(isPartitionFileCommitted("4", 0, 3)).isTrue();
        }

        // second retry
        try (OneInputStreamOperatorTestHarness<RowData, PartitionCommitInfo> harness =
                create(tableRollingPolicy, partitionKeys, conf)) {
            harness.setup();
            harness.initializeState(state);
            harness.open();

            harness.processElement(row("4"), 0);
            harness.processElement(row("4"), 0);
            harness.snapshot(4, 4);
            harness.processElement(row("5"), 5);
            harness.endInput();
            // assert files in all partition have been committed
            assertThat(isPartitionFileCommitted("4", 0, 4)).isTrue();
            assertThat(isPartitionFileCommitted("5", 0, 5)).isTrue();
        }
    }