@Test public void testSetsUserCodeClassLoaderForTimerThreadFactory() throws Throwable { syncLatch = new OneShotLatch(); try (MockEnvironment mockEnvironment = new MockEnvironmentBuilder() .setUserCodeClassLoader(new TestUserCodeClassLoader()) .build()) { TimeServiceTask timerServiceTask = new TimeServiceTask(mockEnvironment); CompletableFuture<Void> invokeFuture = CompletableFuture.runAsync( () -> { try { timerServiceTask.invoke(); } catch (Exception e) { throw new CompletionException(e); } }, TestingUtils.defaultExecutor()); invokeFuture.get(); assertThat(timerServiceTask.getClassLoaders(), hasSize(greaterThanOrEqualTo(1))); assertThat(timerServiceTask.getClassLoaders(), everyItem(instanceOf(TestUserCodeClassLoader.class))); } }
@Override public final void invoke() throws Exception { boolean disposed = false; try { LOG.debug("Initializing {}.", getName()); asyncOperationsThreadPool = Executors.newCachedThreadPool(); CheckpointExceptionHandlerFactory cpExceptionHandlerFactory = createCheckpointExceptionHandlerFactory(); synchronousCheckpointExceptionHandler = cpExceptionHandlerFactory.createCheckpointExceptionHandler( getExecutionConfig().isFailTaskOnCheckpointError(), getEnvironment()); asynchronousCheckpointExceptionHandler = new AsyncCheckpointExceptionHandler(this); stateBackend = createStateBackend(); checkpointStorage = stateBackend.createCheckpointStorage(getEnvironment().getJobID()); if (timerService == null) { ThreadFactory timerThreadFactory = new DispatcherThreadFactory(TRIGGER_THREAD_GROUP, "Time Trigger for " + getName(), getUserCodeClassLoader()); timerService = new SystemProcessingTimeService(this, getCheckpointLock(), timerThreadFactory); } operatorChain = new OperatorChain<>(this, recordWriters); headOperator = operatorChain.getHeadOperator(); init(); if (canceled) { throw new CancelTaskException(); } LOG.debug("Invoking {}", getName()); synchronized (lock) { initializeState(); openAllOperators(); } if (canceled) { throw new CancelTaskException(); } isRunning = true; run(); if (canceled) { throw new CancelTaskException(); } LOG.debug("Finished task {}", getName()); synchronized (lock) { closeAllOperators(); timerService.quiesce(); isRunning = false; } timerService.awaitPendingAfterQuiesce(); LOG.debug("Closed operators for task {}", getName()); operatorChain.flushOutputs(); tryDisposeAllOperators(); disposed = true; } finally { isRunning = false; setShouldInterruptOnCancel(false); Thread.interrupted(); tryShutdownTimerService(); try { cancelables.close(); shutdownAsyncThreads(); } catch (Throwable t) { LOG.error("Could not shut down async checkpoint threads", t); } try { cleanup(); } catch (Throwable t) { LOG.error("Error during cleanup of stream task", t); } if (!disposed) { disposeAllOperators(); } if (operatorChain != null) { synchronized (lock) { operatorChain.releaseOutputs(); } } } }
StreamTask extends AbstractInvokable implements AsyncExceptionHandler { @Override public final void invoke() throws Exception { boolean disposed = false; try { LOG.debug("Initializing {}.", getName()); asyncOperationsThreadPool = Executors.newCachedThreadPool(); CheckpointExceptionHandlerFactory cpExceptionHandlerFactory = createCheckpointExceptionHandlerFactory(); synchronousCheckpointExceptionHandler = cpExceptionHandlerFactory.createCheckpointExceptionHandler( getExecutionConfig().isFailTaskOnCheckpointError(), getEnvironment()); asynchronousCheckpointExceptionHandler = new AsyncCheckpointExceptionHandler(this); stateBackend = createStateBackend(); checkpointStorage = stateBackend.createCheckpointStorage(getEnvironment().getJobID()); if (timerService == null) { ThreadFactory timerThreadFactory = new DispatcherThreadFactory(TRIGGER_THREAD_GROUP, "Time Trigger for " + getName(), getUserCodeClassLoader()); timerService = new SystemProcessingTimeService(this, getCheckpointLock(), timerThreadFactory); } operatorChain = new OperatorChain<>(this, recordWriters); headOperator = operatorChain.getHeadOperator(); init(); if (canceled) { throw new CancelTaskException(); } LOG.debug("Invoking {}", getName()); synchronized (lock) { initializeState(); openAllOperators(); } if (canceled) { throw new CancelTaskException(); } isRunning = true; run(); if (canceled) { throw new CancelTaskException(); } LOG.debug("Finished task {}", getName()); synchronized (lock) { closeAllOperators(); timerService.quiesce(); isRunning = false; } timerService.awaitPendingAfterQuiesce(); LOG.debug("Closed operators for task {}", getName()); operatorChain.flushOutputs(); tryDisposeAllOperators(); disposed = true; } finally { isRunning = false; setShouldInterruptOnCancel(false); Thread.interrupted(); tryShutdownTimerService(); try { cancelables.close(); shutdownAsyncThreads(); } catch (Throwable t) { LOG.error("Could not shut down async checkpoint threads", t); } try { cleanup(); } catch (Throwable t) { LOG.error("Error during cleanup of stream task", t); } if (!disposed) { disposeAllOperators(); } if (operatorChain != null) { synchronized (lock) { operatorChain.releaseOutputs(); } } } } }
StreamTask extends AbstractInvokable implements AsyncExceptionHandler { @Override public final void invoke() throws Exception { boolean disposed = false; try { LOG.debug("Initializing {}.", getName()); asyncOperationsThreadPool = Executors.newCachedThreadPool(); CheckpointExceptionHandlerFactory cpExceptionHandlerFactory = createCheckpointExceptionHandlerFactory(); synchronousCheckpointExceptionHandler = cpExceptionHandlerFactory.createCheckpointExceptionHandler( getExecutionConfig().isFailTaskOnCheckpointError(), getEnvironment()); asynchronousCheckpointExceptionHandler = new AsyncCheckpointExceptionHandler(this); stateBackend = createStateBackend(); checkpointStorage = stateBackend.createCheckpointStorage(getEnvironment().getJobID()); if (timerService == null) { ThreadFactory timerThreadFactory = new DispatcherThreadFactory(TRIGGER_THREAD_GROUP, "Time Trigger for " + getName(), getUserCodeClassLoader()); timerService = new SystemProcessingTimeService(this, getCheckpointLock(), timerThreadFactory); } operatorChain = new OperatorChain<>(this, recordWriters); headOperator = operatorChain.getHeadOperator(); init(); if (canceled) { throw new CancelTaskException(); } LOG.debug("Invoking {}", getName()); synchronized (lock) { initializeState(); openAllOperators(); } if (canceled) { throw new CancelTaskException(); } isRunning = true; run(); if (canceled) { throw new CancelTaskException(); } LOG.debug("Finished task {}", getName()); synchronized (lock) { closeAllOperators(); timerService.quiesce(); isRunning = false; } timerService.awaitPendingAfterQuiesce(); LOG.debug("Closed operators for task {}", getName()); operatorChain.flushOutputs(); tryDisposeAllOperators(); disposed = true; } finally { isRunning = false; setShouldInterruptOnCancel(false); Thread.interrupted(); tryShutdownTimerService(); try { cancelables.close(); shutdownAsyncThreads(); } catch (Throwable t) { LOG.error("Could not shut down async checkpoint threads", t); } try { cleanup(); } catch (Throwable t) { LOG.error("Error during cleanup of stream task", t); } if (!disposed) { disposeAllOperators(); } if (operatorChain != null) { synchronized (lock) { operatorChain.releaseOutputs(); } } } } protected  StreamTask(Environment env); protected  StreamTask(
			Environment environment,
			@Nullable ProcessingTimeService timeProvider); }
