private StreamApplicationDescriptorImpl createStreamGraphWithInvalidStreamStreamJoin() {
    /**
     * Creates the following stream-stream join which is invalid due to partition count disagreement
     * between the 2 input streams.
     *
     *   input1 (64) --
     *                 |
     *                join -> output1 (8)
     *                 |
     *   input3 (32) --
     */
    return new StreamApplicationDescriptorImpl(appDesc -> {
      MessageStream<KV<Object, Object>> messageStream1 = appDesc.getInputStream(input1Descriptor);
      MessageStream<KV<Object, Object>> messageStream3 = appDesc.getInputStream(input3Descriptor);
      OutputStream<KV<Object, Object>> output1 = appDesc.getOutputStream(output1Descriptor);

      messageStream1
          .join(messageStream3,
              mock(JoinFunction.class), mock(Serde.class), mock(Serde.class), mock(Serde.class), Duration.ofHours(2), "j1")
          .sendTo(output1);
    }, config);
  }
the graph looks like the following. number of partitions in parentheses. quotes indicate expected value.

                              input1 (64) -> map -> join -> output1 (8)
                                                      |
         input2 (16) -> partitionBy ("64") -> filter -|
                                                      |
input3 (32) -> filter -> partitionBy ("64") -> map -> join -> output2 (16)
samza-core/src/test/java/org/apache/samza/execution/TestExecutionPlanner.java