ConvenientSourceP extends AbstractProcessor { @Override public void close() { if (initialized) { destroyFn.accept(ctx); } } ConvenientSourceP(
            @Nonnull Function<? super Context, ? extends C> createFn,
            @Nonnull BiConsumer<? super C, ? super SourceBufferConsumerSide<?>> fillBufferFn,
            @Nonnull FunctionEx<? super C, ? extends S> createSnapshotFn,
            @Nonnull BiConsumerEx<? super C, ? super List<S>> restoreSnapshotFn,
            @Nonnull Consumer<? super C> destroyFn,
            @Nonnull SourceBufferConsumerSide<?> buffer,
            @Nullable EventTimePolicy<? super T> eventTimePolicy
    ); @Override boolean isCooperative(); @Override boolean complete(); @Override boolean saveToSnapshot(); @Override boolean finishSnapshotRestore(); @Override void close();  }
@Test public void when_streamingSourceClosesBuffer_then_fails() { StreamSource<Integer> source = SourceBuilder .stream("src", ctx -> null) .<Integer>fillBufferFn((src, buffer) -> buffer.close()) .distributed(1) .build(); exception.expectMessage("streaming source must not close the buffer"); TestSupport .verifyProcessor(((StreamSourceTransform<Integer>) source).metaSupplierFn.apply(noEventTime())) .expectOutput(Collections.emptyList()); }