1public class CustomTransform<R extends ConnectRecord<R>>
2    implements Transformation<R> {
3
4    @Override
5    public R apply(R record) {
6        // Return a transformed record here.
7        return record;
8    }
9
10    @Override
11    public ConfigDef config() {
12        return CONFIG_DEF;
13    }
14
15    @Override
16    public void close() {
17        // Cleanup
18    }
19
20    @Override
21    public void configure(Map<String, ?> configs) {
22        // Initialize
23    }
24}

1import java.util.Map;
2import org.apache.kafka.common.config.ConfigDef;
3import org.apache.kafka.connect.connector.ConnectRecord;
4import org.apache.kafka.connect.data.Field;
5import org.apache.kafka.connect.data.Struct;
6import org.apache.kafka.connect.transforms.Transformation;
7
8public class UppercaseValue<R extends ConnectRecord<R>>
9    implements Transformation<R> {
10
11    private static final String FIELD_CONFIG = "field";
12    private static final ConfigDef CONFIG_DEF = new ConfigDef()
13        .define(FIELD_CONFIG, ConfigDef.Type.STRING, ConfigDef.Importance.HIGH,
14            "Name of the string field to uppercase");
15
16    private String fieldName;
17
18    @Override
19    public R apply(R record) {
20        if (!(record.value() instanceof Struct value)) {
21            throw new IllegalArgumentException("UppercaseValue requires a Struct value");
22        }
23
24        Struct updated = new Struct(value.schema());
25        for (Field field : value.schema().fields()) {
26            updated.put(field, value.get(field));
27        }
28
29        String fieldValue = value.getString(fieldName);
30        if (fieldValue != null) {
31            updated.put(fieldName, fieldValue.toUpperCase());
32        }
33
34        return record.newRecord(
35            record.topic(),
36            record.kafkaPartition(),
37            record.keySchema(),
38            record.key(),
39            record.valueSchema(),
40            updated,
41            record.timestamp(),
42            record.headers()
43        );
44    }
45
46    @Override
47    public ConfigDef config() {
48        return CONFIG_DEF;
49    }
50
51    @Override
52    public void configure(Map<String, ?> configs) {
53        fieldName = configs.get(FIELD_CONFIG).toString();
54    }
55
56    @Override
57    public void close() {
58        // No resources to close.
59    }
60}

1{
2  "transforms": "uppercase",
3  "transforms.uppercase.type": "com.example.UppercaseValue",
4  "transforms.uppercase.field": "name"
5}