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 ; }
Opens a new part file .