private void validateInternalStreams() {
    Set<StreamEdge> internalEdges = new HashSet<>(edges.values());
    internalEdges.removeAll(inputStreams);
    internalEdges.removeAll(sideInputStreams);
    internalEdges.removeAll(outputStreams);

    internalEdges.forEach(edge -> {
      if (edge.getSourceNodes().isEmpty() || edge.getTargetNodes().isEmpty()) {
        throw new IllegalArgumentException(
            String.format("Internal stream %s should have both producers and consumers", edge.getName()));
      }
    });
  }
Validate the internal streams should have both indegree and outdegree greater than 0
samza-core/src/main/java/org/apache/samza/execution/JobGraph.java