@Override
    public void init(Context context) {
      TaskName taskName = context.getTaskContext().getTaskModel().getTaskName();
      if (perTaskFunctionMap.get(taskName) == null) {
        perTaskFunctionMap.put(taskName, new HashMap<>(Collections.singletonMap(opId, BaseTestFunction.this)));
      } else {
        if (perTaskFunctionMap.get(taskName).containsKey(opId)) {
          throw new IllegalStateException(String.format("Multiple init called for op %s in the same task instance %s", opId, this.taskName.getTaskName()));
        }
        perTaskFunctionMap.get(taskName).put(opId, this);
      }
      if (perTaskInitList.get(taskName) == null) {
        perTaskInitList.put(taskName, new ArrayList<>(Collections.singletonList(opId)));
      } else {
        perTaskInitList.get(taskName).add(opId);
      }
      this.taskName = taskName;
      this.numInitCalled++;
    }
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