private void validateInputStreams() {
    inputStreams.forEach(edge -> {
      if (!edge.getSourceNodes().isEmpty()) {
        throw new IllegalArgumentException(
            String.format("Source stream %s should not have producers.", edge.getName()));
      }
      if (edge.getTargetNodes().isEmpty()) {
        throw new IllegalArgumentException(
            String.format("Source stream %s should have consumers.", edge.getName()));
      }
    });
  }
Validate the input streams should have indegree being 0 and outdegree greater than 0
samza-core/src/main/java/org/apache/samza/execution/JobGraph.java