DEV Community: Jan Kleine

How to Join Lists in Kafka Streams Applications

Jan Kleine — Mon, 17 Mar 2025 06:00:00 +0000

I recently joined a project where we do data processing with Kafka Streams applications and came across an interesting problem: "list joins". There was already a working solution to the problem, but I was not quite satisfied, so I dug a little deeper. Since I didn't find much on the topic online, I wanted to share my findings here.

TL;DR The final topology is discussed here and you can find the code here:

iptch / kafka-list-join-demo

A demo showing how to do a list join in a kafka streams application.

Kafka List Join Demo

This project shows two ways to perform a list join in a Kafka streams application. A list join refers to joining a record that contains a list with a KTable, such that each element in the list gets joined with the corresponding element in the KTable.

This image shows the high level idea, joining a persons address list, with the corresponding addresses:

Check out this blog post for a discussion of the approaches.

The tests should cover all relevant cases of message ordering and updates.

Building

To build and test the project, run

./gradlew clean build

View on GitHub

I assume you have some understanding of Kafka Streams, but I'll try to link to relevant documentation and resources where possible. It's worth checking out the Kafka Streams DSL Developer Guide if you are unfamiliar with Kafka Streams.

The Problem

We have a topic with records (of type Outer, e.g., a person) that contain (among other things) a list. The elements in the list reference records on another topic (of type Inner, e.g., addresses). For every element in the outer topic we want to look up the corresponding record in our inner topic, and merge them to enhance our list.

Additional Considerations

There are a few additional constraints we need to keep in mind:

Eventual Consistency: The corresponding records on the inner topic may not be available right away, that is, records may come in after our outer record.
Updates: Our outer record may be updated over time. This may include updates (additions/removals) of inner list elements, but also updates to fields other than the list.
Duplicates: In our specific use case we don't need duplicate list entries. However, all approaches discussed here can be modified to allow duplicates without too much effort.

The Existing Solution

The existing solution to the problem looks roughly like follows:

We set a timestamp on every record on our outer stream (needed later).
We flat map our outer records so we have exactly one inner list element in every flat mapped record, to differentiate the flat records we change the keys to composite keys (of the form <outer-key>$$<inner-key>).
We interpret the resulting stream as a KTable and perform a foreign key left join with the other KTable. The KTable-KTable left join (as opposed to, e.g., a KStream-KTable join) is needed to fulfill our eventual consistency constrain.
We group the the resulting records by the first part of the composite key.
Finally, we reduce the records in each group by appending the lists, but we only consider the newest timestamp we see. Whenever we see a newer timestamp we discard all older records. This ensures that we do not add stale elements to our list that may be deleted in the current list.

The code looks similar to the below, though I left some smaller details out, such as how we forward tombstones. The full code can be found here.

(outerKStream, innerKTable) -> outerKStream
    .mapValues(outerMapper)
    .flatMap(outerFlatMapper)
    .toTable(buildStore(outerSerde, "listJoinFlatStore"))
    .leftJoin(
            innerKTable,
            outer -> outer.getInnerCount() == 0 ? null : outer.getInner(0).getId(),
            outerInnerJoiner,
            buildStore(outerSerde, "listJoinJoinerStore")
    )
    .toStream()
    .groupBy(
            (key, _) -> key.split("\\$\\$")[0],
            Grouped.with(Serdes.String(), outerSerde)
    )
    .reduce(
            outerReducer,
            buildStore(outerSerde, "listJoinReducerStore")
    )
    .toStream()

Issues

While this approach works for our use case, I don't like it for two reasons:

It feels hacky: The timestamps feel a bit hacky and force us to change our data model just for this operation. While we can (and in production do) clear the timestamps after the list join is complete, we are using protobufs for our records. So the timestamp is part of the protobuf definition, even if downstream services don't need it.

It is inefficient: Whenever the original protobuf changes (be it, a change to the list, or any other field), all list elements go through the join again, followed by reducing them all (including stale old records) again. That just feels wasteful.

These issues prompted me to spend some time on the problem and see if I can come up with something better.

Improvements

We want to reduce the number of join and reduce operations and remove the need for a dedicated timestamp field.

For this we need a component that keeps track of the changes in the list of a record so that we only process relevant changes.

The List Join Pre-Processor

The only way I found to accomplish this, is to write a custom Processor using the Processor API provided by Kafka Streams.

The list join pre-processor performs the task of the flat-map operation from before, but maintains internal state¹ to remember what list elements are currently in each record. When a record gets updated, it compares the previous and new list, and (1) only issues flat mapped records for new list elements and (2) issues tombstones for elements removed from the list.

