### Question:
QueryTask implements Runnable { @Override public void run() { try { if (sourceAggregator.queryPromise.isDone()) { return; } if (!sourceAggregator.channelWritable) { sourceAggregator.needScheduling = true; return; } List<QueryTaskSource> taskSources = queryTaskSourceSupplier.get(); int bundlesProcessed = readBundles(taskSources, AggregateConfig.FRAME_READER_READS); if (bundlesProcessed > 0) { sourceAggregator.queryPromise.tryProgress(0, bundlesProcessed); } if (sourceAggregator.completed == sourceAggregator.totalTasks) { if (!sourceAggregator.queryPromise.trySuccess()) { log.warn("Tried to complete queryPromise {} , but failed", sourceAggregator.queryPromise); } } else { if (bundlesProcessed > 0) { pollFailures = 0; sourceAggregator.executor.execute(this); } else { pollFailures += 1; rescheduleSelfWithBackoff(); } } } catch (Throwable e) { if (!sourceAggregator.queryPromise.tryFailure(e)) { log.warn("Tried to fail queryPromise {} , but failed", sourceAggregator.queryPromise, e); } } } QueryTask(MeshSourceAggregator sourceAggregator); @Override void run();  }### Answer:
@Test public void basicSuccess() throws Exception { for (QueryTaskSource x: sourceAggregator.taskSources) { when(x.complete()).thenReturn(false).thenReturn(false).thenReturn(true); when(x.next()).thenReturn(mock(Bundle.class)); } when(sourceAggregator.queryPromise.trySuccess()).thenReturn(true); queryTask.run(); verify(sourceAggregator.consumer, times(4)).send(any(Bundle.class)); verify(sourceAggregator.queryPromise).tryProgress(0, 4); verify(sourceAggregator.queryPromise).trySuccess(); }
@Test public void noActiveTaskSource() throws Exception { for (QueryTaskSource x : sourceAggregator.taskSources) { when(x.next()).thenReturn(null); when(x.complete()).thenReturn(false); when(x.hasNoActiveSources()).thenReturn(true); } queryTask.run(); verify(sourceAggregator, times(4)).tryActivateSource(any(QueryTaskSource.class)); verify(sourceAggregator.executor).execute(queryTask); }
@Test public void oneTask() throws Exception { when(taskSource0.complete()).thenReturn(true); when(taskSource1.complete()).thenReturn(true); when(taskSource2.complete()).thenReturn(true); when(taskSource3.complete()).thenReturn(false).thenReturn(false).thenReturn(true); when(taskSource3.next()).thenReturn(mock(Bundle.class)); when(sourceAggregator.queryPromise.trySuccess()).thenReturn(true); queryTask.run(); verify(sourceAggregator.consumer).send(any(Bundle.class)); verify(sourceAggregator.queryPromise).tryProgress(0, 1); verify(sourceAggregator.queryPromise).trySuccess(); }