@Override
    public void onProcessingTime(long time) throws IOException {
        for (FileWriterBucket<IN> bucket : activeBuckets.values()) {
            bucket.onProcessingTime(time);
        }

        registerNextBucketInspectionTimer();
    }
@Test
    void testOnProcessingTime(@TempDir java.nio.file.Path tempDir) throws Exception {
        Path path = new Path(tempDir.toUri());

        // Create the processing timer service starts from 10.
        ManuallyTriggeredProcessingTimeService processingTimeService =
                new ManuallyTriggeredProcessingTimeService();
        processingTimeService.advanceTo(10);

        FileWriter<String> fileWriter =
                createWriter(
                        path,
                        new FileSinkTestUtils.StringIdentityBucketAssigner(),
                        DefaultRollingPolicy.builder()
                                .withRolloverInterval(Duration.ofMillis(10))
                                .build(),
                        new OutputFileConfig("part-", ""),
                        processingTimeService,
                        5);
        fileWriter.initializeState(Collections.emptyList());

        // Test timer registered timer@15 on startup
        fileWriter.write("test1", new ContextImpl());
        processingTimeService.advanceTo(15);
        fileWriter.write("test2", new ContextImpl());
        processingTimeService.advanceTo(20);

        FileWriterBucket<String> test1Bucket = fileWriter.getActiveBuckets().get("test1");

        assertThat(test1Bucket.getInProgressPart())
                .as("The in-progress part of test1 should be rolled")
                .isNull();
        assertThat(test1Bucket.getPendingFiles().size()).isEqualTo(1);

        FileWriterBucket<String> test2Bucket = fileWriter.getActiveBuckets().get("test2");

        assertThat(test2Bucket.getInProgressPart())
                .as("The in-progress part of test2 should not be rolled")
                .isNotNull();
        assertThat(test2Bucket.getPendingFiles().size()).isEqualTo(0);

        // Close, pre-commit & clear all the pending records.
        processingTimeService.advanceTo(30);
        fileWriter.prepareCommit();

        // Test timer re-registration.
        fileWriter.write("test1", new ContextImpl());
        processingTimeService.advanceTo(35);
        fileWriter.write("test2", new ContextImpl());
        processingTimeService.advanceTo(40);

        test1Bucket = fileWriter.getActiveBuckets().get("test1");

        assertThat(test1Bucket.getInProgressPart())
                .as("The in-progress part of test1 should be rolled")
                .isNull();
        assertThat(test1Bucket.getPendingFiles().size()).isEqualTo(1);

        test2Bucket = fileWriter.getActiveBuckets().get("test2");

        assertThat(test2Bucket.getInProgressPart())
                .as("The in-progress part of test2 should not be rolled")
                .isNotNull();
        assertThat(test2Bucket.getPendingFiles().size()).isEqualTo(0);
    }