@Override
  public <K, R extends KV, JM> MessageStream<JM> join(Table<R> table,
      StreamTableJoinFunction<? extends K, ? super M, ? super R, ? extends JM> joinFn, Object ... args) {
    String opId = this.streamAppDesc.getNextOpId(OpCode.JOIN);
    StreamTableJoinOperatorSpec<K, M, R, JM> joinOpSpec = OperatorSpecs.createStreamTableJoinOperatorSpec(
        ((TableImpl) table).getTableId(), (StreamTableJoinFunction<K, M, R, JM>) joinFn, opId, args);
    this.operatorSpec.registerNextOperatorSpec(joinOpSpec);
    return new MessageStreamImpl<>(this.streamAppDesc, joinOpSpec);
  }
The {@link OperatorSpec} associated with this {@link MessageStreamImpl}
samza-core/src/main/java/org/apache/samza/operators/MessageStreamImpl.java