From 7d147e8cdb184d203eb42ab4197561b9f3e77610 Mon Sep 17 00:00:00 2001 From: Ken Huang Date: Wed, 8 Jul 2026 09:31:07 +0800 Subject: [PATCH 1/3] update the document --- .../kafka/clients/consumer/ConsumerConfig.java | 18 ++++++++++++++++-- docs/operations/basic-kafka-operations.md | 13 +++++++++++-- 2 files changed, 27 insertions(+), 4 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 80130cd7188fb..f370fe8608044 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 @@ -179,8 +179,22 @@ public class ConsumerConfig extends AbstractConfig { "Negative duration is not allowed." + "
  • none: throw exception to the consumer if no previous offset is found for the consumer's group
    • " + "
  • anything else: throw exception to the consumer.
    • " + - "

    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."; + "

    Note that increasing a topic's partition count while this config is set to latest may cause silent " + + "message loss: producers may begin appending records to a newly created partition before the consumer discovers it, " + + "and latest resets the position to the log end offset, skipping any records produced during that discovery gap.

    " + + "

    To avoid this, prefer by_duration:<duration>. When a partition has no committed offset, " + + "by_duration determines the starting position by issuing a ListOffsets lookup for " + + "now() - duration. 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:

    " + + "" + + "

    Consumers with a valid committed offset are unaffected. The reset applies only to partitions whose offset is " + + "missing or out of range, so by_duration does not force existing consumers to replay historical data on restart.

    "; /** * fetch.min.bytes diff --git a/docs/operations/basic-kafka-operations.md b/docs/operations/basic-kafka-operations.md index 4ed37d87417ee..cc8e00e334f11 100644 --- a/docs/operations/basic-kafka-operations.md +++ b/docs/operations/basic-kafka-operations.md @@ -62,8 +62,17 @@ $ 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. - * **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. + + **Recommendation:** Use `auto.offset.reset=by_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 `` to cover the worst-case partition-discovery latency for the group protocol in use: + + | Group protocol | Discovery mechanism | Governing config (default) | Recommended `` | + |----------------------|------------------------------------------------|-----------------------------------------------------------------|-------------------------------------------------------| + | `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 `` accordingly. `` uses the ISO-8601 duration format (`PnDTnHnMn.nS`). This setting is intended to cover the metadata propagation or assignment delay during which newly added partitions may not yet be visible to producers and consumers. During this short window, clients may remain unaware of the new partitions, which can temporarily cause uneven message distribution or consumer lag. The trade-off is bounded: on the first assignment of a partition, the consumer replays at most `` worth of records, which is negligible compared with a full historical replay. To add configs: From a6c610ab4a8df7e936a811af271a94d7c39dddc8 Mon Sep 17 00:00:00 2001 From: Ken Huang Date: Wed, 8 Jul 2026 10:21:19 +0800 Subject: [PATCH 2/3] fix table --- docs/operations/basic-kafka-operations.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/operations/basic-kafka-operations.md b/docs/operations/basic-kafka-operations.md index cc8e00e334f11..0109ce598ae18 100644 --- a/docs/operations/basic-kafka-operations.md +++ b/docs/operations/basic-kafka-operations.md @@ -67,7 +67,7 @@ $ bin/kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic my_topic Size `` to cover the worst-case partition-discovery latency for the group protocol in use: - | Group protocol | Discovery mechanism | Governing config (default) | Recommended `` | + | Group protocol | Discovery mechanism | Governing config (default) | Recommended | |----------------------|------------------------------------------------|-----------------------------------------------------------------|-------------------------------------------------------| | `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` | From 53cb132e7245f9328604c1c28e0205e4f13df504 Mon Sep 17 00:00:00 2001 From: Ken Huang Date: Wed, 8 Jul 2026 11:18:29 +0800 Subject: [PATCH 3/3] update the document --- docs/operations/basic-kafka-operations.md | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/docs/operations/basic-kafka-operations.md b/docs/operations/basic-kafka-operations.md index 0109ce598ae18..320139b8327f5 100644 --- a/docs/operations/basic-kafka-operations.md +++ b/docs/operations/basic-kafka-operations.md @@ -62,7 +62,6 @@ $ 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:` 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 `` to cover the worst-case partition-discovery latency for the group protocol in use: @@ -72,7 +71,9 @@ $ bin/kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic my_topic | `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 `` accordingly. `` uses the ISO-8601 duration format (`PnDTnHnMn.nS`). This setting is intended to cover the metadata propagation or assignment delay during which newly added partitions may not yet be visible to producers and consumers. During this short window, clients may remain unaware of the new partitions, which can temporarily cause uneven message distribution or consumer lag. The trade-off is bounded: on the first assignment of a partition, the consumer replays at most `` worth of records, which is negligible compared with a full historical replay. + If either the heartbeat interval or `metadata.max.age.ms` is tuned away from the default, increase `` accordingly. `` 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 `` 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. To add configs: