Kafka Streams and the EIP Aggregator: Finishing Is Harder Than Grouping

The EIP Aggregator pattern sounds simple at first.

Collect related messages and produce one combined message.

With Kafka Streams, it can look even simpler.

stream
  .groupByKey()
  .aggregate(...)

But in real systems, the difficult part is rarely groupByKey() or aggregate().

The hard questions are these:

Which messages belong to the same group?
When is the group complete?
What should happen to late messages?
Should a message that arrives after completion be ignored or corrected?

An Aggregator is less about “collecting messages” and more about “state with a release condition.”

2D risograph-style illustration where event shapes collect into state groups and only completed groups pass through a release threshold — The Aggregator does not end at collecting events. The center of the design is deciding which groups are complete and which groups need to keep waiting.

What the EIP Aggregator Solves

The Enterprise Integration Patterns Aggregator collects related messages and publishes one combined result message.

Imagine an order flow where these events arrive at different times:

OrderCreated
PaymentApproved
InventoryReserved
ShipmentPrepared

You may want to collect them by orderId and publish a single OrderReadyForFulfillment message.

An Aggregator needs at least four decisions:

Correlation key: which messages belong together
Aggregation strategy: how messages are combined
Completion condition: when the result is released
Message store: where incomplete groups are kept

Kafka Streams naturally supports many of these concerns.

The Kafka record key can be the correlation key.
DSL operators such as aggregate, reduce, and count can express aggregation strategies.
State stores can keep incomplete groups.
Changelog topics can restore state stores.
Windows and grace periods can model time-based release and late-event tolerance.

That makes Kafka Streams a strong tool for implementing Aggregators.

But “strong tool” does not mean “automatically easy.”

When the DSL Is Enough

The simplest Aggregator is a windowed aggregation.

For example, if you want to collect order item events per order for five minutes and emit an aggregate, Kafka Streams DSL may be enough.

KStream<String, OrderItemEvent> items = builder.stream("order-items");

items
    .selectKey((ignoredKey, event) -> event.orderId())
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeAndGrace(
        Duration.ofMinutes(5),
        Duration.ofMinutes(1)
    ))
    .aggregate(
        OrderItemAggregate::empty,
        (orderId, event, aggregate) -> aggregate.add(event),
        Materialized.as("order-item-aggregate-store")
    )
    .toStream()
    .to("order-item-aggregates");

This is exactly the kind of work Kafka Streams is good at. The Kafka Streams DSL documentation describes how the DSL can express stateful transformations such as aggregations, joins, and windowing.

But the code above does not cover the whole EIP Aggregator problem.

It says “collect events in a five-minute window.” It does not answer “has this order received every required event?”

For an order to be ready, you may need rules like these:

OrderCreated must be present.
PaymentApproved must be present.
InventoryReserved must be present.
ShipmentPrepared is required only for certain product types.
PaymentRejected should immediately release a failure result.
If InventoryReserved does not arrive within 10 minutes, release a partial result.

At that point, the main problem is not windowed aggregation. It is completion-condition design.

Finishing Is Harder Than Grouping

Aggregator completion conditions can take several forms.

First, count-based:

Release when N messages with the same correlation key are collected.

Second, event-type based:

Release when all required event types have arrived.

Third, marker-event based:

Release the current aggregate when an OrderClosed event arrives.

Fourth, timeout-based:

Release a partial result 10 minutes after the first message.

Fifth, mixed conditions:

Release immediately when all required events arrive.
Otherwise release a partial result after 10 minutes.
Release a failure result immediately if a failure event arrives.

Real systems make this more complex. Product type, customer type, region, external API response, and correction events can all change the release condition.

So an Aggregator is not just an aggregation function.

Aggregator = correlation + state + completion + release + cleanup

Aggregator lifecycle diagram where events are grouped by correlation key, stored in a state store, released through completion conditions into complete result, partial timeout, or late path, and followed by cleanup and a closed marker — An Aggregator is not just a function that collects events. It is a lifecycle that decides when a stateful group may be released and how its state is cleaned up.

Which Kafka Streams API you choose depends on how simple those five concerns are.

When Windows and Suppression Fit

If you only need to produce a result after a time window closes, the DSL can be very clean.

Kafka Streams provides suppress() to hide intermediate results and emit only the final result when a window closes. The Kafka Streams Window Final Results documentation shows how a windowed computation can suppress intermediate updates and emit a final result when the window closes.

Conceptually:

items
    .selectKey((ignoredKey, event) -> event.orderId())
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeAndGrace(
        Duration.ofMinutes(5),
        Duration.ofMinutes(1)
    ))
    .aggregate(
        OrderItemAggregate::empty,
        (orderId, event, aggregate) -> aggregate.add(event),
        Materialized.as("order-item-window-store")
    )
    .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()))
    .toStream()
    .to("order-item-final-aggregates");

