@Override
  public <K, WV> MessageStream<WindowPane<K, WV>> window(Window<M, K, WV> window, String userDefinedId) {
    String opId = this.streamAppDesc.getNextOpId(OpCode.WINDOW, userDefinedId);
    OperatorSpec<M, WindowPane<K, WV>> op = OperatorSpecs.createWindowOperatorSpec((WindowInternal<M, K, WV>) window, opId);
    this.operatorSpec.registerNextOperatorSpec(op);
    return new MessageStreamImpl<>(this.streamAppDesc, op);
  }
The {@link OperatorSpec} associated with this {@link MessageStreamImpl}
samza-core/src/main/java/org/apache/samza/operators/MessageStreamImpl.java