code: protected Flowable<RecordEnvelope<DO>> convertRecordEnvelope(SO outputSchema, RecordEnvelope<DI> inputRecordEnvelope,
        WorkUnitState workUnitState) throws DataConversionException {
      Iterator<DO> convertedIterable = convertRecord(outputSchema,
          inputRecordEnvelope.getRecord(), workUnitState).iterator();
  
      if (!convertedIterable.hasNext()) {
        inputRecordEnvelope.ack();
        // if the iterable is empty, ack the record, return an empty flowable
        return Flowable.empty();
      }
  
      DO firstRecord = convertedIterable.next();
      if (!convertedIterable.hasNext()) {
        // if the iterable has only one element, use RecordEnvelope.withRecord, which is more efficient
        return Flowable.just(inputRecordEnvelope.withRecord(firstRecord));
      } else {
        // if the iterable has multiple records, use a ForkRecordBuilder
        RecordEnvelope<DI>.ForkRecordBuilder<DO> forkRecordBuilder = inputRecordEnvelope.forkRecordBuilder();
        return Flowable.just(firstRecord).concatWith(Flowable.fromIterable(() -> convertedIterable))
            .map(forkRecordBuilder::childRecord).doOnComplete(forkRecordBuilder::close);
      }
    }
docs: Converts a {@link RecordEnvelope}. This method can be overridden by implementations that need to manipulate the
  {@link RecordEnvelope}, such as to set watermarks or metadata.
  @param outputSchema output schema converted using the {@link Converter#convertSchema} method
  @param inputRecordEnvelope input record envelope with data record to be converted
  @param workUnitState a {@link WorkUnitState} object carrying configuration properties
  @return a {@link Flowable} emitting the converted {@link RecordEnvelope}s
  @throws DataConversionException