Too many parts from single-row inserts, plus a lagging Kafka consumer

Part count, not data volume, is the failure. Each single-row insert creates one part in the partition it targets, on the shard that receives it, and replication then creates that part on the second replica. At 2,500 inserts/second you are producing thousands of parts per second cluster-wide; background merges schedule far fewer than that, so the parts-per-partition count climbs past parts_to_delay_insert (150 by default) and then parts_to_throw_insert (300) - which is exactly the Too many parts (312) you are seeing. The tight retry loop makes it worse: each retry is another insert attempt against an already swollen partition.

Fixes in order of impact:

1. Batch client-side. Buffer events and flush every 5,000-50,000 rows or every second. That turns 2,500 inserts/second into a handful and fixes the ratio at the source. Nothing else is as effective.
2. async_insert = 1 with wait_for_async_insert = 1 if the client cannot ship soon. The server buffers rows and forms a larger part when async_insert_max_data_size or the busy timeout is reached. Keep wait_for_async_insert = 1 so insert errors are still reported to the caller instead of vanishing into a buffer.
3. Backoff with jitter and a retry cap. A hot retry loop on a throttled table is self-inflicted load.
4. Revisit daily partitions. Daily partitioning is fine for retention, but during backfills inserts touch many partitions and multiply the part count; monthly partitions reduce that pressure unless you need per-day DROP PARTITION.

Monitor with system.parts (parts per active partition), system.merges, and the parts_to_delay_insert/parts_to_throw_insert thresholds on the actual table.

A Kafka engine table with kafka_num_consumers = 1 gives you a single consumer thread pulling and flushing, so 11 of the 12 partitions have nobody assigned and throughput is capped at roughly one partition's worth. Set kafka_num_consumers = 12 - more consumers than partitions is wasted, and each consumer maps to a partition, so matching the count is the natural ceiling (bounded by available CPU). Consider kafka_thread_per_consumer = 1 so each consumer flushes into the destination independently rather than serialising through one thread.

Then look at the destination, because lag often is not a Kafka problem. The materialized view inserts one block per poll; if the target ReplicatedMergeTree has an expensive sorting key or is itself suffering from part pressure, flushes stall and back-pressure surfaces as consumer lag. kafka_max_block_size (65,536 rows by default) and kafka_poll_max_batch_size set the batch shape: larger blocks amortise insert overhead but increase per-flush latency.

Diagnosis: system.kafka_consumers shows per-consumer assignment, committed offsets and any exception; system.parts and system.merges on the destination show whether inserts are keeping up; system.query_log on the insert path shows block sizes and durations. If consumers match partitions and the destination is healthy, the remaining suspect is downstream merge capacity - the same class of problem as the Too many parts incident, arriving by a different route.

The Kafka engine delivers at-least-once. Offsets are committed after a block is written to the destination table, so an interruption between the flush and the commit - precisely what a broker restart or a consumer rebalance causes - means the same block is polled again on restart and inserted a second time. A fraction of a percent of a day's traffic is consistent with one bad restart.

Options, ranked by how much they actually buy you:

1. Make the destination idempotent. Carry _partition and _offset from the Kafka engine into the row, and order a ReplacingMergeTree by (topic, partition, offset) - or by event id with offset as the version. Replays then collapse at merge and at FINAL, and you stop depending on restart timing. This is the only approach that is correct regardless of where the interruption lands.
2. insert_deduplication_token. Reliable only if the token is stable across the retry. With Kafka blocks it usually is not, because a re-poll can produce a different block boundary, so it is a partial mitigation rather than a guarantee.
3. insert_deduplicate = 1 on the Replicated table. Drops byte-identical replayed blocks, but only inside the dedup window and only for identical blocks - the same caveat as above.

For this volume, take option 1 and pair it with reconciliation: compare SELECT count() FROM events FINAL against the topic's end-offset sum per partition after a restart drill. Test the drill deliberately - restart a broker during peak ingest and confirm the reconciled counts still match.