Stream topology in Kafka - Commands & Configuration

Introduction

When you want to process data streams in real time, you need to define how data flows and transforms between different steps. A stream topology is a map of these steps and connections, helping you organize and run your data processing smoothly.

When you want to filter and transform live data from sensors before saving it.

When you need to join two streams of data, like user clicks and purchases, to analyze behavior.

When you want to aggregate data over time, such as counting events per minute.

When you want to build a pipeline that reads from one topic, processes data, and writes to another topic.

When you want to visualize or debug how your streaming application processes data step-by-step.

Config File - stream_topology.properties

stream_topology.properties

application.id=stream-topology-example
bootstrap.servers=localhost:9092
processing.guarantee=exactly_once
cache.max.bytes.buffering=10485760
commit.interval.ms=1000

This file sets up the Kafka Streams application properties:

• application.id: Unique name for your stream app.
• bootstrap.servers: Kafka server address.
• processing.guarantee: Ensures data is processed exactly once.
• cache.max.bytes.buffering: Controls memory buffering for performance.
• commit.interval.ms: How often to save progress.

Commands

Create an input topic with 3 partitions to receive streaming data.

Terminal

kafka-topics --create --topic input-topic --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1

Expected Output

Created topic input-topic.

→

--partitions - Number of partitions for parallelism

→

--replication-factor - Number of copies for fault tolerance

Create an output topic where processed data will be sent.

Terminal

kafka-topics --create --topic output-topic --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1

Expected Output

Created topic output-topic.

→

--partitions - Number of partitions for parallelism

→

--replication-factor - Number of copies for fault tolerance

Start a producer to send sample data to the input topic for processing.

Terminal

kafka-console-producer --topic input-topic --bootstrap-server localhost:9092

Expected Output

No output (command runs silently)

→

--topic - Topic to send data to

→

--bootstrap-server - Kafka server address

Consume and display processed data from the output topic to verify the stream topology works.

Terminal

kafka-console-consumer --topic output-topic --bootstrap-server localhost:9092 --from-beginning

Expected Output

processed-data-example

→

--from-beginning - Read all messages from the start

Key Concept

If you remember nothing else from this pattern, remember: a stream topology defines how data flows and transforms step-by-step in a Kafka Streams application.

Common Mistakes

Not creating the input and output topics before running the stream application.

The stream app will fail because it cannot read or write data without these topics.

Always create all required topics with correct partitions and replication before starting the stream.

Using the same topic for input and output in the topology.

This can cause infinite loops or data corruption in the stream processing.

Use separate topics for input and output to keep data flow clear and safe.

Not setting the application.id property uniquely for each stream app instance.

Kafka Streams uses this ID to track state; duplicates cause conflicts and errors.

Set a unique application.id in the properties file for each distinct stream application.

Summary

Create input and output Kafka topics to hold streaming data.

Configure the stream application with properties like application.id and bootstrap servers.

Send data to the input topic and consume processed data from the output topic to verify the flow.

Practice

1. What does a Kafka Streams topology represent?

easy

A. The configuration settings for Kafka brokers

B. The flow and processing steps of data in a Kafka Streams application

C. The list of Kafka topics available in the cluster

D. The schema of messages in a Kafka topic

Solution

1. Step 1: Understand the role of topology in Kafka Streams

   A topology defines how data flows and is processed in a Kafka Streams app.

2. Step 2: Differentiate topology from other Kafka concepts

   It is not about broker configs, topic lists, or message schemas, but about processing steps.

3. Final Answer:

   The flow and processing steps of data in a Kafka Streams application -> Option B

4. Quick Check:

   Topology = Data flow and processing [OK]

Hint: Topology means data flow and processing steps [OK]

Common Mistakes:

• Confusing topology with Kafka broker settings
• Thinking topology lists topics
• Assuming topology defines message schema

2. Which code snippet correctly builds a Kafka Streams topology?

easy

A. StreamsBuilder builder = new StreamsBuilder(); builder.run();

B. StreamsBuilder builder = new StreamsBuilder(); builder.start();

C. StreamsBuilder builder = new StreamsBuilder(); builder.create();

D. StreamsBuilder builder = new StreamsBuilder(); builder.build();

Solution

1. Step 1: Recall the correct method to build topology

   The method to finalize and get the topology is build().

2. Step 2: Check each option's method

   Only StreamsBuilder builder = new StreamsBuilder(); builder.build(); uses build(), others use invalid methods.

3. Final Answer:

   StreamsBuilder builder = new StreamsBuilder(); builder.build(); -> Option D

4. Quick Check:

   Use build() to create topology [OK]

Hint: Use build() method on StreamsBuilder to create topology [OK]

Common Mistakes:

• Using start(), create(), or run() instead of build()
• Confusing topology building with starting the stream

3. Given the code snippet:

StreamsBuilder builder = new StreamsBuilder();
builder.stream("input-topic").mapValues(value -> value.toUpperCase()).to("output-topic");
Topology topology = builder.build();
System.out.println(topology.describe());

