A custom transformation is a user-implemented Kafka Connect record transformation for processing requirements that built-in transformations do not meet. A stock market API record may contain more than 70 metrics, while downstream consumers need only five or six selected fields. section 5.6.3
A practical implementation:
- Implement Kafka Connect’s Transformation interface and put field-selection logic in apply. section 5.6.3
- Configure the retained fields as a comma-separated list of names. section 5.6.3
- Handle schemaless maps and schema-backed Struct values separately. section 5.6.3
- Support applying the transformation to either the record key or value through Key and Value inner classes. section 5.6.3
Listing 5.12 Implementation of the custom transformation
public abstract class MultiFieldExtract<R extends ConnectRecord<R>> #1
implements Transformation<R> {
@Override
public R apply(R connectRecord) { #2
if (operatingValue(connectRecord) == null) { #3
return connectRecord;
} else if (operatingSchema(connectRecord) == null) { #4
return applySchemaless(connectRecord);
} else {
return applyWithSchema(connectRecord); #5
}
}
Use the implementation skeleton to route records through null-value, schemaless, or schema-aware processing. section 5.6.3
Listing 5.13 Transforming without a schema
private R applySchemaless(R connectRecord) {
final Map<String, Object> originalRecord =
requireMap(operatingValue(connectRecord), PURPOSE);
final Map<String, Object> newRecord = new LinkedHashMap<>();
List<Map.Entry<String,Object>> filteredEntryList =
originalRecord.entrySet().stream()
.filter(entry -> fieldNamesToExtract.contains(entry.getKey()))
.toList();
filteredEntryList.forEach(entry -> newRecord.put(entry.getKey(),
entry.getValue()));
return newRecord(connectRecord, null, newRecord);
}
Filter a schemaless record’s map entries, preserve selected values, and create a replacement record. section 5.6.3
Listing 5.14 Transforming with a schema
private R applyWithSchema(R connectRecord) {
final Struct value =
requireStruct(operatingValue(connectRecord), PURPOSE);
Schema updatedSchema = schemaUpdateCache.get(value.schema());
if(updatedSchema == null) {
updatedSchema = makeUpdatedSchema(value.schema());
schemaUpdateCache.put(value.schema(), updatedSchema);
}
final Struct updatedValue = new Struct(updatedSchema);
updatedValue.schema().fields()
.forEach(field -> updatedValue.put(field.name(),
value.get(field.name())));
return newRecord(connectRecord, updatedSchema, updatedValue);
}
Build or reuse an updated schema, copy selected Struct fields, and return the schema-backed replacement record. section 5.6.3