(1) ensures that we only have to join new list elements, reducing duplicate work. (2) allows us to more efficiently remove old list elements, so we don't need to rely on a hacky timestamp.

The pre-processor also always forwards a copy of the current record with an empty list. We need this later to correctly propagate changes to other fields that are not the list.

The logic of the pre-processor, which can also be found here, looks as follows:

public void process(Record<String, TOuter> record) {

    // in case of tombstone or empty list, this map is empty
    Map<String, TOuter> flatValuesMap = flatMapper.apply(record.value()).stream()
            .collect(Collectors.toMap(
                    innerIdStringExtractor,
                    Function.identity(),
                    (_, newValue) -> newValue
            ));

    Set<String> newIds = flatValuesMap.keySet();

    Set<String> oldIds = Optional.ofNullable(listStore.get(record.key()))
            .map(Set::copyOf)
            .orElse(Set.of());

    // if both new and old ids are empty we don't need to do anything and can short circuit
    if (newIds.isEmpty() && oldIds.isEmpty()) {
        return;
    }

    // forward flat mapped records for new list elements
    Sets.SetView<String> addedIds = Sets.difference(newIds, oldIds);
    for (String newId : addedIds) {
        forward(record.key(), newId, flatValuesMap.get(newId));
    }

    // send tombstones for removed list elements
    Sets.SetView<String> removedIds = Sets.difference(oldIds, newIds);
    for (String removedId : removedIds) {
        forward(record.key(), removedId, null);
    }

    // if the current list is empty delete the list from the store and send a tombstone
    // for the empty list record, otherwise save the current list and send an empty list
    // record to propagate changes to other fields
    if (newIds.isEmpty()) {
        forward(record.key(), null, null);
        listStore.put(record.key(), null);
    } else {
        forward(record.key(), null, listCleaner.apply(record.value()));
        listStore.put(record.key(), List.copyOf(newIds));
    }
}

Better Reduce Operation

Now that the flat-map and the subsequent join are improved, we can further improve the reduce operation.

Previously, the reduce operation was responsible for filtering out outdated records (based on the timestamp) and aggregating the list of inner elements. This was done on all flat-mapped values (including outdated values) and needed to happen every time anything in the original outer record changed.

Now, that we can issue tombstones for removed elements, we can transition from a KStream group-by and reduce to a KTable group-by and reduce.

The KTable reduce works a bit differently than the KStream reduce. Where the KStream reduce has a single reducer that receives the current aggregate and any new record (be it a tombstone or a normal record), the KTable reduce has an adder reducer and a remover reducer.

The adder reducer is called when ever an element is added to the KTable and receives the current aggregate and the newly added record.

The remover reducer is called when ever an element is removed from the KTable (via a tombstone) and is provided with the current aggregate and the removed record. This allows us to remove the inner element from the current aggregate, as we know exactly which one got removed.

Our adder looks as follows. Remember that our pre-processor also sends a copy of the original record with an empty list. This way we can update other fields as well.

public Outer apply(Outer currentValue, Outer newValue) {
    if (newValue.getInnerList().isEmpty()) {
        // if list is empty update all fields other than the list
        return newValue.toBuilder()
                .addAllInner(currentValue.getInnerList())
                .build();
    } else {
        // otherwise add inner item to current list
        return currentValue.toBuilder()
                .addAllInner(newValue.getInnerList())
                .build();
    }
}

The full code can also be found here.

Our remover only has to worry about removing the list elements. In our case we do it based on the inner ID.

public Outer apply(Outer currentValue, Outer oldValue) {
    // if oldValue inner list is empty, there is nothing to do
    // the adder handles empty list updates
    if (oldValue.getInnerList().isEmpty()) return currentValue;

    // remove the single inner value from the current list,
    // in this case we do it by id
    Inner innerToRemove = oldValue.getInnerList().getFirst();

    List<Inner> innerList = currentValue.getInnerList().stream()
            .filter(inner -> !inner.getId().equals(innerToRemove.getId()))
            .toList();

    return currentValue.toBuilder()
            .clearInner()
            .addAllInner(innerList)
            .build();
}

The full code can also be found here.

New Topology

Incorporating the two improvements from above, we end up with a topology which looks surprisingly similar to our initial topology.

