private OperatorImpl createAndRegisterOperatorImpl(OperatorSpec prevOperatorSpec, OperatorSpec operatorSpec,
      SystemStream inputStream, Context context) {
    if (!operatorImpls.containsKey(operatorSpec.getOpId()) || operatorSpec instanceof JoinOperatorSpec) {
      // Either this is the first time we've seen this operatorSpec, or this is a join operator spec
      // and we need to create 2 partial join operator impls for it. Initialize and register the sub-DAG.
      OperatorImpl operatorImpl = createOperatorImpl(prevOperatorSpec, operatorSpec, context);
      operatorImpl.init(this.internalTaskContext);
      operatorImpl.registerInputStream(inputStream);

      if (operatorSpec.getScheduledFn() != null) {
        final Scheduler scheduler = operatorImpl.createOperatorScheduler();
        operatorSpec.getScheduledFn().schedule(scheduler);
      }

      // Note: The key here is opImplId, which may not equal opId for some impls (e.g. PartialJoinOperatorImpl).
      // This is currently OK since we don't need to look up a partial join operator impl again during traversal
      // (a join cannot have a cycle).
      operatorImpls.put(operatorImpl.getOpImplId(), operatorImpl);

      Collection<OperatorSpec> registeredSpecs = operatorSpec.getRegisteredOperatorSpecs();
      registeredSpecs.forEach(registeredSpec -> {
        LOG.debug("Creating operator {} with opCode: {}", registeredSpec.getOpId(), registeredSpec.getOpCode());
        OperatorImpl nextImpl = createAndRegisterOperatorImpl(operatorSpec, registeredSpec, inputStream, context);
        operatorImpl.registerNextOperator(nextImpl);
      });
      return operatorImpl;
    } else {
      // the implementation corresponding to operatorSpec has already been instantiated and registered.
      OperatorImpl operatorImpl = operatorImpls.get(operatorSpec.getOpId());
      operatorImpl.registerInputStream(inputStream);

      // We still need to traverse the DAG further to register the input streams.
      Collection<OperatorSpec> registeredSpecs = operatorSpec.getRegisteredOperatorSpecs();
      registeredSpecs.forEach(
        registeredSpec -> createAndRegisterOperatorImpl(operatorSpec, registeredSpec, inputStream, context));
      return operatorImpl;
    }
  }
Traverses the DAG of {@link OperatorSpec}s starting from the provided {@link OperatorSpec},
creates the corresponding DAG of {@link OperatorImpl}s, and returns the root {@link OperatorImpl} node.

@param prevOperatorSpec  the parent of the current {@code operatorSpec} in the traversal
@param operatorSpec  the {@link OperatorSpec} to create the {@link OperatorImpl} for
@param inputStream  the source input stream that we traverse the {@link OperatorSpecGraph} from
@param context the {@link Context} required to instantiate operators
@return  the operator implementation for the operatorSpec
samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImplGraph.java