code: @Override
    public Iterable<GenericRecord> convertRecord(String outputSchema, GenericRecord inputRecord, WorkUnitState workUnit)
        throws DataConversionException {
      try {
        String schemaIdField = workUnit.contains(PAYLOAD_SCHEMA_ID_FIELD) ?
            workUnit.getProp(PAYLOAD_SCHEMA_ID_FIELD) : DEFAULT_PAYLOAD_SCHEMA_ID_FIELD;
        String payloadField = workUnit.contains(PAYLOAD_FIELD) ?
            workUnit.getProp(PAYLOAD_FIELD) : DEFAULT_PAYLOAD_FIELD;
        String schemaKey = String.valueOf(inputRecord.get(schemaIdField));
        Schema payloadSchema = (Schema) this.registry.getSchemaByKey(schemaKey);
        byte[] payload = getPayload(inputRecord, payloadField);
        GenericRecord outputRecord = deserializePayload(payload, payloadSchema);
        if (this.fieldRemover.isPresent()) {
          payloadSchema = this.fieldRemover.get().removeFields(payloadSchema);
        }
        return new SingleRecordIterable<>(AvroUtils.convertRecordSchema(outputRecord, payloadSchema));
      } catch (IOException | SchemaRegistryException | ExecutionException e) {
        throw new DataConversionException(e);
      }
    }
docs: Get actual schema from registry and deserialize payload using it.