This is useful. But the semantics need to be understood clearly.

suppress() does not mean “the business aggregate is complete.” It means the window has closed and Kafka Streams considers that window’s result final.

That fits Aggregators like:

five-minute order item summary
hourly sensor aggregate
daily user behavior count
session-window activity summary

It may not fit Aggregators like:

release immediately when all required event types arrive
release when a marker event arrives
release a failure result when a failure event arrives
release a partial result on timeout
send late correction events to a separate topic

For those, the Processor API is often more natural.

When the Processor API Is Needed

The Kafka Streams Processor API documentation describes how to define custom processors and connect them with state stores for more customized processing logic.

An EIP Aggregator implemented with the Processor API usually looks like this:

process(record):
  key = correlationKey(record)
  aggregate = store.get(key) or new aggregate
  aggregate.add(record)

  if aggregate.isComplete():
    forward(result)
    store.delete(key)
    closedStore.put(key, completionMarker)
  else:
    store.put(key, aggregate)

punctuate(now):
  for each aggregate in store:
    if aggregate.isExpired(now):
      forward(partialResult)
      store.delete(key)

The Java implementation is longer, but the core idea is simple:

public class OrderAggregatorProcessor
    implements Processor<String, OrderEvent, String, OrderAggregateResult> {

  private ProcessorContext<String, OrderAggregateResult> context;
  private KeyValueStore<String, OrderAggregate> store;

  @Override
  public void init(ProcessorContext<String, OrderAggregateResult> context) {
    this.context = context;
    this.store = context.getStateStore("order-aggregate-store");

    context.schedule(
        Duration.ofMinutes(1),
        PunctuationType.WALL_CLOCK_TIME,
        this::emitExpiredAggregates
    );
  }

  @Override
  public void process(Record<String, OrderEvent> record) {
    String orderId = record.value().orderId();
    OrderAggregate aggregate = store.get(orderId);

    if (aggregate == null) {
      aggregate = OrderAggregate.empty(orderId, record.timestamp());
    }

    aggregate = aggregate.add(record.value(), record.timestamp());

    if (aggregate.isComplete()) {
      context.forward(record.withKey(orderId).withValue(aggregate.toResult()));
      store.delete(orderId);
      return;
    }

    store.put(orderId, aggregate);
  }

  private void emitExpiredAggregates(long now) {
    try (KeyValueIterator<String, OrderAggregate> iterator = store.all()) {
      while (iterator.hasNext()) {
        KeyValue<String, OrderAggregate> entry = iterator.next();
        if (entry.value.isExpired(now)) {
          context.forward(new Record<>(
              entry.key,
              entry.value.toPartialResult(),
              now
          ));
          store.delete(entry.key);
        }
      }
    }
  }
}

This is only explanatory code. Production code needs careful handling for serialization, errors, iterator cost, store scan cost, output keys, tombstones, metrics, and replay behavior.

Still, the example shows the difference between the DSL and the Processor API.

The DSL expresses windowed aggregation concisely. The Processor API lets the completion condition and release strategy live directly in code.

Stream Time and Wall-Clock Time Are Different

Timeout-based Aggregators need one more important choice.

Kafka Streams distinguishes stream-time from wall-clock-time. Processor API punctuation can use either PunctuationType.STREAM_TIME or PunctuationType.WALL_CLOCK_TIME.

The Processor API documentation explains that stream-time punctuation depends on stream time advancing as records are processed. If there are no records to process, stream time may not advance.

That matters for timeout Aggregators.

stream-time timeout:
  closes old groups based on event time.
  if input stops, timeout may stop too.

wall-clock-time timeout:
  closes old groups based on real time.
  can check periodically even when no input arrives.

If the business meaning is “the event-time window has closed,” stream-time is natural. If the business meaning is “release a partial result after 10 real minutes with no more events,” wall-clock-time may be more appropriate.

Leaving this ambiguous can create systems that pass tests but time out differently in production.

State Store Is the Center of the Design

An Aggregator is a stateful pattern. Incomplete groups must be remembered somewhere.

In Kafka Streams, that place is the state store. The state store documentation explains that state stores can remember recently received records, track rolling aggregates, and support de-duplication.

From an operational perspective, a state store is not just a local cache.

You need to think about:

size of one aggregate
number of simultaneously open correlation keys
maximum retention before timeout
retention period for closed group markers
RocksDB local state size
changelog topic size
recovery time after rebalance or failure
standby replica requirements

You also need a policy for messages that arrive after completion.

