This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 6ce2682d7bb KAFKA-20539 Prevent Hot Data Loss on Partition Expansion
for Latest Policy (#22784)
6ce2682d7bb is described below
commit 6ce2682d7bb860bd3b875404fb46e2cbf4e6b96d
Author: Ken Huang <[email protected]>
AuthorDate: Tue Jul 28 22:25:23 2026 +0800
KAFKA-20539 Prevent Hot Data Loss on Partition Expansion for Latest Policy
(#22784)
Enhanced the auto.offset.reset documentation in ConsumerConfig.java to
clearly explain the message-loss risk when using latest after increasing
the partition count, and recommended by_duration:<duration> as a safer
alternative.
Updated `basic-kafka-operations.md` with a recommendation section and a
comparison table for suitable `<duration>` values under both the
consumer (KIP-848) and classic group protocols.
Reviewers: Chia-Ping Tsai <[email protected]>
---
.../apache/kafka/clients/consumer/ConsumerConfig.java | 18 ++++++++++++++++--
docs/operations/basic-kafka-operations.md | 12 ++++++++++++
2 files changed, 28 insertions(+), 2 deletions(-)
diff --git
a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
index 3a4e6ceee18..217e37219c2 100644
---
a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
+++
b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
@@ -183,8 +183,22 @@ public class ConsumerConfig extends AbstractConfig {
"Negative duration is not allowed.</li>" +
"<li>none: throw exception to the consumer if no previous offset
is found for the consumer's group</li>" +
"<li>anything else: throw exception to the consumer.</li></ul>" +
- "<p>Note that altering partition numbers while setting this config
to latest may cause message delivery loss since " +
- "producers could start to send messages to newly added partitions
(i.e. no initial offsets exist yet) before consumers reset their offsets.";
+ "<p>Note that increasing a topic's partition count while this
config is set to <code>latest</code> may cause silent " +
+ "message loss: producers may begin appending records to a newly
created partition before the consumer discovers it, " +
+ "and <code>latest</code> resets the position to the log end
offset, skipping any records produced during that discovery gap.</p>" +
+ "<p>To avoid this, prefer
<code>by_duration:<duration></code>. When a partition has no committed
offset, " +
+ "<code>by_duration</code> determines the starting position by
issuing a <code>ListOffsets</code> lookup for " +
+ "<code>now() - duration</code>. If the target timestamp is earlier
than the partition's creation time, the lookup " +
+ "returns the partition's start offset, ensuring that records
produced during the discovery window are still consumed. Size the duration to
cover " +
+ "the worst-case partition-discovery latency for the group protocol
in use:</p>" +
+ "<ul><li>With the <code>consumer</code> group protocol (KIP-848),
newly assigned partitions are pushed on the next " +
+ "group heartbeat, so a value at least as large as
<code>group.consumer.heartbeat.interval.ms</code> " +
+ "(server default 5000 ms) is sufficient, for example
<code>by_duration:PT5S</code>.</li>" +
+ "<li>With the <code>classic</code> group protocol, new partitions
are discovered through periodic metadata refresh " +
+ "and a subsequent rebalance, so the duration must exceed
<code>metadata.max.age.ms</code> (client default " +
+ "300000 ms) plus the rebalance time, for example
<code>by_duration:PT6M</code>.</li></ul>" +
+ "<p>Consumers with a valid committed offset are unaffected. The
reset applies only to partitions whose offset is " +
+ "missing or out of range, so <code>by_duration</code> does not
force existing consumers to replay historical data on restart.</p>";
/**
* <code>fetch.min.bytes</code>
diff --git a/docs/operations/basic-kafka-operations.md
b/docs/operations/basic-kafka-operations.md
index 8710c283a20..0e367205395 100644
--- a/docs/operations/basic-kafka-operations.md
+++ b/docs/operations/basic-kafka-operations.md
@@ -62,6 +62,18 @@ $ bin/kafka-topics.sh --bootstrap-server localhost:9092
--alter --topic my_topic
* **Key Distribution Changes**: If data is partitioned by `hash(key) %
number_of_partitions`, the default partitioner's mapping logic changes when the
partition count increases. This means that messages with the same key may be
routed to different partitions after the expansion, potentially affecting
message ordering guarantees for existing keys. Kafka will not attempt to
automatically redistribute existing data.
* **Potential Data Loss with `auto.offset.reset=latest`**: Existing
consumers configured with `auto.offset.reset=latest` might miss messages
produced to the new partitions during the window between partition creation and
consumer discovery. This occurs because consumers may not immediately detect
the new partitions, and any messages produced to those partitions before the
consumer rebalances will be skipped.
+ **Recommendation:** Use `auto.offset.reset=by_duration:<duration>` instead
of `latest` for consumers that read from topics whose partition count may
increase. When a partition has no committed offset, `by_duration` performs a
`ListOffsets` lookup for `now() - duration` to determine the starting position.
If the target timestamp is earlier than the partition's creation time, the
lookup returns the partition's earliest available offset (its log start
offset), ensuring that records prod [...]
+
+ Size `<duration>` to cover the worst-case partition-discovery latency for
the group protocol in use:
+
+ | Group protocol | Discovery mechanism |
Governing config (default) | Recommended
by_duration value |
+
|----------------------|------------------------------------------------|-----------------------------------------------------------------|-------------------------------------------------------|
+ | `consumer` (KIP-848) | Server pushes assignment on the next heartbeat |
`group.consumer.heartbeat.interval.ms` (`5000` ms, server-side) | `PT5S` or
slightly higher than the heartbeat interval |
+ | `classic` | Client-side periodic metadata refresh |
`metadata.max.age.ms` (`300000` ms, client-side) | `PT6M` or
slightly higher than `metadata.max.age.ms` |
+
+ If either the heartbeat interval or `metadata.max.age.ms` is tuned away
from the default, increase `<duration>` accordingly. **Be aware of the
trade-off:** unlike `latest`, `by_duration` makes the consumer read further
back into the backlog. Whenever a partition has no committed offset — including
a brand-new consumer group or a partition assigned for the first time — the
consumer starts at `now() - duration` and therefore replays up to `<duration>`
worth of already-produced records [...]
+
+ **Clock synchronization requirement:** `by_duration` computes the target
timestamp using the client's wall-clock time (`now() - duration`), while the
`ListOffsets` lookup compares it against broker-side message timestamps.
Accurate clock synchronization between clients and brokers is therefore
required. If the client clock runs ahead of the brokers, the consumer may start
from a position that is too recent and skip records. If the client clock lags
behind, the consumer may start too [...]
* **Metadata Propagation Delay**: New partitions are not immediately visible
to producers and consumers due to metadata refresh intervals (controlled by
`metadata.max.age.ms`). There will be a brief period where clients are unaware
of the new partitions, which may result in uneven distribution of messages or
consumer lag.
* **Risks with Internal Topics**: Users should **never** manually increase
partitions for Kafka's internal state topics such as `__consumer_offsets`,
`__transaction_state`, `__share_group_state`, or `__cluster_metadata`. Doing
so can break coordinator mapping logic, cause state inconsistencies, and lead
to data corruption or system failures. These topics are managed automatically
by Kafka and should not be modified manually.
