@Override
    public void close() {
        if (initialized) {
            destroyFn.accept(ctx);
        }
    }
@Test
    public void when_streamingSourceClosesBuffer_then_fails() {
        StreamSource<Integer> source = SourceBuilder
                .stream("src", ctx -> null)
                .<Integer>fillBufferFn((src, buffer) -> buffer.close())
                .distributed(1) // we use this to avoid forceTotalParallelismOne
                .build();

        assertThatThrownBy(() -> TestSupport
                .verifyProcessor(((StreamSourceTransform<Integer>) source).metaSupplierFn.apply(noEventTime()))
                .expectOutput(Collections.emptyList()))
                .hasMessageContaining("streaming source must not close the buffer");
    }