@VisibleForTesting
  public <M> IntermediateMessageStreamImpl<M> getIntermediateStream(String streamId, Serde<M> serde, boolean isBroadcast) {
    Preconditions.checkNotNull(serde, "serde must not be null for intermediate stream: " + streamId);
    Preconditions.checkState(!inputOperators.containsKey(streamId) && !outputStreams.containsKey(streamId),
        "getIntermediateStream must not be called multiple times with the same streamId: " + streamId);

    if (isBroadcast) {
      intermediateBroadcastStreamIds.add(streamId);
    }

    boolean isKeyed = serde instanceof KVSerde;
    KV<Serde, Serde> kvSerdes = getOrCreateStreamSerdes(streamId, serde);

    InputTransformer transformer = (InputTransformer) getDefaultSystemDescriptor()
        .flatMap(SystemDescriptor::getTransformer).orElse(null);

    InputOperatorSpec inputOperatorSpec =
        OperatorSpecs.createInputOperatorSpec(streamId, kvSerdes.getKey(), kvSerdes.getValue(),
            transformer, isKeyed, this.getNextOpId(OpCode.INPUT, null));
    inputOperators.put(streamId, inputOperatorSpec);
    outputStreams.put(streamId, new OutputStreamImpl(streamId, kvSerdes.getKey(), kvSerdes.getValue(), isKeyed));
    return new IntermediateMessageStreamImpl<>(this, inputOperators.get(streamId), outputStreams.get(streamId));
  }
Internal helper for {@link MessageStreamImpl} to add an intermediate {@link MessageStream} to the graph.
An intermediate {@link MessageStream} is both an output and an input stream.

@param streamId the id of the stream to be created.
@param serde the {@link Serde} to use for the message in the intermediate stream. If null, the default serde
             is used.
@param isBroadcast whether the stream is a broadcast stream.
@param <M> the type of messages in the intermediate {@link MessageStream}
@return  the intermediate {@link MessageStreamImpl}
samza-core/src/main/java/org/apache/samza/application/descriptors/StreamApplicationDescriptorImpl.java