code: 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();
  		}
  	}
docs: Close the operator's resources. They include the emitter thread and the executor to run
  the queue's complete operation.
  
  @param waitForShutdown is true if the method should wait for the resources to be freed;
  otherwise false.
  @throws InterruptedException if current thread has been interrupted