ExtractNewRecordState implements Transformation<R> { private R addFields(List<FieldReference> additionalFields, R originalRecord, R unwrappedRecord) { final Struct value = requireStruct(unwrappedRecord.value(), PURPOSE); Struct originalRecordValue = (Struct) originalRecord.value(); Schema updatedSchema = schemaUpdateCache.computeIfAbsent(value.schema(), s -> makeUpdatedSchema(additionalFields, value.schema(), originalRecordValue)); Struct updatedValue = new Struct(updatedSchema); for (org.apache.kafka.connect.data.Field field : value.schema().fields()) { updatedValue.put(field.name(), value.get(field)); } for (FieldReference fieldReference : additionalFields) { updatedValue = updateValue(fieldReference, updatedValue, originalRecordValue); } return unwrappedRecord.newRecord( unwrappedRecord.topic(), unwrappedRecord.kafkaPartition(), unwrappedRecord.keySchema(), unwrappedRecord.key(), updatedSchema, updatedValue, unwrappedRecord.timestamp()); }  @Override void configure(final Map<String, ?> configs); @Override R apply(final R record); @Override ConfigDef config(); @Override void close();  }
@Test @FixFor({ "DBZ-1452", "DBZ-2504" }) public void testAddFields() { try (final ExtractNewRecordState<SourceRecord> transform = new ExtractNewRecordState<>()) { final Map<String, String> props = new HashMap<>(); props.put(ADD_FIELDS, "op , lsn,id"); props.put(ADD_FIELDS_PREFIX, "prefix."); transform.configure(props); final SourceRecord updateRecord = createUpdateRecord(); final SourceRecord unwrapped = transform.apply(updateRecord); assertThat(((Struct) unwrapped.value()).get("prefix.op")).isEqualTo(Envelope.Operation.UPDATE.code()); assertThat(((Struct) unwrapped.value()).get("prefix.lsn")).isEqualTo(1234); assertThat(((Struct) unwrapped.value()).get("prefix.id")).isEqualTo("571"); } }