(outerKStream, innerKtable) -> outerKStream
    .process(preProcessorSupplier)
    .toTable(buildStore(outerSerde, "listJoinFlatStore"))
    .leftJoin(
            innerKTable,
            // if the inner list is empty the foreign key extractor should
            // return null so the outer is joined with null
            outer -> outer.getInnerCount() == 0 ? null : outer.getInner(0).getId(),
            outerInnerJoiner,
            buildStore(outerSerde, "listJoinJoinerStore")
    )
    .groupBy(
            // group by first part of composite key
            (key, value) -> KeyValue.pair(key.split("\\$\\$")[0], value),
            Grouped.with(Serdes.String(), outerSerde)
    )
    .reduce(
            listAdder,
            listRemover,
            buildStore(outerSerde, "listJoinReducerStore")
    )
    .toStream()

Again, I skipped over some smaller details, like how we forward tombstones. The full topology can be found here.

Since copying this code to different topologies is a bit cumbersome, we decided to wrap this entire sub-topology in a ListJoin utility. This allows easier reusing of the code, as the developer only has to provide a couple key components and make the resulting topologies easier to understand.

// setup
ListJoin<...> listJoin = ListJoin.builder()
        // specify joiner, reducers, etc.
        .build()

// later in the topology
KStream<...> joinedKStream = listJoin.apply(myLeftKStream, myRightKTable);

Conclusion

This topology is probably still not optimal. For one, it may make sense to deduplicate the empty list records to further reduce downstream work. However, in our case this is not necessary.

In any case, this new approach is a clear improvement over the previous timestamp based approach.

Have you solved a similar, or even the same, problem before? If so, please consider leaving a comment as I'd very much like to hear what you did. Likewise, if you have found issues with the above code or have ideas for improvements, I'm keen to hear from you!

Acknowledgments

Cover image: "Close Up Photo of Blue Background" by Harrison Candlin

Jan KleineFollow

Maintaining state can be done via a state store, which is automatically backed up to a changelog topic in kafka, ensuring the processor can tolerate application scaling and recover in case of application failures. ↩

Rusty Backends

Jan Kleine — Tue, 21 Jan 2025 06:30:00 +0000

This post was written together @sekael and @zkck

Rust has been a developer favorite for many years now and is consistently highly admired by developers[1]. It made a buzz when it was introduced as a complementary language to C and Assembly in the Linux kernel with version 6.8[2] and its speed and memory safety make it a popular choice for low-level and performance-critical applications. But with the whole world using the internet, we wanted to find out whether it can also do web and challenge the reigning backend champions like Java Spring or Go.

To find answers, we wanted to get our hands dirty with three popular Rust web frameworks including rocket, axum, and actix, and get a feeling for their performance, features, and most importantly the developer experience.

In this post, we will examine each of these frameworks by implementing the same example API endpoints in each one, connect to a MongoDB database, and run the server in a Docker container.

All three, rocket, axum, and actix, cover the full range of functionality you would expect from a web framework, like routes, handlers, request and response parsing, middleware, state management, database interactions, logging, and testing. Furthermore, all of these frameworks perform well and will meet and surpass the needs of most web applications, so the choice really comes down to ergonomics and how it feels to develop within each framework.

So let’s jump right in…

Rocket (rocket.rs)

rwf2 / Rocket

A web framework for Rust.

Rocket

Rocket is an async web framework for Rust with a focus on usability, security extensibility, and speed.

#[macro_use] extern crate rocket;

#[get("/<name>/<age>")]
fn hello(name: &str, age: u8) -> String {
    format!("Hello, {} year old named {}!", age, name)
}

#[launch]
fn rocket() -> _ {
    rocket::build().mount("/hello", routes![hello])
}

Visiting localhost:8000/hello/John/58, for example, will trigger the hello route resulting in the string Hello, 58 year old named John! being sent to the browser. If an <age> string was passed in that can't be parsed as a u8, the route won't get called, resulting in a 404 error.

Documentation

Rocket is extensively documented:

Overview: A brief…

View on GitHub

Rocket is a web framework for Rust that comes with the most batteries included out of the three we looked at. It positions itself in the same league as Rails (Ruby) or Flask (Python) and aims to offer functionality for anything a web backend might need, including route handling, request and response parsing, database integrations, validation, and so on. One of its main goals is to minimize the amount of boiler plate code you will have to write, and it does so through metaprogramming, i.e. heavy use of Rust macros.

This makes it easy for the developer to integrate her actual logic with the rocket framework. Let’s have a quick look on how you might define a route handler including parameter guards and validation, a database connection pool, and a JSON object.

We can define a route handler simply by adding the appropriate macro to a function.

