code: @Override
    public RecordStreamWithMetadata<D, S> processStream(RecordStreamWithMetadata<D, S> inputStream, WorkUnitState state) {
      Flowable<StreamEntity<D>> filteredStream =
          inputStream.getRecordStream().filter(r -> {
            if (r instanceof ControlMessage) {
              getMessageHandler().handleMessage((ControlMessage) r);
              return true;
            } else if (r instanceof RecordEnvelope) {
              boolean accept = executePolicies(((RecordEnvelope) r).getRecord(), this.results);
              if (!accept) {
                r.ack();
              }
              return accept;
            } else {
              return true;
            }
          });
      filteredStream = filteredStream.doFinally(this::close);
      return inputStream.withRecordStream(filteredStream);
    }
docs: Process the stream and drop any records that fail the quality check.