In this post we are going to cover Kafka Connect, change data capture (CDC) and schemas. These are 7 real Kafka interview questions for data engineers, asked the way they come up on a real project, and the last one is a scenario: moving orders from an operational database to the data lake in near real time, with schema changes.
Kafka Connect is the part of Kafka that moves data in and out of topics with configuration instead of code. CDC turns database changes into events, and schemas keep producers and consumers agreeing on the format. Interviewers ask about these because almost every data engineering pipeline on Kafka starts with a connector and breaks on a schema change sooner or later.
Part 7 of the Real Time Kafka Interview Questions and Answers series on TechBrothersIT.
- The short answer
- Let's set up the cluster and some sample data
- Question 1: The team writes custom code to copy every table into Kafka. When should you use Kafka Connect instead?
- Question 2: Your manager asks how to stream database changes without polling the tables. What is change data capture?
- Question 3: A sink connector stops on one bad record. How do error tolerance and dead letter queues work in Connect?
- Question 4: A producer changed a field type and the consumers broke. Why do you need schemas and a schema registry?
- Question 5: A teammate asks which compatibility mode to use for schema evolution. How do backward and forward compatibility work?
- Question 6: Connect workers keep losing tasks. How do standalone and distributed mode differ?
- Scenario 7: You must move orders from an operational database to the data lake in near real time with schema changes. What is your design?
- The full script
The short answer
Use Kafka Connect to move data in and out of Kafka with configuration, and log-based CDC (Debezium) instead of polling. Protect sink connectors with errors.tolerance=all and a dead letter queue, put a schema registry with BACKWARD compatibility in front of shared topics, and run Connect in distributed mode in production.
Let's set up the cluster and some sample data
We use a local Kafka 4.4 cluster in KRaft mode with 1 broker. Here are the starting files: a small orders.txt, the Connect worker and connector configs, and three helper scripts. status.py waits for a connector task to reach a state, read.py reads a topic from the start, and stop.sh stops the Connect worker.
1001,c1,25.00
1002,c2,40.50
1003,c1,9.99bootstrap.servers=localhost:9092
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=true
offset.storage.file.filename=connect.offsets
# lab only: flush source offsets every second (default 60 s)
offset.flush.interval.ms=1000
plugin.path=plugins
listeners=http://localhost:8307name=orders-file-source
connector.class=FileStreamSource
tasks.max=1
file=orders.txt
topic=orders-rawimport json, sys, time, urllib.request
name, want = sys.argv[1], sys.argv[2]
port = sys.argv[3] if len(sys.argv) > 3 else '8307'
url = 'http://localhost:%s/connectors/%s/status' % (port, name)
for _ in range(90):
try:
s = json.load(urllib.request.urlopen(url))
if s['tasks'] and s['tasks'][0]['state'] == want:
break
except OSError:
pass
time.sleep(1)
print('connector', s['name'], s['connector']['state'])
for t in s['tasks']:
print('task', t['id'], t['state'], t.get('trace', '').split('\n')[0])import sys
from confluent_kafka import Consumer
topic, want = sys.argv[1], int(sys.argv[2])
c = Consumer({'bootstrap.servers': 'localhost:9092', 'group.id': 'reader',
'auto.offset.reset': 'earliest', 'enable.auto.commit': False})
c.subscribe([topic])
rows, empty = [], 0
while len(rows) < want and empty < 30:
m = c.poll(1.0)
if m is None or m.error():
empty += 1
continue
rows.append('offset %d: %s' % (m.offset(), m.value().decode()))
c.close()
print('\n'.join(rows))kill $(cat connect.pid)
while kill -0 $(cat connect.pid) 2>/dev/null; do sleep 0.5; doneQuestion 1: The team writes custom code to copy every table into Kafka. When should you use Kafka Connect instead?
Use Connect whenever the job is moving data into or out of Kafka, and a connector already exists. Source connectors write into Kafka, sink connectors read out of it, and we write configuration, not code.
The flow is simple: a database or files are the source, a source connector writes into a Kafka topic like orders-raw, and a sink connector reads out of Kafka into the lake or warehouse.
Let's try a source connector with no code. worker.properties sets the brokers, the converters and the plugin path. file-source.properties says: read orders.txt, write to orders-raw. We start Connect in standalone mode, wait for the task, and read the topic.
$ connect-standalone.sh worker.properties file-source.properties > connect.log 2>&1 &
$ python status.py orders-file-source RUNNING
connector orders-file-source RUNNING
task 0 RUNNING
$ python read.py orders-raw 3
offset 0: {"schema":{"type":"string","optional":false},"payload":"1001,c1,25.00"}
offset 1: {"schema":{"type":"string","optional":false},"payload":"1002,c2,40.50"}
offset 2: {"schema":{"type":"string","optional":false},"payload":"1003,c1,9.99"}The task is RUNNING, and three lines became three records. The JSON converter wrapped each line in a schema and a payload. Remember that for Question 4.
Now let's add a line to the file and read again.
$ echo "1004,c3,12.00" >> orders.txt
$ python read.py orders-raw 4
offset 0: {"schema":{"type":"string","optional":false},"payload":"1001,c1,25.00"}
offset 1: {"schema":{"type":"string","optional":false},"payload":"1002,c2,40.50"}
offset 2: {"schema":{"type":"string","optional":false},"payload":"1003,c1,9.99"}
offset 3: {"schema":{"type":"string","optional":false},"payload":"1004,c3,12.00"}
$ curl -s localhost:8307/connectors/orders-file-source/offsets | jq -c .
{"offsets":[{"partition":{"filename":"orders.txt"},"offset":{"position":55}}]}Only the new line is sent, at offset 3, because Connect stores the source offset for us, here the byte position in the file: "position":55. Offsets, retries, converters, tasks and a REST API are built in.
Custom code still makes sense when no connector fits, or when it's real business logic. Then write a producer or a stream processing app, not another copy job.
Question 2: Your manager asks how to stream database changes without polling the tables. What is change data capture?
Change data capture reads the database's own transaction log, the write-ahead log (WAL) in Postgres or the binlog in MySQL, and turns every committed insert, update and delete into an event.
First, let's see why polling an updated_at column is not enough.
| Polling a timestamp column | Log-based CDC | |
|---|---|---|
| Deletes | missed: the row is gone | a delete event + tombstone |
| Two updates between polls | only the last one is seen | both, in commit order |
| Load on the database | a query every poll | reads the log |
Polling misses deletes, misses a row that changed twice between polls, and keeps querying the database. Log-based CDC sees every change, in commit order.
The usual tool is Debezium, a set of Kafka Connect source connectors. This Postgres config names the database, a topic prefix, the tables, and the pgoutput plugin.
name=orders-cdc
connector.class=io.debezium.connector.postgresql.PostgresConnector
database.hostname=orders-db
database.port=5432
database.user=cdc_user
database.password=${file:/secrets/cdc.properties:password}
database.dbname=shop
topic.prefix=shop
table.include.list=public.orders
plugin.name=pgoutputIt snapshots the existing rows, then streams changes, one topic per table, like shop.public.orders. Let's look at one change event.
{
"before": {"order_id": 1001, "status": "placed"},
"after": {"order_id": 1001, "status": "shipped"},
"op": "u",
"source": {"connector": "postgresql",
"db": "shop", "table": "orders",
"lsn": 24023128},
"ts_ms": 1793091600123
}Each event has the row after the change, an op code (c, u, d, or r for snapshot reads), and the log position. The before image needs replica identity full on Postgres. And a delete is followed by a tombstone, for compaction.
If the connector stops, Postgres keeps the WAL for it and the disk fills up. Alert on slot lag, and turn on Debezium heartbeats when the captured tables are quiet but the database is busy.
Question 3: A sink connector stops on one bad record. How do error tolerance and dead letter queues work in Connect?
By default, errors.tolerance is none: the first record that fails in the converter or a transform fails the task.
Let's create three payments where the second one is not JSON. The file sink reads them with the JSON converter.
from confluent_kafka import Producer
rows = ['{"id": "p1", "amount": 25.0}',
'id=p2;amount=12.50', # not JSON: a bad record
'{"id": "p3", "amount": 40.0}']
p = Producer({'bootstrap.servers': 'localhost:9092'})
for v in rows:
p.produce('payments', value=v)
p.flush()
print('sent', len(rows), 'payments')name=payments-sink
connector.class=FileStreamSink
tasks.max=1
topics=payments
file=payments.out
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false$ python produce_payments.py
sent 3 paymentsNow we start the sink with the default settings.
$ connect-standalone.sh worker.properties payments-sink.properties > sink.log 2>&1 &
$ python status.py payments-sink FAILED
connector payments-sink RUNNING
task 0 FAILED org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler
$ wc -l payments.out
0 payments.outThe connector is RUNNING, but task 0 is FAILED with "Tolerance exceeded in error handler". The output file has 0 lines. One bad record stopped the whole pipeline.
Next, let's set errors.tolerance=all, name a dead letter queue (DLQ) topic, and turn on context headers. Then we restart the sink.
name=payments-sink
connector.class=FileStreamSink
tasks.max=1
topics=payments
file=payments.out
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
errors.tolerance=all
errors.deadletterqueue.topic.name=payments-dlq
# 1 only because this lab has one broker (default 3)
errors.deadletterqueue.topic.replication.factor=1
errors.deadletterqueue.context.headers.enable=truefrom confluent_kafka import Consumer
c = Consumer({'bootstrap.servers': 'localhost:9092', 'group.id': 'dlq-check',
'auto.offset.reset': 'earliest'})
c.subscribe(['payments-dlq'])
m = None
while m is None or m.error():
m = c.poll(1.0)
print('value:', m.value().decode())
for k, v in m.headers():
if k.endswith(('.topic', '.offset', '.stage', 'exception.class.name', 'exception.message')):
print(k, '=', v.decode().split(':')[0])
c.close()$ connect-standalone.sh worker.properties payments-sink.properties > sink.log 2>&1 &
$ python status.py payments-sink RUNNING
connector payments-sink RUNNING
task 0 RUNNINGThe task is RUNNING now. Let's read the DLQ with dlq.py and check the output file.
$ python dlq.py
value: id=p2;amount=12.50
__connect.errors.topic = payments
__connect.errors.offset = 1
__connect.errors.stage = VALUE_CONVERTER
__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
$ cat payments.out
{amount=25.0, id=p1}
{amount=40.0, id=p3}The bad record id=p2;amount=12.50 landed in payments-dlq, and the two good payments reached the file. The headers say where it came from and why: topic payments, offset 1, the VALUE_CONVERTER stage, and a serialization error. That's enough to fix the record and replay it.
DLQs are only for sink connectors, and they cover converters and transforms. Errors inside the connector's own write go to the DLQ only if the connector reports them (errant record reporter). So alert on the DLQ, and watch task status.
Question 4: A producer changed a field type and the consumers broke. Why do you need schemas and a schema registry?
Kafka only stores bytes. The broker never checks what's inside, so nothing stops a producer from changing a field.
Let's see it. v1_v2.py sends two orders: in the first, amount is a number, and in the second a team sends it as a string. Then totals.py, the finance consumer, adds up the amounts.
import json
from confluent_kafka import Producer
p = Producer({'bootstrap.servers': 'localhost:9092'})
v1 = {'order_id': 1001, 'amount': 25.0} # amount is a number
v2 = {'order_id': 1002, 'amount': '40.50'} # a team sends a string
for order in (v1, v2):
p.produce('orders', value=json.dumps(order))
p.flush()import json
from confluent_kafka import Consumer
c = Consumer({'bootstrap.servers': 'localhost:9092', 'group.id': 'finance',
'auto.offset.reset': 'earliest'})
c.subscribe(['orders'])
total, n = 0.0, 0
while n < 2:
m = c.poll(1.0)
if m is None or m.error():
continue
order, n = json.loads(m.value()), n + 1
total += order['amount']
print('order', order['order_id'], 'total', total)
c.close()$ python v1_v2.py
$ python totals.py
order 1001 total 25.0
Traceback (most recent call last):
File "totals.py", line 13, in <module>
total += order['amount']
TypeError: unsupported operand type(s) for +=: 'float' and 'str'The first order works, with a total of 25.0. The second crashes with TypeError: it can't add a string to a number. Every consumer finds out in production.
A schema makes the format a contract, in Avro, Protobuf or JSON Schema. Here, amount is a decimal.
{"type": "record", "name": "Order", "namespace": "shop",
"fields": [
{"name": "order_id", "type": "long"},
{"name": "amount", "type": {"type": "bytes",
"logicalType": "decimal", "precision": 10, "scale": 2}}
]}We could embed the schema in every message, like the JSON converter with schemas.enable=true did in Question 1, but that repeats it in every record. A schema registry stores each version once. It works like this:
- The producer's serializer registers or looks up the schema.
- It puts a small schema ID in each message, followed by the bytes.
- Consumers fetch the schema by ID.
- An incompatible schema is rejected before anything is written.
The schema registry is not part of Apache Kafka. The common one is Confluent Schema Registry, and the clouds have their own, like AWS Glue.
Question 5: A teammate asks which compatibility mode to use for schema evolution. How do backward and forward compatibility work?
Backward means consumers on the new schema can read data written with the old one. So we may delete fields, and add fields only with a default. Upgrade consumers first. It's the Confluent Schema Registry default.
Let's compare the modes side by side.
| Guarantee | Allowed changes | Upgrade first | |
|---|---|---|---|
| BACKWARD | new schema reads old data | delete fields, add fields with defaults | consumers |
| FORWARD | old schema reads new data | add fields, delete fields with defaults | producers |
| FULL | both directions | add or delete fields with defaults | either order |
| NONE | no check | anything | coordinate by hand |
Forward is the other way round: consumers on the old schema can read data written with the new one. We may add fields, and delete only fields with a default, and we upgrade producers first. Full is both, so either order works, but we may only add or delete fields with defaults. None turns the check off.
These modes check only the latest version. The transitive ones, like BACKWARD_TRANSITIVE, check every version, which matters when consumers replay old data.
Here is version two of our order schema. It adds a currency field with a default of USD, so it's backward compatible.
{"type": "record", "name": "Order", "namespace": "shop",
"fields": [
{"name": "order_id", "type": "long"},
{"name": "amount", "type": {"type": "bytes",
"logicalType": "decimal", "precision": 10, "scale": 2}},
{"name": "currency", "type": "string", "default": "USD"}
]}Backward, or backward transitive for replayed topics, with a default on every new field. A real breaking change goes to a new topic, with a migration plan.
Question 6: Connect workers keep losing tasks. How do standalone and distributed mode differ?
Standalone is one worker process, with configs and source offsets in local files, like the connect.offsets file we used. Fine on a laptop, but if that machine dies, the pipeline stops.
Distributed mode is a group of workers with the same group.id. They keep connector configs, offsets and status in three compacted Kafka topics, and we manage connectors through the REST API.
Let's start a real distributed worker.
bootstrap.servers=localhost:9092
group.id=connect-cluster
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
config.storage.topic=connect-configs
offset.storage.topic=connect-offsets
status.storage.topic=connect-status
# 1 only because this lab has one broker: use 3 in production
config.storage.replication.factor=1
offset.storage.replication.factor=1
status.storage.replication.factor=1
plugin.path=plugins
listeners=http://localhost:8308{"connector.class": "FileStreamSource", "tasks.max": "1",
"file": "orders.txt", "topic": "orders-copy"}$ connect-distributed.sh worker-distributed.properties > dist.log 2>&1 &
$ curl -s localhost:8308/ | jq -c .
{"version":"4.4.0","commit":"1156b2752aaa28ea","kafka_cluster_id":"p1n057GhRYecMDp97v4mSA"}It answers on port 8308 with version 4.4.0. Now one PUT request creates a connector.
$ curl -s -X PUT -H 'Content-Type: application/json' -d @source.json localhost:8308/connectors/orders-copy/config
{"name":"orders-copy","config":{"connector.class":"FileStreamSource","tasks.max":"1","file":"orders.txt"
,"topic":"orders-copy","name":"orders-copy"},"tasks":[],"type":"source"}
$ python status.py orders-copy RUNNING 8308
connector orders-copy RUNNING
task 0 RUNNING
$ curl -s localhost:8308/connectors/orders-copy/status | jq -c '.tasks'
[{"id":0,"state":"RUNNING","worker_id":"localhost:8308","version":"4.4.0"}]Its task 0 is RUNNING on the worker localhost:8308. Let's look at the internal topics.
$ kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic 'connect-.*' | grep PartitionCount | sort
Topic: connect-configs TopicId: <topic-id> PartitionCount: 1 ReplicationFactor: 1 Configs:
min.insync.replicas=1,cleanup.policy=compact
Topic: connect-offsets TopicId: <topic-id> PartitionCount: 25 ReplicationFactor: 1 Configs:
min.insync.replicas=1,cleanup.policy=compact
Topic: connect-status TopicId: <topic-id> PartitionCount: 5 ReplicationFactor: 1 Configs:
min.insync.replicas=1,cleanup.policy=compactThere they are: connect-configs with 1 partition, connect-offsets with 25 and connect-status with 5, all with cleanup.policy=compact.
When a worker dies or joins, the group rebalances and tasks move. A departed worker's tasks wait for scheduled.rebalance.max.delay.ms, five minutes by default, so a quick restart gets them back.
So lost tasks usually mean workers leaving the group: long GC pauses, too little memory, or lost broker connections. Check logs and task status.
Failed tasks don't restart on their own. Use POST /connectors/NAME/restart?includeTasks=true&onlyFailed=true.
Scenario 7: You must move orders from an operational database to the data lake in near real time with schema changes. What is your design?
Capture, don't poll. Let's build it step by step.
First, Debezium reads the orders database log and writes change events to a topic per table, keyed by the primary key, so all changes to one order stay in order.
Then we use Avro or Protobuf with a schema registry, in backward mode. A new nullable column becomes an optional field, the registry accepts it, and old consumers keep working. An incompatible change, like turning a number into text, is rejected, and the connector stops instead of writing bad data.
On the lake side, a sink connector, for example for S3 or Apache Iceberg, writes bronze tables with a DLQ and schema evolution on, so new columns show up.
We run both connectors in distributed mode on several workers, with replication factor three. CDC is at least once by default, so a lake job merges by key and log position into silver, absorbing duplicates and keeping deletes.
In short:
- Capture: Debezium CDC, one topic per table, keyed by primary key.
- Contract: Avro or Protobuf plus a registry, BACKWARD, optional new fields.
- Land: S3 or Iceberg sink, DLQ, schema evolution on.
- Run: distributed workers, RF 3, and an idempotent merge into silver by key and LSN, deletes included.
- Watch: task status, sink lag, the DLQ topic, and the replication slot.
Test a schema change end to end.
Key points to remember
- Kafka Connect: configuration, not code, for moving data in and out
- Log-based CDC with Debezium instead of polling
- errors.tolerance=all + a dead letter queue for sink connectors
- Schema registry; BACKWARD = upgrade consumers first
- Distributed mode: config, offsets, status in Kafka topics
The full script
This is a Python script that writes the files and runs every command in order against a local Kafka 4.4. Run pip install confluent-kafka first.
Show the full script that writes the files and runs every step
# ========================================================================
# Kafka Interview Q&A #7 - Kafka Connect, CDC and schemas
# TechBrothersIT. Sample data is invented for teaching.
# Every step below was run on Apache Kafka 4.4.0 (KRaft mode, 1 node) with confluent-kafka 2.16.0;
# the outputs in the video and the blog post are the real outputs. To run it yourself (Linux, macOS or WSL):
# 1. Download Kafka 4.4.0 from kafka.apache.org, extract it, and put its bin folder on your PATH.
# 2. Start a local cluster (see the Kafka quickstart; node 1 listens on localhost:9092).
# 3. pip install confluent-kafka then python Kafka_Interview_QA_07_Code.py in an empty folder.
# Steps that stop or start a broker say so in a comment: do that by hand at that point.
# ========================================================================
import os, subprocess
def write(path, text):
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
f.write(text.strip("\n") + "\n")
def run(cmd, expect_error=False):
print("\n$ " + cmd, flush=True)
rc = subprocess.run(cmd, shell=True).returncode
if rc != 0 and not expect_error:
raise SystemExit("stopped: this command was expected to succeed")
# Step 0: files used by the steps
write('orders.txt', """\
1001,c1,25.00
1002,c2,40.50
1003,c1,9.99
""")
write('worker.properties', """\
bootstrap.servers=localhost:9092
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=true
offset.storage.file.filename=connect.offsets
# lab only: flush source offsets every second (default 60 s)
offset.flush.interval.ms=1000
plugin.path=plugins
listeners=http://localhost:8307
""")
write('file-source.properties', """\
name=orders-file-source
connector.class=FileStreamSource
tasks.max=1
file=orders.txt
topic=orders-raw
""")
write('status.py', "import json, sys, time, urllib.request\n\nname, want = sys.argv[1], sys.argv[2]\nport = sys.argv[3] if len(sys.argv) > 3 else '8307'\nurl = 'http://localhost:%s/connectors/%s/status' % (port, name)\nfor _ in range(90):\n try:\n s = json.load(urllib.request.urlopen(url))\n if s['tasks'] and s['tasks'][0]['state'] == want:\n break\n except OSError:\n pass\n time.sleep(1)\nprint('connector', s['name'], s['connector']['state'])\nfor t in s['tasks']:\n print('task', t['id'], t['state'], t.get('trace', '').split('\\n')[0])\n")
write('read.py', "import sys\nfrom confluent_kafka import Consumer\n\ntopic, want = sys.argv[1], int(sys.argv[2])\nc = Consumer({'bootstrap.servers': 'localhost:9092', 'group.id': 'reader',\n 'auto.offset.reset': 'earliest', 'enable.auto.commit': False})\nc.subscribe([topic])\nrows, empty = [], 0\nwhile len(rows) < want and empty < 30:\n m = c.poll(1.0)\n if m is None or m.error():\n empty += 1\n continue\n rows.append('offset %d: %s' % (m.offset(), m.value().decode()))\nc.close()\nprint('\\n'.join(rows))\n")
write('stop.sh', """\
kill $(cat connect.pid)
while kill -0 $(cat connect.pid) 2>/dev/null; do sleep 0.5; done
""")
run('mkdir -p plugins && ln -s "$(dirname "$(which kafka-topics.sh)")/../libs/connect-file-4.4.0.jar" plugins/')
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders-raw --partitions 1')
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic payments --partitions 1')
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders --partitions 1')
# Step 1: Kafka Connect standalone: a file source connector, no code
run('timeout 300 connect-standalone.sh worker.properties file-source.properties > connect.log 2>&1 & echo $! > connect.pid')
run('python status.py orders-file-source RUNNING')
run('python read.py orders-raw 3')
# Step 2: A new line in the file: only the new line is sent, and Connect stores the source offset
run('echo "1004,c3,12.00" >> orders.txt')
run('python read.py orders-raw 4')
run('sleep 3; curl -s localhost:8307/connectors/orders-file-source/offsets | jq -c .')
# Step 3: Stop the Connect worker
run('sh stop.sh')
# Step 4: Three payments, the second one is not JSON
write('produce_payments.py', """\
from confluent_kafka import Producer
rows = ['{"id": "p1", "amount": 25.0}',
'id=p2;amount=12.50', # not JSON: a bad record
'{"id": "p3", "amount": 40.0}']
p = Producer({'bootstrap.servers': 'localhost:9092'})
for v in rows:
p.produce('payments', value=v)
p.flush()
print('sent', len(rows), 'payments')
""")
write('payments-sink.properties', """\
name=payments-sink
connector.class=FileStreamSink
tasks.max=1
topics=payments
file=payments.out
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
""")
run('python produce_payments.py')
# Step 5: A file sink with the default errors.tolerance=none: the task fails on the bad record
run('timeout 300 connect-standalone.sh worker.properties payments-sink.properties > sink.log 2>&1 & echo $! > connect.pid')
run('python status.py payments-sink FAILED')
run('sleep 2; wc -l payments.out')
# Step 6: Stop the Connect worker
run('sh stop.sh')
# Step 7: errors.tolerance=all and a dead letter queue topic: restart the sink
write('payments-sink.properties', """\
name=payments-sink
connector.class=FileStreamSink
tasks.max=1
topics=payments
file=payments.out
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
errors.tolerance=all
errors.deadletterqueue.topic.name=payments-dlq
# 1 only because this lab has one broker (default 3)
errors.deadletterqueue.topic.replication.factor=1
errors.deadletterqueue.context.headers.enable=true
""")
write('dlq.py', """\
from confluent_kafka import Consumer
c = Consumer({'bootstrap.servers': 'localhost:9092', 'group.id': 'dlq-check',
'auto.offset.reset': 'earliest'})
c.subscribe(['payments-dlq'])
m = None
while m is None or m.error():
m = c.poll(1.0)
print('value:', m.value().decode())
for k, v in m.headers():
if k.endswith(('.topic', '.offset', '.stage', 'exception.class.name', 'exception.message')):
print(k, '=', v.decode().split(':')[0])
c.close()
""")
run('timeout 300 connect-standalone.sh worker.properties payments-sink.properties > sink.log 2>&1 & echo $! > connect.pid')
run('python status.py payments-sink RUNNING')
# Step 8: The bad record is in the dead letter queue, with headers; the good ones reached the sink
run('python dlq.py')
run('sleep 2; cat payments.out')
# Step 9: Stop the Connect worker
run('sh stop.sh')
# Step 10: Start a distributed worker
write('worker-distributed.properties', """\
bootstrap.servers=localhost:9092
group.id=connect-cluster
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
config.storage.topic=connect-configs
offset.storage.topic=connect-offsets
status.storage.topic=connect-status
# 1 only because this lab has one broker: use 3 in production
config.storage.replication.factor=1
offset.storage.replication.factor=1
status.storage.replication.factor=1
plugin.path=plugins
listeners=http://localhost:8308
""")
write('source.json', """\
{"connector.class": "FileStreamSource", "tasks.max": "1",
"file": "orders.txt", "topic": "orders-copy"}
""")
run('timeout 300 connect-distributed.sh worker-distributed.properties > dist.log 2>&1 & echo $! > connect.pid')
run('until curl -sf localhost:8308/ > /dev/null; do sleep 1; done; curl -s localhost:8308/ | jq -c .')
# Step 11: Create a connector with the REST API; its task runs on the worker
run("curl -s -X PUT -H 'Content-Type: application/json' -d @source.json localhost:8308/connectors/orders-copy/config")
run('python status.py orders-copy RUNNING 8308')
run("curl -s localhost:8308/connectors/orders-copy/status | jq -c '.tasks'")
# Step 12: The three internal topics of distributed mode
run("kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic 'connect-.*' | grep PartitionCount | sort")
# Step 13: Stop the Connect worker
run('sh stop.sh')
# Step 14: A producer changes a field type: the consumer breaks
write('v1_v2.py', """\
import json
from confluent_kafka import Producer
p = Producer({'bootstrap.servers': 'localhost:9092'})
v1 = {'order_id': 1001, 'amount': 25.0} # amount is a number
v2 = {'order_id': 1002, 'amount': '40.50'} # a team sends a string
for order in (v1, v2):
p.produce('orders', value=json.dumps(order))
p.flush()
""")
write('totals.py', """\
import json
from confluent_kafka import Consumer
c = Consumer({'bootstrap.servers': 'localhost:9092', 'group.id': 'finance',
'auto.offset.reset': 'earliest'})
c.subscribe(['orders'])
total, n = 0.0, 0
while n < 2:
m = c.poll(1.0)
if m is None or m.error():
continue
order, n = json.loads(m.value()), n + 1
total += order['amount']
print('order', order['order_id'], 'total', total)
c.close()
""")
run('python v1_v2.py')
run('python totals.py', expect_error=True) # EXPECTED TO FAIL: the video shows this error
Part 8: Stream Processing and Kafka Streams
The sample data on this page is invented for teaching. Every terminal output is the real output of the command above it, run on Apache Kafka 4.4.0 in KRaft mode with confluent-kafka 2.16.0; topic IDs are masked. Part of the Real Time Kafka Interview Questions and Answers series on TechBrothersIT. Features, limits and their status (GA or preview) are current as of October 2026; check the official Apache Kafka documentation (kafka.apache.org) for anything you quote in an interview.
No comments:
Post a Comment
Note: Only a member of this blog may post a comment.