Transform and converter chains in Kafka - Step-by-Step Execution

Process Flow - Transform and converter chains

Input Data

↓

Converter 1: Deserialize

↓

Transform 1: Modify Data

↓

Converter 2: Serialize

↓

Output Data

Data flows through a chain of converters and transforms, converting formats and modifying content step-by-step.

Execution Sample

Kafka

key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
transforms=InsertField
transforms.InsertField.type=org.apache.kafka.connect.transforms.InsertField$Value
transforms.InsertField.static.field=source
transforms.InsertField.static.value=app1

This config deserializes key as string, value as JSON without schema, then adds a static field 'source' with value 'app1' to the value.

Process Table

Step	Component	Input	Action	Output
1	Key Converter	Raw bytes	Deserialize bytes to string	"myKey"
2	Value Converter	Raw bytes	Deserialize bytes to JSON object	{"name":"Alice"}
3	Transform InsertField	{"name":"Alice"}	Add field 'source':'app1'	{"name":"Alice", "source":"app1"}
4	Value Converter	{"name":"Alice", "source":"app1"}	Serialize JSON object to bytes	Serialized bytes
5	Output	Serialized bytes	Send to Kafka topic	Message with key 'myKey' and value with added field

💡 All converters and transforms applied, message ready for output.

Status Tracker

Variable	Start	After Step 1	After Step 2	After Step 3	After Step 4	Final
key	Raw bytes	"myKey"	"myKey"	"myKey"	"myKey"	"myKey"
value	Raw bytes	Raw bytes	{"name":"Alice"}	{"name":"Alice", "source":"app1"}	Serialized bytes	Serialized bytes

Key Moments - 2 Insights

Why do we deserialize the value before applying the transform?

Because transforms work on structured data, not raw bytes. See execution_table step 2 and 3 where value is deserialized before adding the field.

Why is the value serialized again after the transform?

Kafka expects messages as bytes. After modifying the data, we serialize it back to bytes for sending. See execution_table step 4.

Visual Quiz - 3 Questions

Look at the execution table, what is the output of the Key Converter at step 1?

A Serialized bytes

B Raw bytes

C "myKey"

D {"name":"Alice"}

💡 Hint

Check the 'Output' column at step 1 in execution_table.

At which step is the field 'source' added to the value?

A Step 2

B Step 3

C Step 4

D Step 5

💡 Hint

Look at the 'Action' column describing adding the field in execution_table.

If we skip the final serialization (step 4), what would happen?

A Message would be sent as JSON object, causing errors

B Message would be sent as raw bytes

C Message key would be lost

D Transform would not apply

💡 Hint

Refer to the importance of serialization in step 4 and exit_note.

Concept Snapshot

Transform and converter chains process Kafka messages step-by-step.
Converters deserialize and serialize data formats.
Transforms modify the data between converters.
Order matters: deserialize -> transform -> serialize.
This ensures data is correctly formatted and enriched before sending.

Full Transcript

In Kafka, messages pass through converters and transforms in a chain. First, converters deserialize raw bytes into usable data formats like strings or JSON objects. Then, transforms modify this data, for example by adding fields. Finally, converters serialize the data back into bytes for Kafka to send. This flow ensures data is correctly handled and enriched. The example shows a key converted to string, a JSON value deserialized, a field added, then the value serialized again before output.

Practice

1. What is the main purpose of using transform and converter chains in Kafka Connect?

easy

A. To monitor Kafka cluster health

B. To store data permanently in Kafka topics

C. To create new Kafka topics automatically

D. To change data step-by-step during processing

Solution

1. Step 1: Understand transform and converter chains

   Transform and converter chains let you modify data in stages as it moves through Kafka Connect.

2. Step 2: Identify the main purpose

   The main goal is to change or clean data step-by-step automatically before it reaches its destination.

3. Final Answer:

   To change data step-by-step during processing -> Option D

4. Quick Check:

   Transform chains = step-by-step data change [OK]

Hint: Transforms modify data stepwise in Kafka Connect [OK]

Common Mistakes:

• Confusing transform chains with topic creation
• Thinking transforms monitor cluster health
• Assuming transforms store data permanently

