Duplicate event rows observed during a Kafka consumer-group rebalance with multiple workers

Start with packages/core/src/modules/ingest/src/consumer.ts, especially the message processing, heartbeat, and offset resolution flow around the cited lines. Trace how rebalance or heartbeat failures interact with downstream writes and pending handlers; the issue’s reproduction and SQL query provide a way to compare event rows with unique input sequences. Done means clarifying the delivery contract and reliably preventing duplicate analytics effects without losing durable events, with a rebalance integration test.

Summary

On main 7f6f7f9e2b978bd0472807f586d65928f793d08c (including #560), replaying a fixed Kafka backlog into empty Redis and ClickHouse produced 108,327 business-event rows for 100,000 unique input events. Worker logs report a consumer-group rebalance during startup. Kafka lag and all buffers eventually reached zero.

This was not an intentional offset reset against an already-populated sink: each run used a fresh consumer group, empty Redis and empty ClickHouse. The API and load generator were stopped during replay. Rebalancing itself is expected; the concern is duplicate application-visible events following redelivery.

Environment

• OpenPanel revision: 7f6f7f9e2b978bd0472807f586d65928f793d08c.
• KafkaJS 2.2.4, Bun runtime (application image built locally from that revision).
• Redpanda v25.1.9, 25 topic partitions, replication factor 1, 1 broker CPU / 1 GB broker memory configuration.
• ClickHouse 25.10.2.65, Redis 7.2.5.
• Shared local VM: 6 vCPU / 24 GB; each worker limited to 2 CPU / 4 GB; ClickHouse limited to 2 CPU / 8 GB.
• Two worker containers launched via one Compose up command; their startup log timestamps differ by approximately one second.
• group_initial_rebalance_delay=0 in the failing run.

Buffer settings:

EVENT_BUFFER_BATCH_SIZE=10000
EVENT_BUFFER_CHUNK_SIZE=10000
SESSION_BUFFER_BATCH_SIZE=20000
# SESSION_BUFFER_CHUNK_SIZE unset
PROFILE_BUFFER_BATCH_SIZE=500
PROFILE_BUFFER_CHUNK_SIZE=1000
CRON_CONCURRENCY=6
BUFFER_CH_INSERT_CONCURRENCY=5

Schema disclosure: these replay tests used a test-only correction to the sessions sorting key, adding id to avoid a separately reproduced cross-session collapsing problem. The events table and application ingestion code were unchanged. This issue concerns duplicate rows in events, not that separate session issue.

Reproduction scenario

1. Prepare an isolated Kafka topic containing 100,000 tracked events. Use 10,000 distinct device/profile identities, 10 events per identity, and a globally unique properties.seq from 0 to 99,999. The fixture was originally accepted through the API and its initial completed ingestion had 100,000 rows and 100,000 distinct sequences.
2. Stop the API, producer/load generator and other consumers of the test group. Verify topic offsets: all 25 partitions start at 0 and the sum of high watermarks is 100,000.
3. Use a new consumer group and empty Redis/ClickHouse. Let the group start at the oldest messages. Set Redpanda group_initial_rebalance_delay=0 for the reproduction.
4. Start two workers against this backlog. The first can begin processing before the second finishes joining. Capture the group rebalance logs. A controlled follow-up can explicitly add a second worker while the first processes an uncommitted batch; that variation has not yet been tested here.
5. Wait for Kafka lag and event/session/profile buffers to reach zero, then count only the fixture's business-event name (exclude generated session_start).

SELECT
    count() AS event_rows,
    uniqExact(properties['seq']) AS unique_input_events
FROM openpanel.events
WHERE name = 'retest20_keyfix100k_chunk10k';

Observations

Run	Workers	Initial rebalance delay	Business rows	Unique input sequences	Final lag/buffers
Cold-start replay	2	0 ms	108327	100000	All zero
Control	2	5000 ms	100000	100000	All zero
Control	3	5000 ms	100000	100000	All zero

The failing run has 8,327 excess rows (8.327%). Both controls also passed session/profile reconciliation. These are individual runs, not a measured statistical reproduction rate. A five-second initial delay reduced startup overlap in these controls; it is not a general fix for later scaling, restarts, broker failover or crashes.

Representative KafkaJS logs from the failing run:

2026-10-20T03:03:37.586Z [Connection] Response Heartbeat(key: 12, version: 3)
error: The group is rebalancing, so a rejoin is needed
...
2026-10-20T03:03:43.074Z [Runner] The group is rebalancing, re-joining
groupId: oprt-replay100k_w2-20261020

Relevant code / working hypothesis

consumer.ts at the tested revision:

await processMessage(m, batch.partition);
processed.add(m.offset);
processedCount += 1;
if (processedCount % HEARTBEAT_EVERY === 0) {
  await heartbeat();
}
// After all per-key groups finish:
if (await bufferedEventsAreDurable(batch.partition, closeDurabilityWindow)) {
  for (const m of batch.messages) {
    if (!processed.has(m.offset)) break;
    resolveOffset(m.offset);
  }
}

processMessage can produce downstream effects before the batch's offsets are resolved/committed. If a heartbeat rejects because of a rebalance, or ownership changes before offsets are committed, Kafka can redeliver messages whose effects already reached Redis/ClickHouse. The per-key handlers use Promise.all; a rejection also deserves review for other handlers that may still be running.

This is a working hypothesis based on code and correlated logs, not a proven message-by-message trace. We have not yet correlated every duplicate with its original Kafka partition/offset. The source explicitly prefers replay over loss when durability cannot be confirmed; that is sensible at-least-once behavior, but duplicate events affect analytics unless downstream handling is idempotent.

Questions / suggested follow-up

• Is at-least-once delivery with application-visible duplicate event counts the intended contract during rebalances? If so, could the limitations and recommended deduplication strategy be documented?
• Could an integration test trigger a rebalance after downstream side effects but before offset commit, checking for both loss and duplicate analytics effects?
• Can event ingestion and session/profile side effects be made idempotent under replay, using a stable event identifier or topic/partition/offset identity?
• Please avoid committing offsets before durable effects, or adding a standalone dedup marker before writes: either can introduce data loss. Any deduplication mechanism needs crash-safe coordination with the effects it protects.

No production consumer offsets were reset. No proposed code fix is claimed here. The session sorting-key issue and an earlier independent EPIPE observation should be tracked separately rather than assumed to have this same cause.