### Question:
StreamTask extends AbstractInvokable implements AsyncExceptionHandler { public final boolean isRunning() { return isRunning; } protected  StreamTask(Environment env); protected  StreamTask(Environment env, @Nullable ProcessingTimeService timeProvider); protected  StreamTask(
			Environment environment,
			@Nullable ProcessingTimeService timeProvider,
			Thread.UncaughtExceptionHandler uncaughtExceptionHandler); StreamTaskStateInitializer createStreamTaskStateInitializer(); @Override final void invoke(); @Override final void cancel(); MailboxExecutor getTaskMailboxExecutor(); final boolean isRunning(); final boolean isCanceled(); String getName(); Object getCheckpointLock(); CheckpointStorageWorkerView getCheckpointStorage(); StreamConfig getConfiguration(); Map<String, Accumulator<?, ?>> getAccumulatorMap(); StreamStatusMaintainer getStreamStatusMaintainer(); @Override boolean triggerCheckpoint(
			CheckpointMetaData checkpointMetaData,
			CheckpointOptions checkpointOptions,
			boolean advanceToEndOfEventTime); @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; }### Answer:
@Test public void testOperatorClosingBeforeStopRunning() throws Throwable { BlockingCloseStreamOperator.resetLatches(); Configuration taskConfiguration = new Configuration(); StreamConfig streamConfig = new StreamConfig(taskConfiguration); streamConfig.setStreamOperator(new BlockingCloseStreamOperator()); streamConfig.setOperatorID(new OperatorID()); try (MockEnvironment mockEnvironment = new MockEnvironmentBuilder() .setTaskName("Test Task") .setMemorySize(32L * 1024L) .setInputSplitProvider(new MockInputSplitProvider()) .setBufferSize(1) .setTaskConfiguration(taskConfiguration) .build()) { RunningTask<StreamTask<Void, BlockingCloseStreamOperator>> task = runTask(() -> new NoOpStreamTask<>(mockEnvironment)); BlockingCloseStreamOperator.inClose.await(); assertTrue(task.streamTask.isRunning()); BlockingCloseStreamOperator.finishClose.trigger(); task.waitForTaskCompletion(false); assertFalse(task.streamTask.isRunning()); } }