#[get("/texts/<uuid>")]
pub async fn get_text(db: Connection<TextsDatabase>, uuid: Uuid) -> (Status, Value) {
    // ...
    (Status::Ok, json!({"data": data}))
}

If the request parameter cannot be parsed, the appropriate response code is sent instead of calling the function, so we have the usual comfort of type safety. Request bodies are handled very similarly, and of course, parsing integrates seamlessly with serde.

#[derive(Deserialize)]
pub struct Message<'m> {
    pub data: &'m str,
}

#[post("/texts", format = "application/json", data = "<msg>")]
pub async fn post_text(db: Connection<TextsDatabase>, msg: Json<Message<'_>>) -> (Status, Value) {
    // ...
}

Adding a database is as simple as annotating the appropriate struct and initializing it at startup. Of course many popular DBMS's are supported.

#[derive(Database)]
#[database("texts")]
pub struct TextsDatabase(mongodb::Client);

Rocket goes heavy on the nerdy space vocabulary - which adds to its charm. Thus, to get your web server off the ground, you simply launch it. Notice again, how annotations are used to get the rocket going.

#[launch]
fn rocket() -> _ {
    rocket::build()
        .attach(TextsDatabase::init())
        .mount("/", routes![get_text, post_text])
}

We have found rocket very enjoyable to work with, especially if you want to focus on the business logic of your server and feel safe trusting the framework to take handling of the boilerplate code off your hands. If you are rather new to Rust and maybe switching over from other frameworks that make heavy use of annotations, you will feel at home with Rocket.

Axum

tokio-rs / axum

Ergonomic and modular web framework built with Tokio, Tower, and Hyper

axum

axum is a web application framework that focuses on ergonomics and modularity.

More information about this crate can be found in the crate documentation.

High level features

Route requests to handlers with a macro free API.
Declaratively parse requests using extractors.
Simple and predictable error handling model.
Generate responses with minimal boilerplate.
Take full advantage of the tower and tower-http ecosystem of middleware, services, and utilities.

In particular the last point is what sets axum apart from other frameworks axum doesn't have its own middleware system but instead uses tower::Service. This means axum gets timeouts, tracing, compression, authorization, and more, for free. It also enables you to share middleware with applications written using hyper or tonic.

Usage example

use axum::{
    routing::{get, post},
    http::StatusCode,
    Json, Router,
};
use serde::{Deserialize,

…

View on GitHub

If you are looking for a framework that is more bare-metal Rust and comes with less batteries included, you may want to check out the axum framework developed by tokio-rs, the same team that is responsible for the tokio async library. Opting for axum means you will have to take care of some more of the boilerplate code but you are also not fighting a potentially opinionated framework to get your logic just the way you want it. Compared to rocket, axum does not depend heavily on macros or annotations, and it sets itself apart from its peers by not implementing its own middleware but rather relying on tower for this. Through the tower ecosystem axum can offer timeouts, tracing, compression, authorization, and much more, while also enabling you to share your middleware with applications written with other web libraries like hyper.

Looking at axum’s ergonomics, previous Go developers will feel right at home. The syntax looks very similar to a web server written e.g. with the gorilla/mux module in Go. So how would you go about the implementation of your first route handler including connecting to a database and starting the server? Let’s have a look.

The same two functions from above look something like this:

async fn post_text(
    State(state): State<Arc<state::MongoAppState>>,
    Json(text_payload): Json<payloads::TextPayload>,
) -> Result<
    (StatusCode, Json<payloads::InsertedResponse>),
    (StatusCode, Json<payloads::ErrorResponse>),
> {
    // ...
}

async fn get_text(
    State(state): State<Arc<state::MongoAppState>>,
    Path(text_id): Path<String>,
) -> Result<Json<payloads::TextPayload>, (StatusCode, Json<payloads::ErrorResponse>)> {
    // ...
}

No macros, but a lot more verbose. We see the same when we look at the startup code, we have to do a lot more manually, e.g., pass the DB connection along.

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let client = mongodb::Client::with_uri_str("mongodb://...").await.unwrap();

    let shared_state = std::sync::Arc::new(state::MongoAppState::new(client));

    // build our application with a single route
    let app = Router::new()
        .route("/texts", post(post_text))
        .route("/texts/:text_id", get(get_text).delete(delete_text))
        .with_state(shared_state);

    // run our app with hyper, listening globally on port 3000
    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await.unwrap();
    axum::serve(listener, app).await.unwrap();
    Ok(())
}

