code: private void closePartFilesByTime(long currentProcessingTime) throws Exception {
  
  		synchronized (state.bucketStates) {
  			for (Map.Entry<String, BucketState<T>> entry : state.bucketStates.entrySet()) {
  				if ((entry.getValue().lastWrittenToTime < currentProcessingTime - inactiveBucketThreshold)
  						|| (entry.getValue().creationTime < currentProcessingTime - batchRolloverInterval)) {
  					LOG.debug("BucketingSink {} closing bucket due to inactivity of over {} ms.",
  						getRuntimeContext().getIndexOfThisSubtask(), inactiveBucketThreshold);
  					closeCurrentPartFile(entry.getValue());
  				}
  			}
  		}
  	}
docs: Checks for inactive buckets, and closes them. Buckets are considered inactive if they have not been
  written to for a period greater than {@code inactiveBucketThreshold} ms. Buckets are also closed if they are
  older than {@code batchRolloverInterval} ms. This enables in-progress files to be moved to the pending state
  and be finalised on the next checkpoint.