Introduction
Change data capture (CDC) connectors, such as DynamoDB and MongoDB source connectors, often put an entire database row into a single field of a Kafka message as a JSON string. If that row contains PII such as social security numbers or salaries, it flows to every sink connector downstream. That can put you out of compliance with GDPR, CCPA/CPRA, HIPAA or GLBA.
Kafka Connect's built-in MaskField transform can't help here. It masks whole fields of a record, but it can't see inside a JSON string stored in one of those fields.
mask-json-field is an open-source Single Message Transform (SMT) that fills this gap. It parses the embedded JSON, masks the fields you point to with a JSON Pointer path (including values inside arrays), and writes the result back before the record reaches the sink. It works on record keys and values, with Avro or plain string data.
mask-json-field Single Message Transform
| Field Name | Field Value |
| Id | 1 |
| Database | contoso |
| Operation | Insert |
| Document | {"name": "john doe", "salary": 100000, "ssn": "111-22-3333"} |
In this kafka message, the Document value is a String that is a JSON payload of the Database row.
Assume we are persisting this message into a different system using a Sink connector. However, due to privacy, you dont want to persist the salary and ssn.
{
"Id": 1,
"Database": "contoso",
"Operation": "Insert",
"Document": "{\"name\": \"john doe\", \"salary\": 100000, \"ssn\": \"111-22-3333\"}"
}
The mask-json-field SMT allows you to manipulate JSON documents embedded inside kafka messages.
Here is an example of using the transform in a FileStreamSink connector.
{
"name": "json-mask-test",
"connector.class": "FileStreamSink",
"file": "/tmp/file-sink.txt",
"key.converter": "org.apache.kafka.connect.storage.StringConverter",
"value.converter": "org.apache.kafka.connect.storage.StringConverter",
"tasks.max": 1,
"topics": "json-topic",
"transforms": "mask_ssn,mask_salary",
"transforms.mask_ssn.type": "io.github.ferozed.kafka.connect.transforms.MaskJsonField$Value",
"transforms.mask_ssn.CONNECT_FIELD_NAME": "Document",
"transforms.mask_ssn.REPLACEMENT_FIELD_PATH": "/ssn",
"transforms.mask_ssn.REPLACEMENT_VALUE_STRING": "xxx-xx-xxxx",
"transforms.mask_salary.type": "io.github.ferozed.kafka.connect.transforms.MaskJsonField$Value",
"transforms.mask_salary.CONNECT_FIELD_NAME": "Document",
"transforms.mask_salary.REPLACEMENT_FIELD_PATH": "/salary",
"transforms.mask_salary.REPLACEMENT_VALUE_INT": 0,
"errors.tolerance": "all"
}
With this transform, the output message will be:
| Field Name | Field Value |
| Id | 1 |
| Database | contoso |
| Operation | Insert |
| Document | {"name": "john doe", "salary": 0, "ssn": "xxx-xx-xxxx"} |
The raw version of the output will be as follows:
{
"Id": 1,
"Database": "contoso",
"Operation": "Insert",
"Document": "{\"name\": \"john doe\", \"salary\": 0, \"ssn\": \"xxx-xx-xxxx\"}"
}
This tranform also works on arrays. For eg, assume you have a SSN value as an array of strings
["111", "22", "3333" ]
and you want to transform it to:
["xxx", "xx", "3333" ]
Then the connector spec will look like this:
{
"name": "json-mask-test",
"connector.class": "FileStreamSink",
"file": "/tmp/file-sink.txt",
"key.converter": "org.apache.kafka.connect.storage.StringConverter",
"value.converter": "org.apache.kafka.connect.storage.StringConverter",
"tasks.max": 1,
"topics": "json-topic",
"transforms": "mask_salary,mask_ssn_1,mask_ssn_2",
"transforms.mask_salary.type": "io.github.ferozed.kafka.connect.transforms.MaskJsonField$Value",
"transforms.mask_salary.CONNECT_FIELD_NAME": "Document",
"transforms.mask_salary.REPLACEMENT_FIELD_PATH": "/salary",
"transforms.mask_salary.REPLACEMENT_VALUE_INT": 0,
"transforms.mask_ssn_1.type": "io.github.ferozed.kafka.connect.transforms.MaskJsonField$Value",
"transforms.mask_ssn_1.CONNECT_FIELD_NAME": "Document",
"transforms.mask_ssn_1.REPLACEMENT_FIELD_PATH": "/ssn/0",
"transforms.mask_ssn_1.REPLACEMENT_VALUE_STRING": "xxx",
"transforms.mask_ssn_2.type": "io.github.ferozed.kafka.connect.transforms.MaskJsonField$Value",
"transforms.mask_ssn_2.CONNECT_FIELD_NAME": "Document",
"transforms.mask_ssn_2.REPLACEMENT_FIELD_PATH": "/ssn/1",
"transforms.mask_ssn_2.REPLACEMENT_VALUE_STRING": "xx",
"errors.tolerance": "all"
}
There are many more feature that this transform supports. Look at the README.md of the source.
Limitations
This implementation has the following limitations.
- Each value that needs to be transformed needs a transform step. For eg, in the SSN field, there were three values that needed different rules, so they got their own transform steps.
- The implementation is limited to the following data types: STRING, INT, LONG, DOUBLE/FLOAT
- It only works on json documents encoded as strings.
Installation
Here are the steps to download and install the SMT jar in your connect cluster.- Download the latest version from https://central.sonatype.com/artifact/io.github.ferozed.kafka.connect/mask-json-field-transform
- Find the current value for config setting plugin.path from your kafka connect installation.
- Create a subdirectory under one of the existing plugin paths.
- Download the JAR and save it into the new path that was just created.
- Update the plugin path and add the path to the folder containing the new SMT jar.
Conclusion
Kafka Connect and its Single Message Transforms feature provide a robust and scalable framework for data integration within the Kafka ecosystem. The mask-json-field SMT provides organizations with an extensible mechanism to mask out sensitive fields from the data, allowing compliance with various data protection regulations.
