SinkOperatorImpl extends OperatorImpl<M, Void> { @Override protected void handleClose() { this.sinkFn.close(); } SinkOperatorImpl(SinkOperatorSpec<M> sinkOpSpec);   }
@Test public void testSinkOperatorClose() { TestOutputMessageEnvelope mockMsg = mock(TestOutputMessageEnvelope.class); MessageCollector mockCollector = mock(MessageCollector.class); TaskCoordinator mockCoordinator = mock(TaskCoordinator.class); SinkFunction<TestOutputMessageEnvelope> sinkFn = mock(SinkFunction.class); SinkOperatorImpl<TestOutputMessageEnvelope> sinkImpl = createSinkOperator(sinkFn); sinkImpl.handleMessage(mockMsg, mockCollector, mockCoordinator); verify(sinkFn, times(1)).apply(mockMsg, mockCollector, mockCoordinator); verify(sinkFn, times(0)).close(); sinkImpl.handleClose(); verify(sinkFn, times(1)).close(); }