### Question:
SqlStageExecution { public boolean isAnyTaskBlocked() { return getTaskStatuses().stream().anyMatch(TaskStatus::isOutputBufferOverutilized); } private  SqlStageExecution(
            StageStateMachine stateMachine,
            RemoteTaskFactory remoteTaskFactory,
            NodeTaskMap nodeTaskMap,
            boolean summarizeTaskInfo,
            Executor executor,
            FailureDetector failureDetector,
            DynamicFilterService dynamicFilterService); static SqlStageExecution createSqlStageExecution(
            StageId stageId,
            PlanFragment fragment,
            Map<PlanNodeId, TableInfo> tables,
            RemoteTaskFactory remoteTaskFactory,
            Session session,
            boolean summarizeTaskInfo,
            NodeTaskMap nodeTaskMap,
            ExecutorService executor,
            FailureDetector failureDetector,
            DynamicFilterService dynamicFilterService,
            SplitSchedulerStats schedulerStats); StageId getStageId(); StageState getState(); void addStateChangeListener(StateChangeListener<StageState> stateChangeListener); void addFinalStageInfoListener(StateChangeListener<StageInfo> stateChangeListener); void addCompletedDriverGroupsChangedListener(Consumer<Set<Lifespan>> newlyCompletedDriverGroupConsumer); PlanFragment getFragment(); OutputBuffers getOutputBuffers(); void beginScheduling(); synchronized void transitionToSchedulingSplits(); synchronized void schedulingComplete(); synchronized void schedulingComplete(PlanNodeId partitionedSource); synchronized void cancel(); synchronized void abort(); long getUserMemoryReservation(); long getTotalMemoryReservation(); synchronized Duration getTotalCpuTime(); BasicStageStats getBasicStageStats(); StageInfo getStageInfo(); synchronized void addExchangeLocations(PlanFragmentId fragmentId, Set<RemoteTask> sourceTasks, boolean noMoreExchangeLocations); synchronized void setOutputBuffers(OutputBuffers outputBuffers); boolean hasTasks(); List<RemoteTask> getAllTasks(); synchronized Optional<RemoteTask> scheduleTask(InternalNode node, int partition, OptionalInt totalPartitions); synchronized Set<RemoteTask> scheduleSplits(InternalNode node, Multimap<PlanNodeId, Split> splits, Multimap<PlanNodeId, Lifespan> noMoreSplitsNotification); Set<InternalNode> getScheduledNodes(); void recordGetSplitTime(long start); List<TaskStatus> getTaskStatuses(); boolean isAnyTaskBlocked(); @Override String toString();  }### Answer:
@Test public void testIsAnyTaskBlocked() { NodeTaskMap nodeTaskMap = new NodeTaskMap(new FinalizerService()); StageId stageId = new StageId(new QueryId("query"), 0); SqlStageExecution stage = createSqlStageExecution( stageId, createExchangePlanFragment(), ImmutableMap.of(), new MockRemoteTaskFactory(executor, scheduledExecutor), TEST_SESSION, true, nodeTaskMap, executor, new NoOpFailureDetector(), new DynamicFilterService(new DynamicFilterConfig()), new SplitSchedulerStats()); stage.setOutputBuffers(createInitialEmptyOutputBuffers(ARBITRARY)); InternalNode node1 = new InternalNode("other1", URI.create("http: InternalNode node2 = new InternalNode("other2", URI.create("http: MockRemoteTask task1 = (MockRemoteTask) stage.scheduleTask(node1, 1, OptionalInt.empty()).get(); MockRemoteTask task2 = (MockRemoteTask) stage.scheduleTask(node2, 2, OptionalInt.empty()).get(); assertFalse(stage.isAnyTaskBlocked()); task1.setOutputBufferOverUtilized(true); assertTrue(stage.isAnyTaskBlocked()); task2.setOutputBufferOverUtilized(true); assertTrue(stage.isAnyTaskBlocked()); }