private void addOperatorRecursively(HashSet<OperatorImpl> s, OperatorImpl op) {
    List<OperatorImpl> operators = new ArrayList<>();
    operators.add(op);
    while (!operators.isEmpty()) {
      OperatorImpl opImpl = operators.remove(0);
      s.add(opImpl);
      if (!opImpl.registeredOperators.isEmpty()) {
        operators.addAll(opImpl.registeredOperators);
      }
    }
  }
the task assignment looks like the following:

input1 -----> task0, task1 -----> int1
                                   ^
input2 ------> task1, task2--------|
                                   v
input3 ------> task1 -----------> int2
samza-core/src/test/java/org/apache/samza/operators/impl/TestOperatorImplGraph.java