If a late event starts a new aggregate after the original group was completed, the result can be wrong. After OrderReadyForFulfillment has already been emitted, a late PaymentApproved for the same orderId should not start a new order flow.

Production implementations often keep a separate completed marker store.

open aggregate store:
  groups that are not yet complete

closed aggregate marker store:
  completed correlation keys and completion timestamps

Markers do not need to live forever. But they should live at least as long as the late-event or reprocessing policy requires.

Kafka Keys and Repartitioning Are Not Minor Details

An Aggregator needs all messages for the same correlation key to be seen by the same task. In Kafka Streams, that means record keys and partitioning matter.

If the input topic key is already orderId, good. If not, selectKey() or groupBy() may cause repartitioning.

Repartitioning is not bad by itself. The problem is not noticing it.

input key: eventId
correlation key: orderId
aggregator key: orderId

Kafka Streams may create a repartition topic so it can group by orderId. That means another internal topic, another Kafka hop, and operational impact from topology and store names.

When building an Aggregator, check:

Does the input topic key already match the correlation key?
If key changes are required, is repartitioning acceptable?
Are internal topic names and state store names stable?
Does partition count match throughput and key distribution?
Can one key create a very large aggregate and become a hotspot?

The DSL is convenient, but operating Kafka Streams without understanding the internal topology and internal topics is asking for surprises.

Is the Output an Update or a Final Result?

The result of aggregate() is usually a KTable. A KTable represents the latest state. If you convert it to a stream, downstream systems may receive updates whenever the aggregate changes.

But an EIP Aggregator often wants one output:

one result message per completed group

So output semantics must be explicit.

Should intermediate state be emitted?
Should only final results be emitted?
Is a partial result a normal output?
Should timeout results and complete results share a topic?
Should late corrections go to a separate topic?

This is a contract question, not just a technical question. If downstream consumers expect one final result but receive multiple intermediate updates, they become more complicated.

On the other hand, intermediate updates can be useful when downstream systems need progress visibility.

Make the topic meaning visible in names and schemas.

order-aggregate-updates
order-aggregate-finalized
order-aggregate-timeouts
order-aggregate-corrections

That reduces consumer confusion significantly.

Exactly-Once Does Not Replace Completion Logic

Kafka Streams has processing guarantee settings. Exactly-once processing can help tie state store updates and output topic writes together more safely.

But it does not replace Aggregator completion logic.

Exactly-once helps reduce the risk of duplicated processing effects. It does not answer:

Have all required events arrived?
Should timeout release a partial result?
Should late events become corrections?
How should events after group completion be handled?

Those are business rules. Aggregator design, not Kafka Streams processing guarantee, has to answer them.

What to Watch in Operations

Broker metrics are not enough for a Kafka Streams Aggregator.

Watch signals such as:

input topic lag
output topic throughput
state store size
RocksDB memory and disk usage
changelog topic throughput and size
restore time
open aggregate count
expired aggregate count
completed aggregate count
late event count
duplicate event count
partial result count
processor error count

open aggregate count and expired aggregate count are especially important. If they keep growing beyond expectations, the completion condition may be too strict, an upstream event may be missing, or the correlation key may be wrong.

A tool like Konduo is useful when operating Kafka Streams applications because Aggregator delay is rarely just consumer lag. It may be a combined result of state store size, RocksDB I/O, JVM memory, broker changelog traffic, and input partition skew.

Tests Need Time and Reprocessing

Aggregator tests need more than the happy path.

Include cases like these:

all required events arrive in order.
all required events arrive out of order.
the same event arrives twice.
some required events never arrive and timeout fires.
a late event arrives after completion.
a failure event arrives first.
state store is restored from changelog after restart.
reprocessing does not emit duplicate final results.

Kafka Streams provides TopologyTestDriver for topology-level tests. Separating pure aggregation-function tests from topology tests makes completion logic much clearer.

For an Aggregator, the important question is less “how many did we collect?” and more “is it safe to release now?” Tests should be built around that question.

Closing

Kafka Streams is a good tool for implementing the EIP Aggregator pattern.

Kafka key-based partitioning, Streams state stores, windowed aggregations, changelogs, and the Processor API provide many of the building blocks an Aggregator needs.

But thinking about the pattern only as groupByKey().aggregate() misses the important half.

Grouping is the start.
The completion condition is the essence of the Aggregator.

If you only need simple time windows and final results, the DSL with suppress() may be enough. If you need business completion conditions, timeout, partial results, late events, duplicate handling, and closed group markers, the Processor API and explicit state-store design are often more natural.

Before choosing the API, decide these first:

correlation key
completion condition
late-event policy
state retention
output contract

Once those five are clear, the implementation becomes much less unstable. If they remain vague, the Aggregator will behave vaguely in production no matter which Kafka Streams API you choose.