Consolidating a noisy Kafka topic with a debounced aggregator

At Hasaki, when a warehouse doesn’t have enough stock to fulfill an order, the system creates an internal transfer request to move inventory from another location. The existing flow publishes one Kafka message per order, and each message triggers a separate transfer request.

During peak hours, a single warehouse can generate hundreds of these messages in minutes. Each message creates a small transfer request. The warehouse team ends up processing dozens of small requests for the same destination when a single consolidated batch would do.

Before:  Order A001 → transfer #1 for WH-42    (100 transfers in 5 min)
         Order A002 → transfer #2 for WH-42
         Order A003 → transfer #3 for WH-42
         ...

After:   WH-42 (1 consolidated batch)         → downstream

We addressed this with a debounced aggregator between the per-order topic and downstream systems. It buffers messages by warehouse and publishes consolidated batches when the timing conditions are met.

Problem and architecture⌗

The problem⌗

Each order-level event immediately creates work for the warehouse team. When messages arrive in bursts, the team has to process many small transfer requests for the same destination. Downstream systems also face high processing overhead and unpredictable load spikes.

We want to preserve correctness while grouping events into batches by warehouse, so downstream systems receive more predictable traffic.

Overall flow⌗

The aggregator runs as a separate service that consumes the existing topic, so upstream producers don’t need to change.

The service runs in two phases:

1. In the ingest phase, the service receives a Kafka message and uses an UPSERT to write it to the MySQL buffer with status PENDING.
2. In the flush phase, a poller claims ready rows, aggregates them, sends one batch message, and marks the rows SENT.

Kafka remains the source of truth. The MySQL table holds derived data as a buffer so we can control how batches are built.

Ingest phase⌗

We keep ingest simple: each (partition_key, location_id, order_id) has one row. If the order appears again while the row is still pending, we update its payload and timestamp.

INSERT INTO event_buffer_table
  (partition_key, location_id, order_id, payload, status)
VALUES (?, ?, ?, ?, 'PENDING')
ON DUPLICATE KEY UPDATE
  payload = IF(status = 'PENDING', VALUES(payload), payload),
  updated_at = IF(status = 'PENDING', NOW(), updated_at);

The service commits offsets manually (enable.auto.commit: false), only after the database write succeeds. If the service crashes before committing the offset, Kafka replays the message and we safely reprocess it.

Flush phase⌗

A poller, for example one running every 30 seconds, flushes the buffered data for a warehouse when either condition is true:

• Debounce idle: No new update has arrived for 5 minutes.
• Hard window: The oldest pending row is 30 minutes old.

The flush runs in two stages:

1. Claim rows and assign a new batch_id, limiting the number of rows to max_batch_size.
2. Send the claimed batches, then mark their rows as SENT.

-- Step 1. Claim phase: find flushable warehouse buckets (PENDING only)
SELECT location_id
FROM event_buffer_table
WHERE status = 'PENDING'
GROUP BY location_id
HAVING MAX(updated_at) < NOW() - INTERVAL 5 MINUTE
    OR MIN(created_at) < NOW() - INTERVAL 30 MINUTE;

-- Step 2. Claim phase: assign immutable batch_id for one bucket
SET @batch_id := UUID();
UPDATE event_buffer_table
SET status = 'CLAIMED', batch_id = @batch_id, claimed_at = NOW()
WHERE status = 'PENDING' AND location_id = ?
ORDER BY created_at ASC
LIMIT 500;

-- Step 3. Flush phase (separate worker/tick): load all claimed rows
SELECT batch_id, payload
FROM event_buffer_table
WHERE status = 'CLAIMED'
ORDER BY claimed_at ASC, batch_id ASC;

-- Step 4. Process claimed batches in application code, then mark rows as SENT.

Failure modes and how we handle them⌗

Ingest failures⌗

What if the Kafka offset is committed before the database write is durable?⌗

