code: public <DO> RecordStreamWithMetadata<DO, S> mapRecords(Function<RecordEnvelope<D>, RecordEnvelope<DO>> transform) {
      return withRecordStream(this.recordStream.map(entity -> {
        if (entity instanceof RecordEnvelope) {
          return transform.apply((RecordEnvelope<D>) entity);
        } else {
          return (StreamEntity<DO>) entity;
        }
      }));
    }
docs: Apply the input mapping function to {@link RecordEnvelope}s, while letting other kinds of {@link StreamEntity}
  to pass through.