flink-connectors/flink-connector-filesystem/src/main/java/org/apache/flink/streaming/connectors/fs/RollingSink.java
/**
	 * Opens a new part file.
	 *
	 * <p>
	 * This closes the old bucket file and retrieves a new bucket path from the {@code Bucketer}.
	 */
Opens a new part file.

This closes the old bucket file and retrieves a new bucket path from the  Bucketer.
Opens a new part file . This closes the old bucket file and retrieves a new bucket path from the Bucketer .
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);
		while (fs.exists(currentPartPath) ||
				fs.exists(getPendingPathFor(currentPartPath)) ||
				fs.exists(getInProgressPathFor(currentPartPath))) {
			partCounter++;
			currentPartPath = new Path(currentBucketDirectory, partPrefix + "-" + subtaskIndex + "-" + partCounter);
		}
		partCounter++;
		LOG.debug("Next part path is {}", currentPartPath.toString());
		Path inProgressPath = getInProgressPathFor(currentPartPath);
		writer.open(fs, inProgressPath);
		isWriterOpen = true;
	}
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 ) ; while ( fs . exists ( currentPartPath ) || fs . exists ( getPendingPathFor ( currentPartPath ) ) || fs . exists ( getInProgressPathFor ( currentPartPath ) ) ) { partCounter ++ ; currentPartPath = new Path ( currentBucketDirectory , partPrefix + "-" + subtaskIndex + "-" + partCounter ) ; } partCounter ++ ; LOG . debug ( "Next part path is {}" , currentPartPath . toString ( ) ) ; Path inProgressPath = getInProgressPathFor ( currentPartPath ) ; writer . open ( fs , inProgressPath ) ; isWriterOpen = true ; }
Opens a new part file.
Opens a new part file .
// 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. // increase, so we don't have to check for this name next time