code: @Override
    public RecordStreamWithMetadata<DO, SO> processStream(RecordStreamWithMetadata<DI, SI> inputStream,
        WorkUnitState workUnitState) throws SchemaConversionException {
      int maxConcurrentAsyncConversions = workUnitState.getPropAsInt(MAX_CONCURRENT_ASYNC_CONVERSIONS_KEY,
          DEFAULT_MAX_CONCURRENT_ASYNC_CONVERSIONS);
      SO outputSchema = convertSchema(inputStream.getGlobalMetadata().getSchema(), workUnitState);
      Flowable<StreamEntity<DO>> outputStream =
          inputStream.getRecordStream()
              .flatMapSingle(in -> {
                if (in instanceof ControlMessage) {
                  getMessageHandler().handleMessage((ControlMessage) in);
                  return Single.just((ControlMessage<DO>) in);
                } else if (in instanceof RecordEnvelope) {
                  RecordEnvelope<DI> recordEnvelope = (RecordEnvelope<DI>) in;
                  return new SingleAsync(recordEnvelope, convertRecordAsync(outputSchema, recordEnvelope.getRecord(), workUnitState));
                } else {
                  throw new IllegalStateException("Expected ControlMessage or RecordEnvelope.");
                }
              }, false, maxConcurrentAsyncConversions);
      return inputStream.withRecordStream(outputStream, GlobalMetadata.<SI, SO>builderWithInput(inputStream.getGlobalMetadata(),
          Optional.fromNullable(outputSchema)).build());
    }
docs: Return a {@link RecordStreamWithMetadata} with the appropriate modifications.
  @param inputStream
  @param workUnitState
  @return
  @throws SchemaConversionException
  @implNote this processStream does not handle {@link org.apache.gobblin.stream.MetadataUpdateControlMessage}s