KAFKA-20539 Prevent Hot Data Loss on Partition Expansion for Latest Policy

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

@@ -179,8 +179,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:&lt;duration&gt;</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 offset 0, 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 slightly greater than <code>group.consumer.heartbeat.interval.ms</code> " +
+            "(server default 5000&nbsp;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, " +
+            "so the duration must exceed <code>metadata.max.age.ms</code> (client default 300000&nbsp;ms), " +
+            "for example <code>by_duration:PT5M</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>

docs/operations/basic-kafka-operations.md

@@ -62,6 +62,16 @@ $ 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 offset `0`, ensuring that records produced during the partition-discovery gap are still consumed. Consumers with a valid committed offset are unaffected, so restarts do not replay historical data.
+
+    Size `<duration>` to cover the worst-case partition-discovery latency for the group protocol in use:
+
+    | Group protocol       | Discovery mechanism                            | Governing config (default)                                      | Recommended <duration>                                |
+    |----------------------|------------------------------------------------|-----------------------------------------------------------------|-------------------------------------------------------|
+    | `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)                | `PT5M` 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. `<duration>` uses the ISO-8601 duration format (`PnDTnHnMn.nS`). The trade-off is bounded: on the first assignment of a partition, the consumer replays at most `<duration>` worth of records, which is negligible compared with a full historical replay.
   * **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.