@Override
    public void describe(StreamApplicationDescriptor appDescriptor) {
      DelegatingSystemDescriptor sd = new DelegatingSystemDescriptor(SYSTEM_NAME2);
      GenericInputDescriptor<KV<String, PageView>> isd =
          sd.getInputDescriptor(STREAM_ID2, KVSerde.of(new NoOpSerde<>(), new NoOpSerde<>()));
      appDescriptor.getInputStream(isd)
          .sink((m, collector, coordinator) -> {
            RECEIVED.add(m.getValue());
          });
    }
Simple high-level application without shuffle stages.
samza-test/src/test/java/org/apache/samza/test/drain/DrainHighLevelApiIntegrationTest.java