Developing Event-driven Applications with Apache Kafka and Red Hat AMQ Streams

You can increase the processing rate (i.e. the amount of data that your application can process over time) of a Kafka Streams topology by adding more threads or new instances of your stream processing application. When you add more instances of your application, Kafka Streams distributes some of the tasks that were running among the new instances. Therefore, to achieve scalability in Kafka Streams, it is important to know the number of partitions that the topic has.

You can have as many tasks as the number of partitions you have in the processing topics. If you start more instances than the number of partitions, then the redundant instances do not process any data. However, in the case that any of the active instances crashes down, Kafka uses the redundant instances as a fall back. To have more active instances than the number of partitions in the topic, you can use Kafka Streams to move the records to another topic with more partitions. You would only use this topic for the Kafka Streams processing.

For example, consider a topic with four partitions and only one application instance. The single instance reads all the data from all partitions, as the following image illustrates.

Figure 4.5: Single Kafka Streams instance reading from four partitions

If you start another application instance, however, then the reading workload is distributed across the different instances, as the following image illustrates.

Figure 4.6: Two Kafka Streams instances reading from four partitions

You can decrease your Kafka Streams application processing rate by removing threads or application instances. The tasks assigned to the stopped instances are distributed across the remaining instances. The state of the stopped instances is also copied to the remaining instances, which allows you to resume the transformations.

For example, consider two instances of Kafka Streams reading from a topic with two partitions called users, where records are keyed by user ID. Every instance reads from a partition, as the following image illustrates.

Figure 4.7: Users' records keyed by ID

Consider that you must count the number of users that belong to the same country. To group the users by the country field, you use the groupBy method of the KStream object. Then, you use the count aggregation.

KStream<Integer, User> userStreams = builder.stream(
        "users",
        Consumed.with(intSerde, userSerde)
    );

userStreams.groupBy((key, value) -> value.country).count();

However, users with the same country are read by different Kafka Streams instances, which means that it is not possible to group them because no single instance has all the records. To overcome this problem, Kafka Streams creates another topic that uses the country field as key, instead of id. This way, users with the same key (the same country) go to the same partition and a single instance reads them. This is an example of repartitioning.

Figure 4.8: Users' records keyed by country as a result of groupBy

These kinds of topics are called partition topics, and they only work in the context of Kafka Streams without modifying the original topic.

Technically, when Kafka Streams performs repartitioning, it splits your topology in to sub-topologies. If you describe the topology from the previous example, the result should be similar to the following output.

Topologies:
   Sub-topology: 0
    Source: KSTREAM-SOURCE-0000000000 (topics: [users]) 
      --> KSTREAM-KEY-SELECT-0000000001
    Processor: KSTREAM-KEY-SELECT-0000000001 (stores: [])
      --> KSTREAM-FILTER-0000000005
      <-- KSTREAM-SOURCE-0000000000
    Processor: KSTREAM-FILTER-0000000005 (stores: [])
      --> KSTREAM-SINK-0000000004
      <-- KSTREAM-KEY-SELECT-0000000001
    Sink: KSTREAM-SINK-0000000004 (topic: KSTREAM-AGGREGATE-STATE-STORE-0000000002-repartition) 
      <-- KSTREAM-FILTER-0000000005

  Sub-topology: 1
    Source: KSTREAM-SOURCE-0000000006 (topics: [KSTREAM-AGGREGATE-STATE-STORE-0000000002-repartition]) 
      --> KSTREAM-AGGREGATE-0000000003
    Processor: KSTREAM-AGGREGATE-0000000003 (stores: [KSTREAM-AGGREGATE-STATE-STORE-0000000002])
      --> none
      <-- KSTREAM-SOURCE-0000000006

Both stateless and stateful transformations can cause repartitioning. Repartitioning is usually triggered when you change the key of a record, so any transformation that involves modifying keys can cause repartitioning. Some transformations mark the stream for repartition, which means that Kafka Streams only performs the repartition if any transformation applied after requires it.

For example, the map and flatmap transformations mark the stream for repartition because they allow you to modify the key of records. If you apply a group or join transformation after, then data is repartitioned.

The groupBy transformation always triggers repartitioning.