What will the printed topology description include?

medium

A. A sink node for 'input-topic' and a source node for 'output-topic'

B. Only a source node for 'input-topic' without any processing

C. A source node for 'input-topic', a processor node for mapValues, and a sink node for 'output-topic'

D. An error because mapValues is not allowed in topology

Solution

1. Step 1: Analyze the stream processing steps

   The code reads from 'input-topic', applies mapValues to uppercase, then writes to 'output-topic'.

2. Step 2: Understand topology description output

   The description shows source nodes (input), processor nodes (mapValues), and sink nodes (output).

3. Final Answer:

   A source node for 'input-topic', a processor node for mapValues, and a sink node for 'output-topic' -> Option C

4. Quick Check:

   Topology describe shows source, processor, sink [OK]

Hint: Topology describe shows source, processor, and sink nodes [OK]

Common Mistakes:

• Thinking mapValues is not part of topology
• Confusing source and sink nodes
• Expecting no processor nodes in description

4. Identify the error in this Kafka Streams topology code:

StreamsBuilder builder = new StreamsBuilder();
builder.stream("input-topic").filter(value -> value.length() > 5);
Topology topology = builder.build();
System.out.println(topology.describe());

medium

A. The filtered stream is not sent to any sink topic, so no output node is created

B. The filter method is invalid and causes a syntax error

C. StreamsBuilder cannot build topology without a source topic

D. Calling build() before defining streams causes runtime error

Solution

1. Step 1: Check the stream processing chain

   The code filters the stream but does not write the result to any output topic.

2. Step 2: Understand topology implications

   Without a sink node, the topology has no output, which is usually an error or incomplete.

3. Final Answer:

   The filtered stream is not sent to any sink topic, so no output node is created -> Option A

4. Quick Check:

   Missing sink node causes incomplete topology [OK]

Hint: Always send processed stream to sink topic [OK]

Common Mistakes:

• Assuming filter causes syntax error
• Thinking build() requires no streams
• Ignoring missing sink topic after processing

5. You want to create a Kafka Streams topology that reads from two topics, merges their streams, filters messages containing 'error', and writes to an output topic. Which code snippet correctly builds this topology?

hard

A. StreamsBuilder builder = new StreamsBuilder();
KStream stream1 = builder.stream("topic1");
KStream stream2 = builder.stream("topic2");
KStream merged = stream1.merge(stream2);
merged.filter((k, v) -> v.contains("error")).to("output-topic");
Topology topology = builder.build();

B. StreamsBuilder builder = new StreamsBuilder();
KStream stream = builder.stream("topic1", "topic2");
stream.filter(v -> v.contains("error")).to("output-topic");
Topology topology = builder.build();

C. StreamsBuilder builder = new StreamsBuilder();
KStream stream1 = builder.stream("topic1");
KStream stream2 = builder.stream("topic2");
KStream merged = stream1.join(stream2, (v1, v2) -> v1 + v2);
merged.filter((k, v) -> v.contains("error")).to("output-topic");
Topology topology = builder.build();

D. StreamsBuilder builder = new StreamsBuilder();
builder.stream("topic1").filter(v -> v.contains("error")).to("output-topic");
builder.stream("topic2").filter(v -> v.contains("error")).to("output-topic");
Topology topology = builder.build();

Solution

1. Step 1: Understand the requirement

   Read from two topics, merge streams, filter messages with 'error', write to output.

2. Step 2: Evaluate each option

   StreamsBuilder builder = new StreamsBuilder(); KStream stream1 = builder.stream("topic1"); KStream stream2 = builder.stream("topic2"); KStream merged = stream1.merge(stream2); merged.filter((k, v) -> v.contains("error")).to("output-topic"); Topology topology = builder.build(); merges streams correctly, filters, then writes once. StreamsBuilder builder = new StreamsBuilder(); KStream stream = builder.stream("topic1", "topic2"); stream.filter(v -> v.contains("error")).to("output-topic"); Topology topology = builder.build(); incorrectly tries to stream two topics in one call. StreamsBuilder builder = new StreamsBuilder(); KStream stream1 = builder.stream("topic1"); KStream stream2 = builder.stream("topic2"); KStream merged = stream1.join(stream2, (v1, v2) -> v1 + v2); merged.filter((k, v) -> v.contains("error")).to("output-topic"); Topology topology = builder.build(); joins streams (not merge). StreamsBuilder builder = new StreamsBuilder(); builder.stream("topic1").filter(v -> v.contains("error")).to("output-topic"); builder.stream("topic2").filter(v -> v.contains("error")).to("output-topic"); Topology topology = builder.build(); filters separately and writes twice, not merging.

3. Final Answer:

   Option A correctly merges, filters, and writes to output topic -> Option A

4. Quick Check:

   Merge streams, filter, then write [OK]

Hint: Merge streams before filtering and writing output [OK]

Common Mistakes:

• Using join instead of merge for combining streams
• Trying to stream multiple topics in one call incorrectly
• Writing filtered streams separately without merging