AsyncFlatmapOperatorImpl extends OperatorImpl<M, RM> { @Override protected void handleClose() { this.transformFn.close(); } AsyncFlatmapOperatorImpl(AsyncFlatMapOperatorSpec<M, RM> opSpec);   }
@Test public void testAsyncFlatMapOperatorClose() { AsyncFlatMapOperatorSpec<TestMessageEnvelope, TestOutputMessageEnvelope> mockOp = mock(AsyncFlatMapOperatorSpec.class); AsyncFlatMapFunction<TestMessageEnvelope, TestOutputMessageEnvelope> txfmFn = mock(AsyncFlatMapFunction.class); when(mockOp.getTransformFn()).thenReturn(txfmFn); AsyncFlatmapOperatorImpl<TestMessageEnvelope, TestOutputMessageEnvelope> opImpl = new AsyncFlatmapOperatorImpl<>(mockOp); verify(txfmFn, times(0)).close(); opImpl.handleClose(); verify(txfmFn, times(1)).close(); }