### Question:
Buckets { void onProcessingTime(long timestamp) throws Exception { for (Bucket<IN, BucketID> bucket : activeBuckets.values()) { bucket.onProcessingTime(timestamp); } } Buckets(
			final Path basePath,
			final BucketAssigner<IN, BucketID> bucketAssigner,
			final BucketFactory<IN, BucketID> bucketFactory,
			final PartFileWriter.PartFileFactory<IN, BucketID> partFileWriterFactory,
			final RollingPolicy<IN, BucketID> rollingPolicy,
			final int subtaskIndex); @VisibleForTesting long getMaxPartCounter();  }### Answer:
@Test public void testOnProcessingTime() throws Exception { final File outDir = TEMP_FOLDER.newFolder(); final Path path = new Path(outDir.toURI()); final OnProcessingTimePolicy<String, String> rollOnProcessingTimeCountingPolicy = new OnProcessingTimePolicy<>(2L); final Buckets<String, String> buckets = createBuckets(path, rollOnProcessingTimeCountingPolicy, 0); buckets.onElement("test", new TestUtils.MockSinkContext(1L, 2L, 3L)); buckets.onProcessingTime(7L); Assert.assertEquals(1L, rollOnProcessingTimeCountingPolicy.getOnProcessingTimeRollCounter()); final Map<String, Bucket<String, String>> activeBuckets = buckets.getActiveBuckets(); Assert.assertEquals(1L, activeBuckets.size()); Assert.assertTrue(activeBuckets.keySet().contains("test")); final Bucket<String, String> bucket = activeBuckets.get("test"); Assert.assertEquals("test", bucket.getBucketId()); Assert.assertEquals(new Path(path, "test"), bucket.getBucketPath()); Assert.assertEquals("test", bucket.getBucketId()); Assert.assertNull(bucket.getInProgressPart()); Assert.assertEquals(1L, bucket.getPendingPartsForCurrentCheckpoint().size()); Assert.assertTrue(bucket.getPendingPartsPerCheckpoint().isEmpty()); }