@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 testCommitFileWhenPartitionIsCommittableByPartitionTime() 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 day with partition-time trigger
        Configuration conf = getPartitionCommitTriggerConf(Duration.ofDays(1).toMillis());

        long currentTimeMillis = System.currentTimeMillis();

        Date nextYear = new Date(currentTimeMillis + Duration.ofDays(365).toMillis());
        String nextYearPartition = "d=" + dateFormat.format(nextYear);
        Date yesterday = new Date(currentTimeMillis - Duration.ofDays(1).toMillis());
        String yesterdayPartition = "d=" + dateFormat.format(yesterday);
        Date today = new Date(currentTimeMillis);
        String todayPartition = "d=" + dateFormat.format(today);
        Date tomorrow = new Date(currentTimeMillis + Duration.ofDays(1).toMillis());
        String tomorrowPartition = "d=" + dateFormat.format(tomorrow);

        OperatorSubtaskState state;
        try (OneInputStreamOperatorTestHarness<RowData, PartitionCommitInfo> harness =
                create(tableRollingPolicy, partitionKeys, conf)) {
            harness.setup();
            harness.initializeEmptyState();
            harness.open();
            harness.processElement(row(yesterdayPartition), 0);
            harness.processWatermark(currentTimeMillis);
            state = harness.snapshot(1, 1);
            harness.notifyOfCompletedCheckpoint(1);
            // assert yesterday partition file is committed
            assertThat(isPartitionFileCommitted(yesterdayPartition, 0, 0)).isTrue();
        }

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

            // simulate waiting for 1 day
            currentTimeMillis += Duration.ofDays(1).toMillis();
            harness.processWatermark(currentTimeMillis);
            harness.snapshot(2, 2);
            harness.notifyOfCompletedCheckpoint(2);
            // assert today partition file is committed
            assertThat(isPartitionFileCommitted(todayPartition, 0, 2)).isTrue();
            // assert tomorrow partition file isn't committed
            assertThat(isPartitionFileCommitted(tomorrowPartition, 0, 1)).isFalse();

            // simulate waiting for 1 day again, now tomorrow partition is committable
            currentTimeMillis += Duration.ofDays(1).toMillis();
            harness.processWatermark(currentTimeMillis);
            state = harness.snapshot(3, 3);
            harness.notifyOfCompletedCheckpoint(3);
            assertThat(isPartitionFileCommitted(tomorrowPartition, 0, 1)).isTrue();

            harness.processElement(row(nextYearPartition), 0);
        }

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

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