You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by fr...@apache.org on 2026/09/02 10:28:20 UTC
(kafka) branch trunk updated: KAFKA-20507: Document impact of committing offsets without a leader epoch (#23183)
This is an automated email from the ASF dual-hosted git repository.
frankvicky 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 ad9b3d12605 KAFKA-20507: Document impact of committing offsets without a leader epoch (#23183)
ad9b3d12605 is described below
commit ad9b3d12605e6437711eb7dff4b51613cff30812
Author: Gavin Wang <z2...@uwaterloo.ca>
AuthorDate: Wed Sep 2 06:28:12 2026 -0400
KAFKA-20507: Document impact of committing offsets without a leader epoch (#23183)
Offsets committed without a leader epoch (e.g. `new
OffsetAndMetadata(offset)`) cannot be validated with the
OffsetsForLeaderEpoch API when a consumer resumes from them. If such an
offset no longer exists after log truncation following a partition
leader change, the consumer fails with OFFSET_OUT_OF_RANGE on fetch and
falls back to the auto.offset.reset policy, which can cause large-scale
duplicate processing (earliest) or data loss (latest). This surfaced in
KAFKA-19902, where an application committed epoch-less offsets from a
rebalance listener; the existing javadoc recommended including the epoch
but never explained the consequences of omitting it.
Document the consequences in the OffsetAndMetadata constructors that
leave the epoch empty, in the "Manual Offset Control" section and the
manual commitSync/commitAsync variants of the KafkaConsumer javadoc, and
in seek(TopicPartition, long), which clears the epoch of the consumer
position. Javadoc-only change, no behavior change.
Reviewers: Uros (github:uros-b), TengYao Chi <fr...@apache.org>
---
.../kafka/clients/consumer/KafkaConsumer.java | 44 ++++++++++++++++++----
.../kafka/clients/consumer/OffsetAndMetadata.java | 32 ++++++++++++++++
2 files changed, 68 insertions(+), 8 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
index 0c520f7671a..299936665e3 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
@@ -287,8 +287,19 @@ import static org.apache.kafka.common.utils.Utils.propsToMap;
* <b>Note: The committed offset should always be the offset of the next message that your application will read.</b>
* Thus, when calling {@link #commitSync(Map) commitSync(offsets)} you should use {@code nextRecordToBeProcessed.offset()}
* or if {@link ConsumerRecords} is exhausted already {@link ConsumerRecords#nextOffsets()} instead.
- * You should also add the leader epoch as commit metadata, which can be obtained from
- * {@link ConsumerRecord#leaderEpoch()} or {@link ConsumerRecords#nextOffsets()}.
+ * <p>
+ * <b>You should also include the leader epoch of the consumed record in the committed
+ * {@link OffsetAndMetadata}</b>, which can be obtained from {@link ConsumerRecord#leaderEpoch()} or
+ * {@link ConsumerRecords#nextOffsets()}, e.g.
+ * {@code new OffsetAndMetadata(record.offset() + 1, record.leaderEpoch(), "")}. The consumer uses the
+ * committed leader epoch to validate that the committed offset still exists in the partition when
+ * resuming from it, and to correct the position if the log was truncated after a partition leader
+ * change. Offsets committed without a leader epoch (e.g. constructed with
+ * {@link OffsetAndMetadata#OffsetAndMetadata(long)}) skip this validation: should such an offset no
+ * longer exist, the consumer resuming from it fails with an {@code OFFSET_OUT_OF_RANGE} error and falls
+ * back to the {@code auto.offset.reset} policy, which may cause duplicate processing (reset to earliest)
+ * or data loss (reset to latest). The offsets returned by {@link ConsumerRecords#nextOffsets()}, as used
+ * in the example above, already include the leader epoch.
*
* <h4><a name="manualassignment">Manual Partition Assignment</a></h4>
*
@@ -1076,8 +1087,12 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
* rebalance and also on startup. As such, if you need to store offsets in anything other than Kafka, this API
* should not be used. The committed offset should be the next message your application will consume,
* i.e. {@code nextRecordToBeProcessed.offset()} (or {@link ConsumerRecords#nextOffsets()}).
- * You should also add the leader epoch as commit metadata, which can be obtained from
- * {@link ConsumerRecord#leaderEpoch()} or {@link ConsumerRecords#nextOffsets()}.
+ * You should also include the leader epoch of the consumed record in the committed
+ * {@link OffsetAndMetadata}, which can be obtained from {@link ConsumerRecord#leaderEpoch()} or
+ * {@link ConsumerRecords#nextOffsets()}. Without it, the consumer cannot validate the committed
+ * offset when resuming from it and falls back to the {@code auto.offset.reset} policy if the offset
+ * no longer exists in the partition; see the "Manual Offset Control" section of the class
+ * documentation for details.
* If automatic group management with {@link #subscribe(Collection)} is used,
* then the committed offsets must belong to the currently auto-assigned partitions.
* <p>
@@ -1130,8 +1145,12 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
* rebalance and also on startup. As such, if you need to store offsets in anything other than Kafka, this API
* should not be used. The committed offset should be the next message your application will consume,
* i.e. {@code nextRecordToBeProcessed.offset()} (or {@link ConsumerRecords#nextOffsets()}).
- * You should also add the leader epoch as commit metadata, which can be obtained from
- * {@link ConsumerRecord#leaderEpoch()} or {@link ConsumerRecords#nextOffsets()}.
+ * You should also include the leader epoch of the consumed record in the committed
+ * {@link OffsetAndMetadata}, which can be obtained from {@link ConsumerRecord#leaderEpoch()} or
+ * {@link ConsumerRecords#nextOffsets()}. Without it, the consumer cannot validate the committed
+ * offset when resuming from it and falls back to the {@code auto.offset.reset} policy if the offset
+ * no longer exists in the partition; see the "Manual Offset Control" section of the class
+ * documentation for details.
* If automatic group management with {@link #subscribe(Collection)} is used,
* then the committed offsets must belong to the currently auto-assigned partitions.
* <p>
@@ -1219,8 +1238,12 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
* rebalance and also on startup. As such, if you need to store offsets in anything other than Kafka, this API
* should not be used. The committed offset should be the next message your application will consume,
* i.e. {@code nextRecordToBeProcessed.offset()} (or {@link ConsumerRecords#nextOffsets()}).
- * You should also add the leader epoch as commit metadata, which can be obtained from
- * {@link ConsumerRecord#leaderEpoch()} or {@link ConsumerRecords#nextOffsets()}.
+ * You should also include the leader epoch of the consumed record in the committed
+ * {@link OffsetAndMetadata}, which can be obtained from {@link ConsumerRecord#leaderEpoch()} or
+ * {@link ConsumerRecords#nextOffsets()}. Without it, the consumer cannot validate the committed
+ * offset when resuming from it and falls back to the {@code auto.offset.reset} policy if the offset
+ * no longer exists in the partition; see the "Manual Offset Control" section of the class
+ * documentation for details.
* If automatic group management with {@link #subscribe(Collection)} is used,
* then the committed offsets must belong to the currently auto-assigned partitions.
* <p>
@@ -1266,6 +1289,11 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
* <p>
* Note that, the seek offset won't change to the in-flight fetch request, it will take effect in next fetch request.
* So, the consumer might wait for {@code fetch.max.wait.ms} before starting to fetch the records from desired offset.
+ * <p>
+ * Note that this method clears the leader epoch of the consumer position, so offsets committed before any
+ * further records are consumed (e.g. via {@link #commitSync()}) will not include a leader epoch and cannot
+ * be validated when a consumer resumes from them. Use {@link #seek(TopicPartition, OffsetAndMetadata)} to
+ * seek with a leader epoch; see the "Manual Offset Control" section of the class documentation for details.
*
* @param partition the TopicPartition on which the seek will be performed.
* @param offset the next offset returned by poll().
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/OffsetAndMetadata.java b/clients/src/main/java/org/apache/kafka/clients/consumer/OffsetAndMetadata.java
index 5d4d1550c16..fbb2b0295c7 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/OffsetAndMetadata.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/OffsetAndMetadata.java
@@ -27,6 +27,18 @@ import java.util.Optional;
* The Kafka offset commit API allows users to provide additional metadata (in the form of a string)
* when an offset is committed. This can be useful (for example) to store information about which
* node made the commit, what time the commit was made, etc.
+ * <p>
+ * Besides the offset and the metadata string, this class carries the <i>leader epoch</i> of the consumed
+ * record, which can be obtained from {@link ConsumerRecord#leaderEpoch()} or
+ * {@link ConsumerRecords#nextOffsets()}. The consumer uses the committed leader epoch to validate that
+ * the committed offset still exists in the partition when resuming from it, and to correct the position
+ * if the log was truncated after a partition leader change. If an offset is committed without a leader
+ * epoch, this validation is skipped: should the committed offset no longer exist, the consumer resuming
+ * from it fails with an {@code OFFSET_OUT_OF_RANGE} error and falls back to the
+ * {@code auto.offset.reset} policy, which may cause duplicate processing (reset to earliest) or data
+ * loss (reset to latest). For this reason, prefer
+ * {@link #OffsetAndMetadata(long, Optional, String)} and include the leader epoch of the consumed
+ * record whenever it is available.
*/
@InterfaceAudience.Public
public class OffsetAndMetadata implements Serializable {
@@ -42,6 +54,10 @@ public class OffsetAndMetadata implements Serializable {
/**
* Construct a new OffsetAndMetadata object for committing through {@link KafkaConsumer}.
+ * This is the preferred constructor for manually committed offsets: including the leader epoch of the
+ * consumed record enables the consumer to validate the committed offset when resuming from it
+ * (see the class-level documentation), e.g.
+ * {@code new OffsetAndMetadata(record.offset() + 1, record.leaderEpoch(), "")}.
*
* @param offset The offset to be committed
* @param leaderEpoch Optional leader epoch of the last consumed record
@@ -61,6 +77,14 @@ public class OffsetAndMetadata implements Serializable {
/**
* Construct a new OffsetAndMetadata object for committing through {@link KafkaConsumer}.
+ * <p>
+ * Note that the leader epoch of the committed offset will be empty, which disables offset validation
+ * when the consumer resumes from it: if the offset no longer exists in the partition (e.g. due to log
+ * truncation after a partition leader change), the consumer resets the position according to the
+ * {@code auto.offset.reset} policy rather than correcting it. Prefer
+ * {@link #OffsetAndMetadata(long, Optional, String)} and include the leader epoch of the consumed
+ * record whenever it is available.
+ *
* @param offset The offset to be committed
* @param metadata Non-null metadata
*/
@@ -71,6 +95,14 @@ public class OffsetAndMetadata implements Serializable {
/**
* Construct a new OffsetAndMetadata object for committing through {@link KafkaConsumer}. The metadata
* associated with the commit will be empty.
+ * <p>
+ * Note that the leader epoch of the committed offset will be empty, which disables offset validation
+ * when the consumer resumes from it: if the offset no longer exists in the partition (e.g. due to log
+ * truncation after a partition leader change), the consumer resets the position according to the
+ * {@code auto.offset.reset} policy rather than correcting it. Prefer
+ * {@link #OffsetAndMetadata(long, Optional, String)} and include the leader epoch of the consumed
+ * record whenever it is available.
+ *
* @param offset The offset to be committed
*/
public OffsetAndMetadata(long offset) {