FlatmapOperatorImpl extends OperatorImpl<M, RM> { @Override protected void handleClose() { this.transformFn.close(); } FlatmapOperatorImpl(StreamOperatorSpec<M, RM> streamOpSpec);   }
@Test public void testStreamOperatorClose() { StreamOperatorSpec<TestMessageEnvelope, TestOutputMessageEnvelope> mockOp = mock(StreamOperatorSpec.class); FlatMapFunction<TestMessageEnvelope, TestOutputMessageEnvelope> txfmFn = mock(FlatMapFunction.class); when(mockOp.getTransformFn()).thenReturn(txfmFn); FlatmapOperatorImpl<TestMessageEnvelope, TestOutputMessageEnvelope> opImpl = new FlatmapOperatorImpl<>(mockOp); verify(txfmFn, times(0)).close(); opImpl.handleClose(); verify(txfmFn, times(1)).close(); }