private void stopResources ( boolean waitForShutdown ) throws InterruptedException { emitter . stop ( ) ; emitterThread . interrupt ( ) ; executor . shutdown ( ) ; if ( waitForShutdown ) { try { if ( ! executor . awaitTermination ( 365L , TimeUnit . DAYS ) ) { executor . shutdownNow ( ) ; } } catch ( InterruptedException e ) { executor . shutdownNow ( ) ; Thread . currentThread ( ) . interrupt ( ) ; } /* * FLINK-5638: If we have the checkpoint lock we might have to free it for a while so * that the emitter thread can complete/react to the interrupt signal. */ if ( Thread . holdsLock ( checkpointingLock ) ) { while ( emitterThread . isAlive ( ) ) { checkpointingLock . wait ( 100L ) ; } } emitterThread . join ( ) ; } else { executor . shutdownNow ( ) ; } }
Close the operator s resources . They include the emitter thread and the executor to run the queue s complete operation .