public ExecutionPlan plan(ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc) {
    validateConfig();

    // Create physical job graph based on stream graph
    JobGraph jobGraph = createJobGraph(appDesc);

    // Fetch the external streams partition info
    setInputAndOutputStreamPartitionCount(jobGraph);

    // Group streams participating in joins together into sets
    List<StreamSet> joinedStreamSets = groupJoinedStreams(jobGraph);

    // Set partitions of intermediate streams if any
    if (!jobGraph.getIntermediateStreamEdges().isEmpty()) {
      new IntermediateStreamManager(config).calculatePartitions(jobGraph, joinedStreamSets);
    }

    // Verify every group of joined streams has the same partition count
    joinedStreamSets.forEach(ExecutionPlanner::validatePartitions);

    return jobGraph;
  }
The ExecutionPlanner creates the physical execution graph for the {@link ApplicationDescriptorImpl}, and
the intermediate topics needed for the execution.
samza-core/src/main/java/org/apache/samza/execution/ExecutionPlanner.java