@Override
    public void describe(StreamApplicationDescriptor appDescriptor) {
      DelegatingSystemDescriptor sd = new DelegatingSystemDescriptor(SYSTEM_NAME1);
      GenericInputDescriptor<KV<String, PageView>> isd =
          sd.getInputDescriptor(STREAM_ID1, KVSerde.of(new NoOpSerde<>(), new NoOpSerde<>()));
      appDescriptor.getInputStream(isd)
          .map(KV::getValue)
          .partitionBy(PageView::getMemberId, pv -> pv,
              KVSerde.of(new IntegerSerde(), new TestTableData.PageViewJsonSerde()), "p11")
          .map(kv -> KV.of(kv.getKey() * 31, kv.getValue()))
          .partitionBy(KV::getKey, KV::getValue, KVSerde.of(new IntegerSerde(), new TestTableData.PageViewJsonSerde()), "p21")
          .sink((m, collector, coordinator) -> {
            RECEIVED.add(m.getValue());
          });
    }
High-Level Job with multiple shuffle stages.
samza-test/src/test/java/org/apache/samza/test/drain/DrainHighLevelApiIntegrationTest.java