@Override
  public void sink(SinkFunction<? super M> sinkFn) {
    String opId = this.streamAppDesc.getNextOpId(OpCode.SINK);
    SinkOperatorSpec<M> op = OperatorSpecs.createSinkOperatorSpec(sinkFn, opId);
    this.operatorSpec.registerNextOperatorSpec(op);
  }
The {@link OperatorSpec} associated with this {@link MessageStreamImpl}
samza-core/src/main/java/org/apache/samza/operators/MessageStreamImpl.java