public void close() {
    List<OperatorImpl> initializationOrder = new ArrayList<>(operatorImpls.values());
    List<OperatorImpl> finalizationOrder = Lists.reverse(initializationOrder);
    finalizationOrder.forEach(OperatorImpl::close);
  }
Get the {@link InputOperatorImpl} corresponding to the provided input {@code systemStream}.

@param systemStream  input {@link SystemStream}
@return  the {@link InputOperatorImpl} that starts processing the input message
samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImplGraph.java