StreamApplication getRepartitionJoinStreamApplication() {
    return appDesc -> {
      MessageStream<KV<String, Object>> input1 = appDesc.getInputStream(input1Descriptor);
      MessageStream<KV<String, Object>> input2 = appDesc.getInputStream(input2Descriptor);
      OutputStream<KV<String, Object>> output = appDesc.getOutputStream(outputDescriptor);
      JoinFunction<String, Object, Object, KV<String, Object>> mockJoinFn = mock(JoinFunction.class);
      input1
          .partitionBy(KV::getKey, KV::getValue, defaultSerde, "p1")
          .map(kv -> kv.value)
          .join(input2.map(kv -> kv.value), mockJoinFn,
              new StringSerde(), new JsonSerdeV2<>(Object.class), new JsonSerdeV2<>(Object.class),
              Duration.ofHours(1), "j1")
          .sendTo(output);
    };
  }
Unit test base class to set up commonly used test application and configuration.
samza-core/src/test/java/org/apache/samza/execution/ExecutionPlannerTestBase.java