After completing this section, you should be able to process a basic data stream with Kafka Streams.

The era of mobile devices, edge computing, and big data has brought increasingly growing data generation rates, as well as a demand to process these large volumes promptly and efficiently. To be successful, modern applications must handle large volumes of data, while maintaining acceptable throughput levels and minimizing latency. Traditional database-centred data processing models, however, are becoming insufficient to handle these requirements.

As a result of these demands, a new data processing technique has emerged: data stream processing.

Stream processing is the action of reading data from a source as it is produced, running computations on this data, and returning the result. When compared with database-focused batch processing, stream processing presents important differences:

Operations

Because data streams are sequences, developers process them by applying well-known operations of sequential data structures, such as map, forEach, and reduce, among others.

Performance

In real-time sequence processing, high throughput and responsiveness are essential. Therefore, stream processing applications must be able to meet the following requirements:

Applications

Applications processing data streams are usually called data streaming applications or streaming applications.

If the data stream contains events occurring in an application, then this is an event streaming application. Event streaming applications are a popular choice to implement the event-driven architecture (EDA) pattern.

Apache Kafka Streams is a library for developing distributed stream processing applications. The library is part of the Apache Kafka ecosystem, and is built on top of the Kafka producer and consumer APIs, providing abstractions that represent data streams, tables and transformations. These abstractions allow developers to build stream processing models, with built-in scalability, elasticity, fault-tolerance, and security.

Some of its distinguishing characteristics are the following:

Kafka Streams applications require the following libraries from the org.apache.kafka group:

To install Kafka Streams in a Maven project, you must add the kafka-streams dependency to your pom.xml file.

<dependencies>
    ...output omitted...
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-streams</artifactId>
        <version>2.8.0</version>
    </dependency>
</dependency>

kafka-streams depends on kafka-clients, so you do not need to explicitly include the kafka-clients Maven dependency.

[user@host ~] mvn quarkus:add-extension -Dextensions="kafka-streams"

<dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-kafka-streams</artifactId>
</dependency>

Kafka Streams applications are organized according to the following logical steps:

The following is an example of a very basic Kafka Streams application, illustrating the preceding steps:

...output omitted...

import java.util.Properties;

import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.Produced;

...output omitted...

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "your-application");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-server-host:kafka-server-port");
...more props omitted...
StreamsConfig config = new StreamsConfig(settings); 

StreamsBuilder builder = new StreamsBuilder(); 

builder 
    .stream(
        "topic-name",
        Consumed.with(keySerde, valueSerde)
    ) 
    .foreach((key, value) ->
        System.out.println("Received payment: " + value)
    ) 
    .to(
        "output-topic",
        Produced.with(keySerde, valueSerde)
    ); 

Topology topology = builder.build(); 

KafkaStreams streams = new KafkaStreams(topology, config); 

streams.start(); 

The Kafka Streams DSL offers a variety of abstractions to represent data streams and tables, as well as different operations. These abstractions and operations are covered later in the course.

@Produces
public Topology buildTopology() {

    StreamsBuilder builder = new StreamsBuilder();

    builder
        .stream(
            "input-topic",
            Consumed.with(keySerde, valueSerde)
        )
        .filter((key, value) -> value > 0)
        .to(
            "output-topic",
            Produced.with(keySerde, valueSerde)
        );

    Topology topology = builder.build();
    return topology;
}

quarkus.kafka-streams.application-id=your-app-id
quarkus.kafka-streams.bootstrap-servers = kafka-server-host:kafka-server-port
quarkus.kafka-streams.topics = input-topic,output-topic

...output omitted...