@Override
      public void describe(StreamApplicationDescriptor appDescriptor) {
        DelegatingSystemDescriptor sd = new DelegatingSystemDescriptor("test");
        GenericInputDescriptor<KV<String, PageView>> isd =
            sd.getInputDescriptor("PageView", KVSerde.of(new NoOpSerde<>(), new NoOpSerde<>()));
        appDescriptor.getInputStream(isd)
            .map(KV::getValue)
            .partitionBy(PageView::getMemberId, pv -> pv,
                KVSerde.of(new IntegerSerde(), new TestTableData.PageViewJsonSerde()), "p1")
            .sink((m, collector, coordinator) -> {
              RECEIVED.add(m.getValue());
            });
      }
This test uses an array as a bounded input source, and does a partitionBy() and sink() after reading the input.
It verifies the pipeline will stop and the number of output messages should equal to the input.
samza-test/src/test/java/org/apache/samza/test/controlmessages/EndOfStreamIntegrationTest.java