docs: 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());
  				}
  			}
  		}
  	}