Ecosystem

What it is and why it exists

Brokers do not look inside records. To them a key and a value are two arrays of bytes. This keeps Kafka fast and general, and it leaves three problems to you.

• Meaning. A producer team changes a field and every consumer breaks. A schema registry stores the agreed shape of each topic's records and refuses changes that would break readers.
• Integration. Most data starts in a database and ends in a database, a search index or a warehouse. Kafka Connect is a framework, part of Apache Kafka, that runs ready-made connectors for those systems, so that you configure a pipeline and do not program it.
• Consistency and failure. A service must change its database and publish an event, and both must happen or neither. And some records will always fail in processing. The outbox pattern, change data capture and dead-letter topics are the standard answers.

Failure scenarios

Hands-on lab

k() { docker exec -i kafka-1 /opt/kafka/bin/"$@"; }
B=kafka-1:19092

Part A, step 1. Start a Connect worker

docker exec kafka-1 sh -c 'cat > /tmp/connect.properties <<EOF
bootstrap.servers=kafka-1:19092
group.id=connect-lab
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter
offset.storage.topic=connect-offsets
config.storage.topic=connect-configs
status.storage.topic=connect-status
plugin.path=/opt/kafka/libs/connect-file-4.3.1.jar
listeners=http://0.0.0.0:8083
EOF'
docker exec -d kafka-1 sh -c '/opt/kafka/bin/connect-distributed.sh /tmp/connect.properties > /tmp/connect.log 2>&1'

# the image has wget but not curl; define two helpers for the HTTP interface
get()  { docker exec kafka-1 wget -q -O - "http://localhost:8083$1"; }
post() { docker exec kafka-1 wget -q -O - --header 'Content-Type: application/json' --post-data "$2" "http://localhost:8083$1"; }

# wait about 20 seconds
get /
get /connector-plugins
k kafka-topics.sh --bootstrap-server $B --list | grep connect

{"version":"4.3.1","commit":"26b251a451ce941d","kafka_cluster_id":"4L6g3nShT-eMCtK--X86sw"}

FileStreamSinkConnector (sink), FileStreamSourceConnector (source),
MirrorCheckpointConnector, MirrorHeartbeatConnector, MirrorSourceConnector

connect-configs
connect-offsets
connect-status

Part A, step 2. File to topic to file

docker exec kafka-1 sh -c 'printf "line one\nline two\n" > /tmp/in.txt'

post /connectors '{"name":"file-source","config":{"connector.class":"org.apache.kafka.connect.file.FileStreamSourceConnector","tasks.max":"1","file":"/tmp/in.txt","topic":"connect-lines"}}'
post /connectors '{"name":"file-sink","config":{"connector.class":"org.apache.kafka.connect.file.FileStreamSinkConnector","tasks.max":"1","file":"/tmp/out.txt","topics":"connect-lines"}}'

# wait 8 seconds, then add a line while the connectors run
docker exec kafka-1 sh -c 'echo "line three" >> /tmp/in.txt'
# wait 6 seconds
docker exec kafka-1 cat /tmp/out.txt
get /connectors
get /connectors/file-source/status
k kafka-consumer-groups.sh --bootstrap-server $B --describe --group connect-file-sink

line one
line two
line three

["file-source","file-sink"]
{"name":"file-source","connector":{"state":"RUNNING",…},"tasks":[{"id":0,"state":"RUNNING",…}],"type":"source"}

GROUP              TOPIC          PARTITION  LOG-END-OFFSET
connect-file-sink  connect-lines  0          3

No code was written. The third line appeared in the output file a few seconds after it was added to the input file. The sink connector is a consumer group named connect- plus its name, so the lag tools from module 03 work on it.

Part A, step 3. One bad record, no tolerance

Put three records in a topic. The middle one is not JSON. Start a sink that expects JSON.

k kafka-topics.sh --bootstrap-server $B --create --topic json-in --partitions 1 --replication-factor 3
printf '{"id":1}\nthis is not json\n{"id":2}\n' | k kafka-console-producer.sh --bootstrap-server $B --topic json-in

post /connectors '{"name":"json-strict","config":{"connector.class":"org.apache.kafka.connect.file.FileStreamSinkConnector","tasks.max":"1","file":"/tmp/strict.txt","topics":"json-in","value.converter":"org.apache.kafka.connect.json.JsonConverter","value.converter.schemas.enable":"false"}}'
# wait 10 seconds
get /connectors/json-strict/status
docker exec kafka-1 cat /tmp/strict.txt

