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

The org.apache.kafka.common.serialization package provides concrete serializers for primitive and basic types, such as IntegerSerde or StringSerde.

There are multiple ways to (de)serialize complex objects, such as the following methods:

The KStream interface represents a data stream. Use this object to read data from a Kafka topic as a stream, to apply transformations to the stream, and to send the result to another Kafka topic.

To create a new KStream object that reads from a Kafka topic, call the StreamsBuilder#stream method. Specify the topic name, and the key and value serdes.

import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.common.serialization.Serdes;
import io.quarkus.kafka.client.serialization.ObjectMapperSerde;

// ...class omitted...

Serde<String> keySerde = Serdes.String();
ObjectMapperSerde<Price> valueSerde = new ObjectMapperSerde<>(Price.class);

KStream prices = builder.stream(
    "prices",
    Consumed.with(keySerde, valueSerde)
);

The resulting KStream instance creates a source processor in the application topology graph. When reading a topic, Kafka Streams distributes the Kafka topic partitions as stream partitions across your application replicas to achieve scalability. Consequently, a KStream object running on a particular application instance only keeps track of data from one or more stream partitions.

After you have created a stream, you can start specifying stream transformations.

KTable objects allow you to interpret a data stream as the change log of a table. From this perspective, each record read from the stream represents a change in the table. Kafka Streams interprets the stream as follows:

Tables in Kafka Streams are particularly useful for stateful operations, for example:

KTable objects manage an internal state store. Consequently, every node of a Kafka Streams application that uses a table contains a local state store. The same applies to GlobalKTable objects, which are covered later in this section.

With other messaging solutions, you might have to use an external database, or additional logic, to implement the preceding examples. With a KTable object, you can easily perform such operations in a streaming fashion, and benefit from built-in fault-tolerance capabilities. These state stores are fault-tolerant, because Kafka Streams replicates them into Kafka topics.

To read a Kafka topic as a table, use the StreamsBuilder#table method. Specify the topic name, the key and value serdes, and, optionally, the name of the table store.

import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.Materialized;

// ...class and serdes omitted...

KTable pricesTable = builder.table(
    "prices",
    Consumed.with(keySerde, valueSerde),
    Materialized.as("pricesTableStore")
        .withKeySerde(keySerde)
        .withValueSerde(valueSerde)
)

The resulting KTable instance creates a source processor in the application topology graph. Kafka Streams also distributes partitions when reading a topic into a table. Consequently, a KTable object running on a particular application instance only keeps track of a subset of all the stream partitions.

The third argument specifies the name of the table store and the custom serdes configurations to materialize the table to the store. You must use this argument if you intend to use the interactive queries feature to query the table. If you do not specify a store name, then Kafka Streams assigns an internal, non-queryable store to the table. Interactive queries are covered later in the course.

Sometimes, you need every node of your application to have a full copy of the data, instead of a subset of the partitions. For example, you might want to do the following:

For these cases, Kafka Streams provides the GlobalKTable interface. Unlike KTable, which partitions the stream across the application instances, GlobalKTable objects are not partitioned.

For example, assume that your application reads a topic as a table change log. The topic uses three partitions and the application runs on three nodes.

Loading the full set of partitions on each node allows you to join records by non-key values, and makes join co-partitioning unnecessary. Joins are covered later in the course.

To create a new GlobalKTable object, use the StreamsBuilder#globalTable method.

import org.apache.kafka.streams.kstream.GlobalKTable;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.Materialized;

// ...class and serdes omitted...

GlobalKTable pricesTable = builder.globalTable(
    "prices",
    Consumed.with(keySerde, valueSerde),
    Materialized.as("pricesGlobalTableStore")
        .withKeySerde(keySerde)
        .withValueSerde(valueSerde)
)

You can either group a stream by the Kafka topic key or a new key.

Both operations return a KGroupedStream. For example:

import org.apache.kafka.streams.kstream.KGroupedStream;
import org.apache.kafka.streams.kstream.Grouped;

// ...class and serdes omitted...

KGroupedStream<String, String> purchasesByProduct = stream.groupBy(
    (key, value) -> value.productId, 
    Grouped.with( 
      keySerde,
      valueSerde
    )
);

Grouping a stream might cause data repartition. In particular, groupBy always repartitions data, due to the creation of a grouped stream with a new key.

After grouping the stream, you must aggregate the data to create a KTable object. The KGroupedStream interface provides the following aggregation methods:

These aggregations return a KTable object. You can also use them in a windowed form, so the aggregation only computes records within a given time period.

A basic example of grouping and aggregating a stream into a table displays as follows:

import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Grouped;

// ...class and serdes omitted...

KTable<Integer, Long> positionsReportedByVehicleType = vehiclePositions 
    .groupBy(
        (key, value) -> value.vehicleType, 
        Grouped.with(Serdes.Integer(), vehicleSerde) 
    )
    .count();