Transform and converter chains in Kafka - Commands & Configuration

Introduction

When data moves through Kafka Connect, it often needs changes to fit the target system. Transform and converter chains let you change data format and content step-by-step, like a factory line, so the data is ready to use.

When you want to change the format of messages from JSON to Avro before sending to a database.

When you need to add or remove fields from data as it moves through Kafka Connect.

When you want to convert data encoding from string to bytes for a specific sink system.

When you want to apply multiple small changes to data in order, like trimming spaces then changing case.

When you want to ensure data is compatible with the target system by applying converters and transforms in sequence.

Config File - connector-with-transform.yaml

connector-with-transform.yaml

name: example-connector
connector.class: FileStreamSink
tasks.max: 1
topic: example-topic
file: /tmp/output.txt
key.converter: org.apache.kafka.connect.storage.StringConverter
value.converter: org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable: false
transforms: trim,upperCase
transforms.trim.type: org.apache.kafka.connect.transforms.ReplaceField$Value
transforms.trim.whitelist: message
transforms.upperCase.type: org.apache.kafka.connect.transforms.ValueToKey
transforms.upperCase.fields: message

This YAML configures a Kafka Connect sink connector that writes data to a file.

key.converter and value.converter set how keys and values are converted between Kafka and Connect.

transforms defines a chain of transformations applied to the data before writing.

Each transform is named and configured to modify data step-by-step.

Commands

Start a producer to send messages to the example-topic in Kafka.

Terminal

kafka-console-producer --broker-list localhost:9092 --topic example-topic

Expected Output

No output (command runs silently)

→

--broker-list - Specifies Kafka broker address

→

--topic - Specifies the topic to send messages to

Start Kafka Connect in standalone mode with the connector config that applies transform and converter chains.

Terminal

kafka-connect-standalone connect-standalone.properties connector-with-transform.yaml

Expected Output

[2024-06-01 12:00:00,000] INFO Kafka Connect started (org.apache.kafka.connect.cli.ConnectStandalone)

[2024-06-01 12:00:01,000] INFO Connector example-connector started (org.apache.kafka.connect.runtime.WorkerConnector)

Check the output file to see the transformed data after passing through converter and transform chains.

Terminal

cat /tmp/output.txt

Expected Output

{"message":"HELLO WORLD"}

Key Concept

If you remember nothing else from this pattern, remember: transform and converter chains let you change data format and content step-by-step in Kafka Connect to fit your target system.

Common Mistakes

Not enabling schemas when using JsonConverter causes data format errors.

Without schemas enabled, the converter may not correctly interpret data structure.

Set value.converter.schemas.enable to true when using JsonConverter if your data has schemas.

Applying transforms in the wrong order leads to unexpected data results.

Transforms run in sequence; wrong order can break data or cause errors.

Plan and order transforms carefully to match the desired data changes.

Using incompatible converters for key and value causes connector failures.

Key and value converters must match the data format expected by source and sink.

Choose converters that match your data format and target system requirements.

Summary

Configure key and value converters to handle data format conversion.

Define a chain of transforms to modify data step-by-step before sending to the sink.

Run Kafka Connect with the connector config and verify transformed data 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