2. Which of the following is the correct way to list multiple transforms in Kafka Connect configuration?

easy

A. transforms=InsertField,ReplaceField

B. transforms=[InsertField, ReplaceField]

C. transforms: InsertField; ReplaceField

D. transforms = InsertField ReplaceField

Solution

1. Step 1: Recall Kafka Connect transform list syntax

   Transforms are listed as a comma-separated string without brackets or semicolons.

2. Step 2: Match the correct syntax

   transforms=InsertField,ReplaceField uses commas correctly: transforms=InsertField,ReplaceField.

3. Final Answer:

   transforms=InsertField,ReplaceField -> Option A

4. Quick Check:

   Comma-separated list = correct syntax [OK]

Hint: Use commas to separate transforms in config [OK]

Common Mistakes:

• Using brackets or semicolons instead of commas
• Adding spaces around commas incorrectly
• Using colons or spaces without commas

3. Given this Kafka Connect transform chain configuration:
transforms=InsertField,ReplaceField
transforms.InsertField.type=org.apache.kafka.connect.transforms.InsertField$Value
transforms.InsertField.static.field=source
transforms.InsertField.static.value=app1
transforms.ReplaceField.type=org.apache.kafka.connect.transforms.ReplaceField$Value
transforms.ReplaceField.renames=oldName:newName
What will happen to a record with value {"oldName":"data"}?

medium

A. The record value becomes {"newName":"data"} only

B. The record value becomes {"oldName":"data", "source":"app1"}

C. The record value becomes {"source":"app1", "newName":"data"}

D. The record value remains unchanged

Solution

1. Step 1: Apply InsertField transform

   This adds a field "source" with value "app1" to the record, so value becomes {"oldName":"data", "source":"app1"}.

2. Step 2: Apply ReplaceField transform

   This renames "oldName" to "newName", so final value is {"source":"app1", "newName":"data"}.

3. Final Answer:

   {"source":"app1", "newName":"data"} -> Option C

4. Quick Check:

   InsertField adds, ReplaceField renames = final value [OK]

Hint: Apply transforms in order listed [OK]

Common Mistakes:

• Applying transforms in wrong order
• Forgetting InsertField adds a new field
• Assuming ReplaceField deletes old field without rename

4. You configured a transform chain but get an error: "Unknown transform type". Which of these fixes the problem?

medium

A. Remove all transforms from the config

B. Correct the transform type class name in the config

C. Restart Kafka Connect without changing config

D. Increase Kafka Connect heap size

Solution

1. Step 1: Understand the error cause

   "Unknown transform type" means Kafka Connect cannot find the class name specified for the transform.

2. Step 2: Fix the transform type

   Correcting the transform type class name in the configuration ensures Kafka Connect can load it properly.

3. Final Answer:

   Correct the transform type class name in the config -> Option B

4. Quick Check:

   Fix class name to resolve unknown transform error [OK]

Hint: Check transform class names carefully [OK]

Common Mistakes:

• Ignoring typo in transform type
• Restarting without fixing config
• Assuming memory issues cause this error

5. You want to create a transform chain that first removes sensitive fields, then adds a timestamp, and finally converts the record to JSON string. Which order of transforms is correct?

hard

A. RemoveFields, InsertField (timestamp), StringConverter

B. StringConverter, RemoveFields, InsertField (timestamp)

C. InsertField (timestamp), RemoveFields, StringConverter

D. RemoveFields, StringConverter, InsertField (timestamp)

Solution

1. Step 1: Remove sensitive fields first

   Removing fields must happen before adding new data or converting formats to avoid leaking sensitive info.

2. Step 2: Add timestamp after cleaning

   InsertField adds the timestamp field next, so the record has the new info before conversion.

3. Step 3: Convert record to JSON string last

   StringConverter should be last to convert the final cleaned and enriched record into a JSON string.

4. Final Answer:

   RemoveFields, InsertField (timestamp), StringConverter -> Option A

5. Quick Check:

   Clean, enrich, then convert = correct order [OK]

Hint: Clean data first, then add fields, convert last [OK]

Common Mistakes:

• Converting to string before removing fields
• Adding fields after conversion
• Wrong transform order causing data errors