flink-connectors/flink-connector-filesystem/src/main/java/org/apache/flink/streaming/connectors/fs/RollingSink.java
/**
	 * Determines whether we should change the bucket file we are writing to.
	 *
	 * <p>
	 * This will roll if no file was created yet, if the file size is larger than the specified size
	 * or if the {@code Bucketer} determines that we should roll.
	 */
Determines whether we should change the bucket file we are writing to.

This will roll if no file was created yet, if the file size is larger than the specified size
or if the  Bucketer determines that we should roll.
Determines whether we should change the bucket file we are writing to . This will roll if no file was created yet if the file size is larger than the specified size or if the Bucketer determines that we should roll .
private boolean shouldRoll() throws IOException {
		boolean shouldRoll = false;
		int subtaskIndex = getRuntimeContext().getIndexOfThisSubtask();
		if (!isWriterOpen) {
			shouldRoll = true;
			LOG.debug("RollingSink {} starting new initial bucket. ", subtaskIndex);
		}
		if (bucketer.shouldStartNewBucket(new Path(basePath), currentBucketDirectory)) {
			shouldRoll = true;
			LOG.debug("RollingSink {} starting new bucket because {} said we should. ", subtaskIndex, bucketer);
			partCounter = 0;
		}
		if (isWriterOpen) {
			long writePosition = writer.getPos();
			if (isWriterOpen && writePosition > batchSize) {
				shouldRoll = true;
				LOG.debug(
						"RollingSink {} starting new bucket because file position {} is above batch size {}.",
						subtaskIndex,
						writePosition,
						batchSize);
			}
		}
		return shouldRoll;
	}
private boolean shouldRoll ( ) throws IOException { boolean shouldRoll = false ; int subtaskIndex = getRuntimeContext ( ) . getIndexOfThisSubtask ( ) ; if ( ! isWriterOpen ) { shouldRoll = true ; LOG . debug ( "RollingSink {} starting new initial bucket. " , subtaskIndex ) ; } if ( bucketer . shouldStartNewBucket ( new Path ( basePath ) , currentBucketDirectory ) ) { shouldRoll = true ; LOG . debug ( "RollingSink {} starting new bucket because {} said we should. " , subtaskIndex , bucketer ) ; partCounter = 0 ; } if ( isWriterOpen ) { long writePosition = writer . getPos ( ) ; if ( isWriterOpen && writePosition > batchSize ) { shouldRoll = true ; LOG . debug ( "RollingSink {} starting new bucket because file position {} is above batch size {}." , subtaskIndex , writePosition , batchSize ) ; } } return shouldRoll ; }
Determines whether we should change the bucket file we are writing to.
Determines whether we should change the bucket file we are writing to .
// we will retrieve a new bucket base path in openNewPartFile so reset the part counter