private void validateReachability() {
    // validate all nodes are reachable from the input streams
    final Set<JobNode> reachable = findReachable();
    if (reachable.size() != nodes.size()) {
      Set<JobNode> unreachable = new HashSet<>(nodes.values());
      unreachable.removeAll(reachable);
      throw new IllegalArgumentException(String.format("Jobs %s cannot be reached from Sources.",
          String.join(", ", unreachable.stream().map(JobNode::getJobNameAndId).collect(Collectors.toList()))));
    }
  }
Validate all nodes are reachable by input streams.
samza-core/src/main/java/org/apache/samza/execution/JobGraph.java