Kafka Streams assigns partitions to tasks, and then tasks to instances.
Here’s how it works under the hood, using a simple Kafka Streams application that reads from a topic input-topic and writes to output-topic.
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.Topology;
import java.util.Properties;
public class KafkaStreamsPartitionAssignmentDemo {
public static void main(String[] args) {
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "partition-assignment-demo");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CONFIG, Serdes.String().getClass());
// Setting commit interval to a small value for faster rebalancing demonstration
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.AT_LEAST_ONCE);
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 5000); // 5 seconds
StreamsBuilder builder = new StreamsBuilder();
builder.stream("input-topic")
.mapValues(value -> "Processed: " + value)
.to("output-topic");
Topology topology = builder.build();
KafkaStreams streams = new KafkaStreams(topology, props);
// Register a shutdown hook, so that we can gracefully close the Streams application.
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
System.out.println("Shutting down Kafka Streams application...");
streams.close();
System.out.println("Kafka Streams application shut down.");
}));
System.out.println("Starting Kafka Streams application...");
streams.start();
System.out.println("Kafka Streams application started.");
}
}
First, create an input-topic with a few partitions. Let’s say 3 partitions:
kafka-topics --create --topic input-topic --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1
Now, run the KafkaStreamsPartitionAssignmentDemo application. You’ll see it start up. If you run a second instance of the same application (with the same application.id), you’ll observe a rebalance. Kafka Streams will distribute the partitions of input-topic among the available instances.
The core concept is that Kafka Streams doesn’t assign partitions directly to application instances. Instead, it maps partitions to tasks. A task is the smallest unit of parallelism in Kafka Streams. For a stream processing application, a task typically processes one or more partitions of an input topic.
If your topology is simple (like the example above, a single stream().mapValues().to()), each task will be responsible for a single partition. If your topology has branches or joins, Kafka Streams might create tasks that are responsible for multiple partitions from different topics, or even multiple partitions from the same topic if they feed into different processing steps within the same task.
The number of tasks is determined by the maximum number of partitions across all input topics for the source of your topology. In our example, input-topic has 3 partitions. Therefore, Kafka Streams will create 3 tasks.
Let’s say you have 2 instances of your KafkaStreamsPartitionAssignmentDemo application running. Kafka Streams will then assign these 3 tasks to the 2 instances. It’s a round-robin assignment.
Instance 1 might get Task 0 and Task 1. Instance 2 might get Task 2.
Each task then internally manages the state for the partitions assigned to it. If a task is responsible for partition 0, it will read from partition 0, process records, and write to the output topic. The key here is that Kafka’s partition model guarantees that all records with the same key go to the same partition. Since a task is assigned a partition, it’s guaranteed to process all records for the keys that map to that partition. This is how Kafka Streams ensures ordered processing per key.
When you start your application, it joins a consumer group. The APPLICATION_ID_CONFIG acts as the consumer group ID. Kafka’s group coordination mechanism kicks in. The broker acts as the group coordinator. When a new instance joins or an existing one leaves, a rebalance is triggered. During a rebalance, the assigned partitions (and thus tasks) are redistributed among the active instances in the group.
If you were to inspect the Kafka consumer groups, you’d see your application.id as a group. The Kafka Streams client uses Kafka’s consumer API under the hood. The "consumer" associated with your application ID will be managed by the Kafka broker for group coordination.
The most surprising thing is that the number of tasks is determined by the maximum number of partitions of any input topic that is a source in your topology. If you have two source topics, topicA with 5 partitions and topicB with 2 partitions, and both are sources, Kafka Streams will create 5 tasks. Each of these 5 tasks will then be assigned partitions from topicA. Some tasks might also be assigned partitions from topicB if the topology dictates it and there are enough tasks to cover topicB’s partitions. The assignment logic is complex but aims to maximize parallelism based on the bottleneck partition count.
The next thing you’ll encounter is how Kafka Streams handles stateful operations and the underlying changelog topics when tasks are reassigned.