### Question:
ExchangeOperator implements SourceOperator { @Override public void finish() { close(); } ExchangeOperator(
            OperatorContext operatorContext,
            PlanNodeId sourceId,
            PagesSerde serde,
            ExchangeClient exchangeClient); @Override PlanNodeId getSourceId(); @Override Supplier<Optional<UpdatablePageSource>> addSplit(Split split); @Override void noMoreSplits(); @Override OperatorContext getOperatorContext(); @Override void finish(); @Override boolean isFinished(); @Override ListenableFuture<?> isBlocked(); @Override boolean needsInput(); @Override void addInput(Page page); @Override Page getOutput(); @Override void close(); static final CatalogName REMOTE_CONNECTOR_ID; }### Answer:
@Test public void testFinish() throws Exception { SourceOperator operator = createExchangeOperator(); operator.addSplit(newRemoteSplit(TASK_1_ID)); operator.addSplit(newRemoteSplit(TASK_2_ID)); operator.addSplit(newRemoteSplit(TASK_3_ID)); operator.noMoreSplits(); taskBuffers.getUnchecked(TASK_1_ID).addPages(1, false); taskBuffers.getUnchecked(TASK_2_ID).addPages(1, false); taskBuffers.getUnchecked(TASK_3_ID).addPages(1, false); waitForPages(operator, 3); assertEquals(operator.isFinished(), false); assertEquals(operator.needsInput(), false); assertEquals(operator.getOutput(), null); operator.finish(); waitForFinished(operator); }