Axum is a very powerful web framework that benefits from the expertise of the tokio-rs team in creating great Rust tooling. It is being very actively developed and relies on other state-of-the-art crates for middleware. You will most likely enjoy axum the most if you are already familiar with Rust, enjoy having control over a lot of detail in implementation, or maybe if you are switching from another low-level web language like Go.

Actix (actix.rs)

actix / actix-web

Actix Web is a powerful, pragmatic, and extremely fast web framework for Rust.

Actix Web

Actix Web is a powerful, pragmatic, and extremely fast web framework for Rust

Features

Supports HTTP/1.x and HTTP/2
Streaming and pipelining
Powerful request routing with optional macros
Full Tokio compatibility
Keep-alive and slow requests handling
Client/server WebSockets support
Transparent content compression/decompression (br, gzip, deflate, zstd)
Multipart streams
Static assets
SSL support using OpenSSL or Rustls
Middlewares (Logger, Session, CORS, etc)
Integrates with the awc HTTP client
Runs on stable Rust 1.72+

Documentation

Example

Dependencies:

[dependencies]
actix-web = "4"

Code:

use actix_web::{get, web, App, HttpServer, Responder};
#[get("/hello/{name}")]
async fn greet(name: web::Path<String>) -> impl Responder {
    format!("Hello {name}!")
}

#[actix_web::main

…

View on GitHub

Last but not least we want to take a look at actix-web. It combines aspects from both previous frameworks, rocket as well as axum, which is resembled in its ergonomics. Annotations are back on the menu and macros are more heavily used, especially in the definition of route handlers. When implementing middleware or database connections, however, you will not find the same level of abstraction as you might with rocket. Actix is extremely fast 3 and it aims to cater to both experienced Rust developers as well as newcomers who are just starting with Rust development. It ships with its own middleware, e.g. for logging, session management, or cross-origin resource sharing, and allows you to expand the framework with your own middleware that can hook into actix. Let’s have a look at how route handlers, database connections, and starting the server are handled in actix.

#[post("/texts")]
async fn post_text(client: web::Data<Client>, payload: web::Json<TextResponse>) -> impl Responder {
    // ...
}

#[get("/texts/{uuid}")]
async fn get_text(client: web::Data<Client>, uuid: web::Path<Uuid>) -> impl Responder {
    // ...
}

Again, we have a lot less boilerplate. Though launching the server falls in between the first two examples with regards to verbosity.

#[actix_web::main] // or #[tokio::main]
async fn main() -> std::io::Result<()> {
    let db_client = Client::with_uri_str("mongodb://....").await.expect("failed to connect");

    HttpServer::new(move || {
        App::new()
            .wrap(Logger::default())
            .app_data(web::Data::new(db_client.clone()))
            .service(post_text)
            .service(get_text)
    })
    .bind(("0.0.0.0", 8080))?
    .run()
    .await
}

The benchmarks speak for themselves, actix is indeed blazingly fast! And that does not come at the cost of developer experience, which actix manages to keep to a very high standard. It ships with more batteries included than axum, but does not abstract everything quite as much as rocket. If you already feel comfortable writing programs in Rust and other low-level web programming languages, you will enjoy developing in actix, and your user will enjoy the performance!

Acknolegements

Selim wrote the initial draft of this post and Selim, Zak, and I each implemented the sample API using one of the above frameworks.

Cover image: "Brown Chains" by Miguel Á. Padriñán

Selim KaelinFollow

I am a mountain athlete at heart, enjoy learning about and writing code, and passionate about Rust, Cloud, and maps!

DEV Community: Jan Kleine

How to Join Lists in Kafka Streams Applications

iptch / kafka-list-join-demo

A demo showing how to do a list join in a kafka streams application.

Kafka List Join Demo

Building

The Problem

Additional Considerations

The Existing Solution

Issues

Improvements

The List Join Pre-Processor

Better Reduce Operation

New Topology

Conclusion

Acknowledgments

Jan KleineFollow

Rusty Backends

Rocket (rocket.rs)

rwf2 / Rocket

A web framework for Rust.

Rocket

Documentation

Axum

tokio-rs / axum

Ergonomic and modular web framework built with Tokio, Tower, and Hyper

axum

High level features

Usage example

Actix (actix.rs)

actix / actix-web

Actix Web is a powerful, pragmatic, and extremely fast web framework for Rust.

Actix Web

Features

Documentation

Example

Acknolegements

Selim KaelinFollow

Zak CookFollow

Jan KleineFollow