Do I need to write Java to use Kafka Connect?

No. A connector is a plugin — you configure one with JSON over the REST API (connector.class, plus its own settings) and Connect's worker process runs it. Writing a custom connector is a Java API for cases with no existing plugin, but hundreds of ready-made connectors exist for databases, files, cloud storage, and more.

What's the difference between standalone and distributed mode?

Standalone mode runs one worker process with connector configs in local files or via REST, and stores offsets in a local file — good for development and single-node use, like this lesson. Distributed mode runs a cluster of workers that share load and store offsets in Kafka topics, so a worker crashing doesn't lose track of position; that's the production mode.

How does a source connector know where it left off after a restart?

Every source task periodically commits an offset — an arbitrary JSON object it defines the shape of. FileStreamSourceConnector's offset is just a byte position in the file. On restart it reads that position back and resumes from there, which is why restarting this lesson's connector on an unchanged file produces zero new records: it already knows it read to the end.

Is FileStreamSourceConnector meant for production use?

No — Apache Kafka ships it as a reference implementation to demonstrate the Connect framework, not as a supported production connector. Real deployments use connectors like Debezium (database CDC), the S3 sink, or JDBC source/sink, which handle schema evolution, exactly-once delivery, and restart semantics far beyond what the file connector attempts.

Kafka Connect: Source and Sink Connectors, and a Live Offset

Move data in and out of Kafka without writing a producer or consumer — a real FileStreamSource connector, a FileStreamSink connector, and the REST API that reports exactly how far each one has read.

By SkillByExample · Published Sep 16, 2026

kafka kafka-connect connectors integration

Producing and consuming by hand works for an application you control. Getting data into or out of Kafka from something you don’t control — a directory of log files, a database, a warehouse — is what Kafka Connect exists for.

Starting a Connect worker

Connect runs as its own process, separate from the broker, with a REST API on port 8083:

bin/connect-standalone.sh config/connect-standalone.properties
curl -s localhost:8083/

{"version":"3.9.0","commit":"a60e31147e6b01ee","kafka_cluster_id":"lesson-cluster-01"}

The worker is up and talking to the cluster. No connectors are running yet — it’s an empty runtime waiting for configuration.

A source connector: file into topic

FileStreamSourceConnector ships with Kafka as a reference implementation — it tails a file and produces each line as a record. Seed a file, then register the connector over REST:

printf "line one\nline two\nline three\n" > /tmp/connect-input.txt

curl -s -X POST -H "Content-Type: application/json" localhost:8083/connectors -d '{
  "name": "local-file-source",
  "config": {
    "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
    "tasks.max": "1",
    "file": "/tmp/connect-input.txt",
    "topic": "connect-test"
  }
}'

{"name":"local-file-source","config":{"connector.class":"org.apache.kafka.connect.file.FileStreamSourceConnector","tasks.max":"1","file":"/tmp/connect-input.txt","topic":"connect-test","name":"local-file-source"},"tasks":[{"connector":"local-file-source","task":0}],"type":"source"}

Check what actually landed in the topic:

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic connect-test --from-beginning --timeout-ms 8000

line one
line two
line three
Processed a total of 3 messages

Three lines in the file, three records in the topic — no producer code written.

The REST API reports real status and offsets

curl -s localhost:8083/connectors/local-file-source/status

{"name":"local-file-source","connector":{"state":"RUNNING","worker_id":"172.17.0.2:8083"},"tasks":[{"id":0,"state":"RUNNING","worker_id":"172.17.0.2:8083"}],"type":"source"}

curl -s localhost:8083/connectors/local-file-source/offsets

{"offsets":[{"partition":{"filename":"/tmp/connect-input.txt"},"offset":{"position":29}}]}

The offset isn’t a Kafka offset — it’s whatever the connector defines. For a file source that’s a byte position: 29 bytes into /tmp/connect-input.txt, exactly the length of the three lines written so far. Append a fourth line and the same file keeps streaming:

echo "line four" >> /tmp/connect-input.txt

$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic connect-test --from-beginning --timeout-ms 8000
line one
line two
line three
line four
Processed a total of 4 messages

No restart, no re-registration — FileStreamSourceConnector polls the file for new content on its own schedule.

A sink connector: topic into file

The mirror image — FileStreamSinkConnector reads a topic and appends each record to a file:

curl -s -X POST -H "Content-Type: application/json" localhost:8083/connectors -d '{
  "name": "local-file-sink",
  "config": {
    "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
    "tasks.max": "1",
    "file": "/tmp/connect-output.txt",
    "topics": "connect-test"
  }
}'

cat /tmp/connect-output.txt

line one
line two
line three
line four

Every record the source connector produced, written back out to a file by the sink — a complete file-to-Kafka-to-file pipeline with two JSON POST requests and zero custom code.

Listing and checking connectors

curl -s localhost:8083/connectors

["local-file-source","local-file-sink"]

Both connectors, by name — the same names used in every status/offset/delete call above. In production this REST API is how you’d script deploys, health checks, and rollbacks.

Practice

1. Delete the source connector, then check whether the sink still has data to read.

curl -s -X DELETE localhost:8083/connectors/local-file-source
curl -s localhost:8083/connectors

["local-file-sink"]

Deleting a connector stops and removes it, but doesn’t touch the topic or any records already produced. The sink connector is unaffected — it just stops receiving new records once nothing is producing to connect-test.

2. Delete and re-add the source connector on the unchanged file. How many new records land in the topic?

curl -s localhost:8083/connectors/local-file-source/offsets

{"offsets":[{"partition":{"filename":"/tmp/connect-input.txt"},"offset":{"position":39}}]}

Still 39 — the same position as before deletion, and the topic still holds exactly the 4 records from before. A source connector’s offset tracks position, not whether it has run before; a file whose length hasn’t changed has nothing new to report, deleted-and-recreated or not.

3. Point a new source connector at a file that doesn't exist yet. What does its status show?

{"name":"local-file-source-missing","connector":{"state":"RUNNING","worker_id":"172.17.0.4:8083"},"tasks":[{"id":0,"state":"RUNNING","worker_id":"172.17.0.4:8083"}]}

Both connector and task report RUNNING — no failure. The worker log shows why:

WARN [local-file-source-missing|task-0] Couldn't find file /tmp/does-not-exist.txt for
FileStreamSourceTask, sleeping to wait for it to be created

It polls once a second, waiting for the file to appear, rather than failing outright — useful against a log file that a different process hasn’t created yet, less useful as a signal that a config typo exists. Nothing landed in connect-test-missing, since the task never got past the wait loop.

4. Pause the sink connector, then resume it.

curl -s -X PUT localhost:8083/connectors/local-file-sink/pause
curl -s localhost:8083/connectors/local-file-sink/status

{"name":"local-file-sink","connector":{"state":"PAUSED","worker_id":"172.17.0.4:8083"},"tasks":[{"id":0,"state":"PAUSED","worker_id":"172.17.0.4:8083"}],"type":"sink"}

curl -s -X PUT localhost:8083/connectors/local-file-sink/resume
curl -s localhost:8083/connectors/local-file-sink/status

{"name":"local-file-sink","connector":{"state":"RUNNING","worker_id":"172.17.0.4:8083"},"tasks":[{"id":0,"state":"RUNNING","worker_id":"172.17.0.4:8083"}],"type":"sink"}

Records keep accumulating in the topic while paused — Connect doesn’t lose them, it just stops consuming. Resuming picks up exactly where the committed offset left off, the same way a consumer group would.

5. Change the sink's tasks.max to 3. Does it actually run 3 tasks?

curl -s -X PUT -H "Content-Type: application/json" localhost:8083/connectors/local-file-sink/config \
  -d '{"connector.class":"org.apache.kafka.connect.file.FileStreamSinkConnector","tasks.max":"3","file":"/tmp/connect-output.txt","topics":"connect-test"}'

{"name":"local-file-sink","connector":{"state":"RUNNING","worker_id":"172.17.0.4:8083"},"tasks":[{"id":0,"state":"RUNNING","worker_id":"172.17.0.4:8083"},{"id":1,"state":"RUNNING","worker_id":"172.17.0.4:8083"},{"id":2,"state":"RUNNING","worker_id":"172.17.0.4:8083"}],"type":"sink"}

Yes — three tasks, all RUNNING. Connect starts exactly as many tasks as you ask for; it doesn’t clamp to partition count at the config level. But connect-test still has only 1 partition, and a sink task consumes whole partitions via a consumer group — so the group rebalance protocol hands that one partition to a single task, and the other two sit connected with nothing assigned. Three tasks running is not the same thing as three tasks doing work.

Next: securing that same cluster so not every client can read or write every topic.