StreamTask extends AbstractInvokable implements AsyncExceptionHandler { @Override public final void invoke() throws Exception { boolean disposed = false; try { LOG.debug("Initializing {}.", getName()); asyncOperationsThreadPool = Executors.newCachedThreadPool(); CheckpointExceptionHandlerFactory cpExceptionHandlerFactory = createCheckpointExceptionHandlerFactory(); synchronousCheckpointExceptionHandler = cpExceptionHandlerFactory.createCheckpointExceptionHandler( getExecutionConfig().isFailTaskOnCheckpointError(), getEnvironment()); asynchronousCheckpointExceptionHandler = new AsyncCheckpointExceptionHandler(this); stateBackend = createStateBackend(); checkpointStorage = stateBackend.createCheckpointStorage(getEnvironment().getJobID()); if (timerService == null) { ThreadFactory timerThreadFactory = new DispatcherThreadFactory(TRIGGER_THREAD_GROUP, "Time Trigger for " + getName(), getUserCodeClassLoader()); timerService = new SystemProcessingTimeService(this, getCheckpointLock(), timerThreadFactory); } operatorChain = new OperatorChain<>(this, recordWriters); headOperator = operatorChain.getHeadOperator(); init(); if (canceled) { throw new CancelTaskException(); } LOG.debug("Invoking {}", getName()); synchronized (lock) { initializeState(); openAllOperators(); } if (canceled) { throw new CancelTaskException(); } isRunning = true; run(); if (canceled) { throw new CancelTaskException(); } LOG.debug("Finished task {}", getName()); synchronized (lock) { closeAllOperators(); timerService.quiesce(); isRunning = false; } timerService.awaitPendingAfterQuiesce(); LOG.debug("Closed operators for task {}", getName()); operatorChain.flushOutputs(); tryDisposeAllOperators(); disposed = true; } finally { isRunning = false; setShouldInterruptOnCancel(false); Thread.interrupted(); tryShutdownTimerService(); try { cancelables.close(); shutdownAsyncThreads(); } catch (Throwable t) { LOG.error("Could not shut down async checkpoint threads", t); } try { cleanup(); } catch (Throwable t) { LOG.error("Error during cleanup of stream task", t); } if (!disposed) { disposeAllOperators(); } if (operatorChain != null) { synchronized (lock) { operatorChain.releaseOutputs(); } } } } protected  StreamTask(Environment env); protected  StreamTask(
			Environment environment,
			@Nullable ProcessingTimeService timeProvider); StreamTaskStateInitializer createStreamTaskStateInitializer(); @Override final void invoke(); @Override final void cancel(); final boolean isRunning(); final boolean isCanceled(); String getName(); Object getCheckpointLock(); CheckpointStorage getCheckpointStorage(); StreamConfig getConfiguration(); Map<String, Accumulator<?, ?>> getAccumulatorMap(); StreamStatusMaintainer getStreamStatusMaintainer(); @Override boolean triggerCheckpoint(CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions); @Override void triggerCheckpointOnBarrier(
			CheckpointMetaData checkpointMetaData,
			CheckpointOptions checkpointOptions,
			CheckpointMetrics checkpointMetrics); @Override void abortCheckpointOnBarrier(long checkpointId, Throwable cause); ExecutorService getAsyncOperationsThreadPool(); @Override void notifyCheckpointComplete(long checkpointId); ProcessingTimeService getProcessingTimeService(); @Override void handleAsyncException(String message, Throwable exception); @Override String toString(); CloseableRegistry getCancelables(); @VisibleForTesting static List<RecordWriter<SerializationDelegate<StreamRecord<OUT>>>> createRecordWriters(
			StreamConfig configuration,
			Environment environment); }
