class 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,
			final PartFileConfig partFileConfig);

}