@Override
    public void close() {
      if (this.taskName == null) {
        throw new IllegalStateException("Close called before init");
      }
      if (perTaskFunctionMap.get(this.taskName) == null || !perTaskFunctionMap.get(this.taskName).containsKey(opId)) {
        throw new IllegalStateException("Close called before init");
      }

      if (perTaskCloseList.get(this.taskName) == null) {
        perTaskCloseList.put(taskName, new ArrayList<>(Collections.singletonList(opId)));
      } else {
        perTaskCloseList.get(taskName).add(opId);
      }

      this.numCloseCalled++;
    }
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