### Question:
DynamicFilterService { public void stageCannotScheduleMoreTasks(StageId stageId, int numberOfTasks) { DynamicFilterContext context = dynamicFilterContexts.get(stageId.getQueryId()); if (context == null) { return; } context.stageCannotScheduleMoreTasks(stageId, numberOfTasks); executor.submit(() -> collectDynamicFilters(stageId, Optional.empty())); } @Inject  DynamicFilterService(DynamicFilterConfig dynamicFilterConfig); @VisibleForTesting  DynamicFilterService(ExecutorService executor); @PreDestroy void stop(); void registerQuery(SqlQueryExecution sqlQueryExecution, SubPlan fragmentedPlan); @VisibleForTesting void registerQuery(
            QueryId queryId,
            Set<DynamicFilterId> dynamicFilters,
            Set<DynamicFilterId> lazyDynamicFilters,
            Set<DynamicFilterId> replicatedDynamicFilters); DynamicFiltersStats getDynamicFilteringStats(QueryId queryId, Session session); void removeQuery(QueryId queryId); boolean isCollectingTaskNeeded(QueryId queryId, PlanFragment plan); void unblockStageDynamicFilters(QueryId queryId, PlanFragment plan); DynamicFilter createDynamicFilter(QueryId queryId, List<DynamicFilters.Descriptor> dynamicFilterDescriptors, Map<Symbol, ColumnHandle> columnHandles); void addTaskDynamicFilters(TaskId taskId, Map<DynamicFilterId, Domain> newDynamicFilters); void stageCannotScheduleMoreTasks(StageId stageId, int numberOfTasks);  }### Answer:
@Test public void testStageCannotScheduleMoreTasks() { DynamicFilterService dynamicFilterService = new DynamicFilterService(newDirectExecutorService()); DynamicFilterId filterId1 = new DynamicFilterId("df1"); Expression df1 = expression("DF_SYMBOL1"); QueryId queryId = new QueryId("query"); StageId stageId1 = new StageId(queryId, 1); dynamicFilterService.registerQuery( queryId, ImmutableSet.of(filterId1), ImmutableSet.of(filterId1), ImmutableSet.of()); DynamicFilter dynamicFilter = dynamicFilterService.createDynamicFilter( queryId, ImmutableList.of(new DynamicFilters.Descriptor(filterId1, df1)), ImmutableMap.of(Symbol.from(df1), new TestingColumnHandle("probeColumnA"))); assertTrue(dynamicFilter.getCurrentPredicate().isAll()); assertFalse(dynamicFilter.isComplete()); CompletableFuture<?> blockedFuture = dynamicFilter.isBlocked(); assertFalse(blockedFuture.isDone()); dynamicFilterService.addTaskDynamicFilters( new TaskId(stageId1, 0), ImmutableMap.of(filterId1, singleValue(INTEGER, 1L))); assertTrue(dynamicFilter.getCurrentPredicate().isAll()); assertFalse(dynamicFilter.isComplete()); assertFalse(blockedFuture.isDone()); dynamicFilterService.stageCannotScheduleMoreTasks(stageId1, 1); assertEquals(dynamicFilter.getCurrentPredicate(), TupleDomain.withColumnDomains(ImmutableMap.of( new TestingColumnHandle("probeColumnA"), singleValue(INTEGER, 1L)))); assertTrue(dynamicFilter.isComplete()); assertTrue(blockedFuture.isDone()); assertFalse(blockedFuture.isCompletedExceptionally()); }