{"name":"json-strict","connector":{"state":"RUNNING",…},
 "tasks":[{"id":0,"state":"FAILED",…,"trace":"org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler …

(strict.txt is empty)

The connector says RUNNING and its task says FAILED. The output file is empty: the two good records were not delivered either.

Part A, step 4. The same, with a dead-letter topic

post /connectors '{"name":"json-dlq","config":{"connector.class":"org.apache.kafka.connect.file.FileStreamSinkConnector","tasks.max":"1","file":"/tmp/dlq-ok.txt","topics":"json-in","value.converter":"org.apache.kafka.connect.json.JsonConverter","value.converter.schemas.enable":"false","errors.tolerance":"all","errors.deadletterqueue.topic.name":"json-in-dlq","errors.deadletterqueue.topic.replication.factor":"3","errors.deadletterqueue.context.headers.enable":"true"}}'
# wait 12 seconds
get /connectors/json-dlq/status
docker exec kafka-1 cat /tmp/dlq-ok.txt
k kafka-console-consumer.sh --bootstrap-server $B --topic json-in-dlq --from-beginning \
  --formatter-property print.headers=true --timeout-ms 6000

{"name":"json-dlq","connector":{"state":"RUNNING",…},"tasks":[{"id":0,"state":"RUNNING",…}],"type":"sink"}

{id=1}
{id=2}

__connect.errors.topic:json-in,__connect.errors.partition:0,__connect.errors.offset:1,
__connect.errors.connector.name:json-dlq,__connect.errors.task.id:0,
__connect.errors.stage:VALUE_CONVERTER,
__connect.errors.class.name:org.apache.kafka.connect.json.JsonConverter,
__connect.errors.exception.class.name:org.apache.kafka.connect.errors.DataException,
__connect.errors.exception.message:Converting byte[] to Kafka Connect data failed due to serialization error: ,
__connect.errors.exception.stacktrace:… Unrecognized token 'this': was expecting (JSON String, Number, …)
    this is not json

The task keeps running and both good records are delivered. The bad record is in the dead-letter topic with its original value, and the headers say exactly where it came from (topic json-in, partition 0, offset 1), which stage failed, and why.

Part B, step 1. Start the schema registry

docker compose up -d schema-registry
R=http://localhost:8086
H='Content-Type: application/vnd.schemaregistry.v1+json'
curl -s $R/config

{"compatibilityLevel":"BACKWARD"}

Part B, step 2. Register a schema and evolve it

Version 1 has id and amount. Version 2 adds currency with a default. Version 3 tries to add customerId without a default.

V1='{"schema":"{\"type\":\"record\",\"name\":\"Order\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"}]}"}'
V2='{"schema":"{\"type\":\"record\",\"name\":\"Order\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"},{\"name\":\"currency\",\"type\":\"string\",\"default\":\"CAD\"}]}"}'
V3='{"schema":"{\"type\":\"record\",\"name\":\"Order\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"},{\"name\":\"currency\",\"type\":\"string\",\"default\":\"CAD\"},{\"name\":\"customerId\",\"type\":\"string\"}]}"}'

curl -s -X POST -H "$H" -d "$V1" $R/subjects/orders-value/versions
curl -s -X POST -H "$H" -d "$V2" $R/subjects/orders-value/versions
curl -s -X POST -H "$H" -d "$V3" $R/subjects/orders-value/versions
curl -s $R/subjects/orders-value/versions

{"id":1,"version":1,…,"schemaType":"AVRO",…}
{"id":2,"version":2,…}
{"error_code":40901,"message":"Schema being registered is incompatible with an earlier schema for subject
 \"orders-value\", details: [{errorType:'READER_FIELD_MISSING_DEFAULT_VALUE', description:'The field
 'customerId' at path '/fields/3' in the new schema has no default value and is missing in the old schema' …
[1,2]

The registry refused version 3. A consumer with the new schema could not read an old record, because the old record has no customerId and the schema gives no default to use.

Part B, step 3. Test a change without registering it

Version 4 removes amount. Test it under the default mode, then switch the subject to FULL and test again.

V4='{"schema":"{\"type\":\"record\",\"name\":\"Order\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"currency\",\"type\":\"string\",\"default\":\"CAD\"}]}"}'
curl -s -X POST -H "$H" -d "$V4" "$R/compatibility/subjects/orders-value/versions/latest?verbose=true"

curl -s -X PUT -H "$H" -d '{"compatibility":"FULL"}' $R/config/orders-value
curl -s -X POST -H "$H" -d "$V4" "$R/compatibility/subjects/orders-value/versions/latest?verbose=true"

{"is_compatible":true,"messages":[]}

{"compatibility":"FULL"}
{"is_compatible":false,"messages":["{errorType:'READER_FIELD_MISSING_DEFAULT_VALUE', description:'The field
 'amount' at path '/fields/1' in the old schema has no default value and is missing in the new schema' …

Removing a field is allowed under BACKWARD: new consumers simply ignore it in old data. Under FULL it is refused, because old consumers still expect amount and would find nothing. This test call is what a build pipeline should run before a release.

Part B, step 4. See where the registry keeps its data

k kafka-topics.sh --bootstrap-server $B --describe --topic _schemas | head -1

Topic: _schemas  PartitionCount: 1  ReplicationFactor: 3  Configs: min.insync.replicas=2,cleanup.policy=compact

Common mistakes and misconceptions

Staff-level questions

Self-test