If the service crashes after committing the offset but before the database commits the write, the message is skipped during processing. We avoid this by committing offsets manually, only after the database write succeeds.

If the service crashes before committing the offset, Kafka replays the message. If it crashes afterward, the row is already durably stored. The tradeoff is that end-to-end processing latency depends on database write latency.

Flush failures⌗

What if too many messages arrive in 30 minutes and the process runs out of memory?⌗

If we rely only on a hard 30-minute window, the buffered data for one warehouse can grow too large. If the process dies, recovery can become slow and create load spikes.

We use debounce idle to flush earlier after a 5-minute quiet period and limit the number of rows in each claim with max_batch_size.

This splits the work across smaller batches, keeps latency bounded, and reduces memory pressure. More frequent flushes can slightly reduce how much data we consolidate into each batch.

What if messages keep arriving for a warehouse and debounce never triggers?⌗

Even a small but continuous stream of messages keeps changing updated_at, so the idle condition never becomes true. Rows can stay pending indefinitely.

We keep a hard 30-minute window measured from the oldest pending row.

This ensures that each warehouse’s data keeps moving through the system even without quiet periods. Debounce optimizes batching when there are quiet periods, while the hard window ensures processing continues when messages keep arriving. This property is called liveness. The tradeoff is that some flushes happen before an ideal batch size is reached.

What if sending succeeds but the service crashes before marking the batch as sent, then restarts?⌗

Without a stable batch identity, retries can regroup orders inconsistently, such as A with B on the first attempt and A with C on the next. This creates consistency problems downstream.

We assign batch_id during the claim phase and keep it unchanged until the status is SENT. On retry, we resend the same claimed rows with the same batch_id.

Downstream consumers use batch_id as an idempotency key to deduplicate messages. This preserves at-least-once delivery while avoiding repeated business operations, but it requires downstream consumers to deduplicate by batch_id.

What if flush queries become slow under heavy load?⌗

Flushing large amounts of data without a limit increases lock duration, query latency, and the risk of unusually slow processing.

We separate claiming and flushing into two stages, limit each claim with max_batch_size, and retry from the CLAIMED state. If Kafka is temporarily unavailable after a claim, the rows remain CLAIMED and are picked up on the next tick.

Each tick handles a limited amount of work, and claimed batches that haven’t been sent are retried. A warehouse may need multiple ticks before all its rows are processed.

What if flush jobs run at the same time in a multi-pod deployment?⌗

Two pods can claim and flush concurrently, leaving claim ordering, contention, and retry timing undefined. Higher isolation levels such as repeatable read can reduce anomalies, but they don’t explicitly enforce a single runner across the deployment.

Before each flush tick, every pod tries to acquire a Redis lock. Only the lock holder runs the job; the other pods skip that tick. The lock TTL is 60 seconds and the job timeout is 54 seconds (90% of the TTL). The six-second gap accounts for clock differences, scheduling delays, and network latency.

The lock enforces one active flush worker at a time across pods and allows automatic recovery if a pod crashes. The timeout must be shorter than the TTL. If both were 60 seconds, even 10 ms of drift could let the lock expire before the first job stops. Another pod could then acquire the lock and start a second flush.

The 90% cutoff is a heuristic, not a guarantee. A long process pause or network delay can still exceed the margin. Flush cadence also depends on Redis availability.

What if the buffer table keeps growing forever?⌗

Naively using bulk DELETE to remove old rows causes heavy InnoDB fragmentation and reduces throughput.

We use monthly range partitions in MySQL and drop old partitions on a schedule, after checking that they contain no unsent rows.

Dropping a partition is a metadata operation, so it avoids deleting large amounts of data row by row. The tradeoff is that partition management requires operational discipline and safety checks.

Downstream contract⌗

Downstream systems consume one consolidated message per warehouse flush:

• key: location_id
• identity: batch_id
• payload: order_ids, aggregated items, and flushed_at

Downstream consumers must deduplicate by batch_id to keep processing idempotent.