docs: private void openNewPartFile(Path bucketPath, BucketState<T> bucketState) throws Exception {
  		closeCurrentPartFile(bucketState);
  
  		if (!fs.exists(bucketPath)) {
  			try {
  				if (fs.mkdirs(bucketPath)) {
  					LOG.debug("Created new bucket directory: {}", bucketPath);
  				}
  			} catch (IOException e) {
  				throw new RuntimeException("Could not create new bucket path.", e);
  			}
  		}
  
  		// The following loop tries different partCounter values in ascending order until it reaches the minimum
  		// that is not yet used. This works since there is only one parallel subtask that tries names with this
  		// subtask id. Otherwise we would run into concurrency issues here. This is aligned with the way we now
  		// clean the base directory in case of rescaling.
  
  		int subtaskIndex = getRuntimeContext().getIndexOfThisSubtask();
  		Path partPath = assemblePartPath(bucketPath, subtaskIndex, bucketState.partCounter);
  		while (fs.exists(partPath) ||
  				fs.exists(getPendingPathFor(partPath)) ||
  				fs.exists(getInProgressPathFor(partPath))) {
  			bucketState.partCounter++;
  			partPath = assemblePartPath(bucketPath, subtaskIndex, bucketState.partCounter);
  		}
  
  		// Record the creation time of the bucket
  		bucketState.creationTime = processingTimeService.getCurrentProcessingTime();
  
  		// increase, so we don't have to check for this name next time
  		bucketState.partCounter++;
  
  		LOG.debug("Next part path is {}", partPath.toString());
  		bucketState.currentFile = partPath.toString();
  
  		Path inProgressPath = getInProgressPathFor(partPath);
  		if (bucketState.writer == null) {
  			bucketState.writer = writerTemplate.duplicate();
  			if (bucketState.writer == null) {
  				throw new UnsupportedOperationException(
  					"Could not duplicate writer. " +
  						"Class '" + writerTemplate.getClass().getCanonicalName() + "' must implement the 'Writer.duplicate()' method."
  				);
  			}
  		}
  
  		bucketState.writer.open(fs, inProgressPath);
  		bucketState.isWriterOpen = true;
  	}