@Override
  public MessageStream<M> broadcast(Serde<M> serde, String userDefinedId) {
    String opId = this.streamAppDesc.getNextOpId(OpCode.BROADCAST, userDefinedId);
    IntermediateMessageStreamImpl<M> intermediateStream = this.streamAppDesc.getIntermediateStream(opId, serde, true);
    BroadcastOperatorSpec<M> broadcastOperatorSpec =
        OperatorSpecs.createBroadCastOperatorSpec(intermediateStream.getOutputStream(), opId);
    this.operatorSpec.registerNextOperatorSpec(broadcastOperatorSpec);
    return intermediateStream;
  }
The {@link OperatorSpec} associated with this {@link MessageStreamImpl}
samza-core/src/main/java/org/apache/samza/operators/MessageStreamImpl.java