Hands-On: FileStream Sink

Summary

The FileStream Sink Connector is the counterpart to the FileStream Source Connector we used in Chapter 5. While simple, it's perfect for learning the fundamentals of how sink connectors work.

UNDERSTANDING FILESTREAM SINK

The FileStream Sink Connector consumes messages from a Kafka topic and appends each message value as a line in an output file. It's straightforward but demonstrates all the core concepts of sink connectors.

Here's how the FileStream Sink Connector works. It consumes from a Kafka topic and writes each message to a file.

The connector subscribes to the topic, consumes new messages as they arrive, and appends each message value to the output file. Each message becomes a single line in the file.

CONNECTOR CONFIGURATION

The following shows how to configure a FileStream Sink Connector. Like all connectors, it's configured using JSON.

1{
2  "name": "file-sink-connector",
3  "config": {
4    "connector.class": "FileStreamSink",
5    "tasks.max": "1",
6    "file": "/tmp/output-file.txt",
7    "topics": "messages-topic"
8  }
9}

name: A unique name for your connector instance.

connector.class: Specifies the FileStream Sink connector class. You can use "FileStreamSink" or the full class name "org.apache.kafka.connect.file.FileStreamSinkConnector".

tasks.max: Maximum number of tasks. For FileStream Sink, use 1 since multiple tasks can't write to the same file safely.

file: The absolute path to the output file. If omitted, output goes to standard output (stdout).

topics: The Kafka topic(s) to consume from. Can be a comma-separated list for multiple topics.

HANDS-ON: RUNNING FILESTREAM SINK

walk through running a FileStream Sink Connector in standalone mode.

Here's the workflow we'll follow to set up and run the connector.

Step 1: Create the Kafka Topic

First, create a topic that the sink connector will consume from:

1kafka-topics --create \
2  --bootstrap-server localhost:9092 \
3  --topic messages-topic \
4  --partitions 3 \
5  --replication-factor 1

Step 2: Configure the Connector

Create a connector configuration file at /tmp/filestream-sink.properties:

1name=file-sink-connector
2connector.class=org.apache.kafka.connect.file.FileStreamSinkConnector
3tasks.max=1
4file=/tmp/output-file.txt
5topics=messages-topic

Step 3: Start Kafka Connect in Standalone Mode

Run Kafka Connect with the FileStream Sink configuration:

1connect-standalone \
2  config/connect-standalone.properties \
3  /tmp/filestream-sink.properties

Once the connector starts, it subscribes to the topic and waits for messages. You'll see log output showing it's connected and ready to consume.

Step 4: Produce Messages

In another terminal, produce some messages to the topic:

1kafka-console-producer \
2  --bootstrap-server localhost:9092 \
3  --topic messages-topic

Then type messages:

1Hello from Kafka!
2This is message 2
3Message 3 here

Press Ctrl+C to exit the producer.

Step 5: Verify Output File

Check the output file to see the consumed messages:

1cat /tmp/output-file.txt

You should see:

1Hello from Kafka!
2This is message 2
3Message 3 here

Perfect! The sink connector consumed the messages from Kafka and wrote them to the file, one message per line.

UNDERSTANDING THE DATA FLOW

Let's visualize what happened during this hands-on exercise.

The producer sent messages to Kafka. The sink connector polled the topic, received the messages in batches, wrote them to the file, and committed offsets to track its progress.

OFFSET MANAGEMENT

The FileStream Sink Connector tracks which messages it has consumed using Kafka consumer offsets.

In standalone mode, offsets are stored in a local file. In distributed mode, they're stored in an internal Kafka topic. The connector commits offsets only after successfully writing to the file, ensuring at-least-once delivery.

TESTING RESTART BEHAVIOR

Let's verify that the connector doesn't re-consume messages after a restart.

Step 1: Stop the Connector

Press Ctrl+C to stop the running connector.

Step 2: Produce More Messages

While the connector is stopped, produce more messages:

1kafka-console-producer \
2  --bootstrap-server localhost:9092 \
3  --topic messages-topic

Type:

1Message 4 (connector stopped)
2Message 5 (connector stopped)

Step 3: Restart the Connector

Restart the connector:

1connect-standalone \
2  config/connect-standalone.properties \
3  /tmp/filestream-sink.properties

Step 4: Verify File Contents

Check the file again:

1cat /tmp/output-file.txt

You should now see:

1Hello from Kafka!
2This is message 2
3Message 3 here
4Message 4 (connector stopped)
5Message 5 (connector stopped)

The connector resumed from where it left off, consuming only the new messages. It didn't re-process the first three messages because it had committed offsets for them.

COMBINING SOURCE AND SINK

For a complete example, you can run both FileStream Source and FileStream Sink to create a simple file-to-file pipeline.

The source connector reads from input.txt and produces to Kafka. The sink connector consumes from Kafka and writes to output.txt. This demonstrates how Kafka Connect can build data pipelines without any custom code.

Let's recap what we learned in this hands-on lesson:

Key ideas

The FileStream Sink Connector consumes messages from Kafka topics and writes them to files. It's simple but demonstrates core sink connector concepts.

Configuration specifies the topic to consume from, the output file path, and the connector class.

The connector continuously polls for new messages and appends them to the file.

Offset management ensures the connector doesn't re-consume messages after restarts.

Like the source connector, offsets are stored locally in standalone mode and in Kafka topics in distributed mode.

You can combine source and sink connectors to build complete data pipelines without writing code.

While FileStream Sink is basic, the concepts apply to all sink connectors: configuration, consumption, offset management, and reliable delivery.