StreamTask extends AbstractInvokable implements AsyncExceptionHandler { @Override public final void invoke() throws Exception { boolean disposed = false; try { LOG.debug("Initializing {}.", getName()); asyncOperationsThreadPool = Executors.newCachedThreadPool(); CheckpointExceptionHandlerFactory cpExceptionHandlerFactory = createCheckpointExceptionHandlerFactory(); synchronousCheckpointExceptionHandler = cpExceptionHandlerFactory.createCheckpointExceptionHandler( getExecutionConfig().isFailTaskOnCheckpointError(), getEnvironment()); asynchronousCheckpointExceptionHandler = new AsyncCheckpointExceptionHandler(this); stateBackend = createStateBackend(); checkpointStorage = stateBackend.createCheckpointStorage(getEnvironment().getJobID()); if (timerService == null) { ThreadFactory timerThreadFactory = new DispatcherThreadFactory(TRIGGER_THREAD_GROUP, "Time Trigger for " + getName(), getUserCodeClassLoader()); timerService = new SystemProcessingTimeService(this, getCheckpointLock(), timerThreadFactory); } operatorChain = new OperatorChain<>(this, recordWriters); headOperator = operatorChain.getHeadOperator(); init(); if (canceled) { throw new CancelTaskException(); } LOG.debug("Invoking {}", getName()); synchronized (lock) { initializeState(); openAllOperators(); } if (canceled) { throw new CancelTaskException(); } isRunning = true; run(); if (canceled) { throw new CancelTaskException(); } LOG.debug("Finished task {}", getName()); synchronized (lock) { closeAllOperators(); timerService.quiesce(); isRunning = false; } timerService.awaitPendingAfterQuiesce(); LOG.debug("Closed operators for task {}", getName()); operatorChain.flushOutputs(); tryDisposeAllOperators(); disposed = true; } finally { isRunning = false; setShouldInterruptOnCancel(false); Thread.interrupted(); tryShutdownTimerService(); try { cancelables.close(); shutdownAsyncThreads(); } catch (Throwable t) { LOG.error("Could not shut down async checkpoint threads", t); } try { cleanup(); } catch (Throwable t) { LOG.error("Error during cleanup of stream task", t); } if (!disposed) { disposeAllOperators(); } if (operatorChain != null) { synchronized (lock) { operatorChain.releaseOutputs(); } } } } protected  StreamTask(Environment env); protected  StreamTask(
			Environment environment,
			@Nullable ProcessingTimeService timeProvider); StreamTaskStateInitializer createStreamTaskStateInitializer(); @Override final void invoke(); @Override final void cancel(); final boolean isRunning(); final boolean isCanceled(); String getName(); Object getCheckpointLock(); CheckpointStorage getCheckpointStorage(); StreamConfig getConfiguration(); Map<String, Accumulator<?, ?>> getAccumulatorMap(); StreamStatusMaintainer getStreamStatusMaintainer(); @Override boolean triggerCheckpoint(CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions); @Override void triggerCheckpointOnBarrier(
			CheckpointMetaData checkpointMetaData,
			CheckpointOptions checkpointOptions,
			CheckpointMetrics checkpointMetrics); @Override void abortCheckpointOnBarrier(long checkpointId, Throwable cause); ExecutorService getAsyncOperationsThreadPool(); @Override void notifyCheckpointComplete(long checkpointId); ProcessingTimeService getProcessingTimeService(); @Override void handleAsyncException(String message, Throwable exception); @Override String toString(); CloseableRegistry getCancelables(); @VisibleForTesting static List<RecordWriter<SerializationDelegate<StreamRecord<OUT>>>> createRecordWriters(
			StreamConfig configuration,
			Environment environment); static final ThreadGroup TRIGGER_THREAD_GROUP; }