void registerInputStream(SystemStream input) {
    this.inputStreams.add(input);

    usedInCurrentTask = usedInCurrentTask
        || taskModel.getSystemStreamPartitions().stream().anyMatch(ssp -> ssp.getSystemStream().equals(input));
  }
Register an operator that this operator should propagate its results to.

@param nextOperator  the next operator to propagate results to
samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImpl.java