docs: private void openNewPartFile() throws Exception {
  		closeCurrentPartFile();
  
  		Path newBucketDirectory = bucketer.getNextBucketPath(new Path(basePath));
  
  		if (!newBucketDirectory.equals(currentBucketDirectory)) {
  			currentBucketDirectory = newBucketDirectory;
  			try {
  				if (fs.mkdirs(currentBucketDirectory)) {
  					LOG.debug("Created new bucket directory: {}", currentBucketDirectory);
  				}
  			} catch (IOException e) {
  				throw new RuntimeException("Could not create base path for new rolling file.", e);
  			}
  		}
  
  		int subtaskIndex = getRuntimeContext().getIndexOfThisSubtask();
  		currentPartPath = new Path(currentBucketDirectory, partPrefix + "-" + subtaskIndex + "-" + partCounter);
  
  		// This should work since there is only one parallel subtask that tries names with
  		// our subtask id. Otherwise we would run into concurrency issues here.
  		while (fs.exists(currentPartPath) ||
  				fs.exists(getPendingPathFor(currentPartPath)) ||
  				fs.exists(getInProgressPathFor(currentPartPath))) {
  			partCounter++;
  			currentPartPath = new Path(currentBucketDirectory, partPrefix + "-" + subtaskIndex + "-" + partCounter);
  		}
  
  		// increase, so we don't have to check for this name next time
  		partCounter++;
  
  		LOG.debug("Next part path is {}", currentPartPath.toString());
  
  		Path inProgressPath = getInProgressPathFor(currentPartPath);
  		writer.open(fs, inProgressPath);
  		isWriterOpen = true;
  	}