code: @Override
    public RecordStreamWithMetadata<DI, SI> processStream(RecordStreamWithMetadata<DI, SI> inputStream,
        WorkUnitState workUnitState) throws StreamProcessingException {
      init(workUnitState);
  
      setInputGlobalMetadata(inputStream.getGlobalMetadata(), workUnitState);
  
      Flowable<StreamEntity<DI>> outputStream =
          inputStream.getRecordStream()
              .flatMap(in -> {
                if (in instanceof ControlMessage) {
                  if (in instanceof MetadataUpdateControlMessage) {
                    setInputGlobalMetadata(((MetadataUpdateControlMessage) in).getGlobalMetadata(), workUnitState);
                  }
  
                  getMessageHandler().handleMessage((ControlMessage) in);
                  return Flowable.just(in);
                } else if (in instanceof RecordEnvelope) {
                  RecordEnvelope<DI> recordEnvelope = (RecordEnvelope<DI>) in;
                  Iterable<ControlMessage<DI>> injectedBeforeIterable =
                      injectControlMessagesBefore(recordEnvelope, workUnitState);
                  Iterable<ControlMessage<DI>> injectedAfterIterable =
                      injectControlMessagesAfter(recordEnvelope, workUnitState);
  
                  if (injectedBeforeIterable == null && injectedAfterIterable == null) {
                    // nothing injected so return the record envelope
                    return Flowable.just(recordEnvelope);
                  } else {
                    Flowable<StreamEntity<DI>> flowable;
  
                    if (injectedBeforeIterable != null) {
                      flowable = Flowable.<StreamEntity<DI>>fromIterable(injectedBeforeIterable)
                          .concatWith(Flowable.just(recordEnvelope));
                    } else {
                      flowable = Flowable.just(recordEnvelope);
                    }
  
                    if (injectedAfterIterable != null) {
                      flowable.concatWith(Flowable.fromIterable(injectedAfterIterable));
                    }
                    return flowable;
                  }
                } else {
                  throw new UnsupportedOperationException();
                }
              }, 1);
      outputStream = outputStream.doOnComplete(this::close);
      return inputStream.withRecordStream(outputStream, inputStream.getGlobalMetadata());
    }
docs: Apply injections to the input {@link RecordStreamWithMetadata}.
  {@link ControlMessage}s may be injected before, after, or around the input record.
  A {@link MetadataUpdateControlMessage} will update the current input {@link GlobalMetadata} and pass the
  updated input {@link GlobalMetadata} to the next processor to propagate the metadata update down the pipeline.