MINOR: Update Consumer and Producer JavaDocs for committing offsets #18336

mjsax · 2024-12-28T00:38:47Z

The consumer/producer JavaDocs still contain instruction for naively computing the offset to be committed.
This PR updates the JavaDocs with regard to the improvements of KIP-1094.

The consumer/producer JavaDocs still contain instruction for naively computing the offset to be committed. This PR updates the JavaDocs with regard to the improvements of KIP-1094.

AndrewJSchofield

Thanks for the PR. A few small comments, but looks like a good improvement.

AndrewJSchofield · 2024-12-28T08:19:19Z

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

- * which is at position 5 has consumed records with offsets 0 through 4 and will next receive the record with offset 5. There
- * are actually two notions of position relevant to the user of the consumer:
+ * which is at position 5 has consumed records with offsets 0 through 4 and will next receive the record with offset 5.
+ * Note that offsets are not guaranteed to be consecutive (eg., for compacted topic, or&mdash;independent of "read_committed"


I suggest the following the parentheses "(such as compacted topic or when records have been produced using transactions)".

AndrewJSchofield · 2024-12-28T08:19:42Z

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

+ * which is at position 5 has consumed records with offsets 0 through 4 and will next receive the record with offset 5.
+ * Note that offsets are not guaranteed to be consecutive (eg., for compacted topic, or&mdash;independent of "read_committed"
+ * mode&mdash; transactional topics). For example, if the consumer did read a record with offset 4, but 5 is not an offset
+ * with a record, it's position might advance to 6 (or higher) directly. Similarly, if the consumer's position is 5,


nit: its not it's

AndrewJSchofield · 2024-12-28T08:33:01Z

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

@@ -984,7 +990,9 @@ public void commitSync(Duration timeout) {
     * This commits offsets to Kafka. The offsets committed using this API will be used on the first fetch after every
     * 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. lastProcessedMessageOffset + 1. If automatic group management with {@link #subscribe(Collection)} is used,
+     * i.e. {@code nextRecordToBeProcessed.offset()} (or {@link ConsumerRecords#nextOffsets()}).
+     * You should also add the {@link ConsumerRecord#leaderEpoch()} (or {@code nextOffsets().get(...).leaderEpoch()})


Maybe "You should also add the leader epoch as commit metadata, which can be obtained from {@link ConsumerRecord#leaderEpoch()} or {@link ConsumerRecords#nextOffsets}." I didn't find the nextOffsets().get(...).leaderEpoch() that easy to follow and the hyperlink to the ConsumerRecords seems nicer to me.

AndrewJSchofield · 2024-12-28T08:33:34Z

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

@@ -1033,7 +1041,9 @@ public void commitSync(final Map<TopicPartition, OffsetAndMetadata> offsets) {
     * This commits offsets to Kafka. The offsets committed using this API will be used on the first fetch after every
     * 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. lastProcessedMessageOffset + 1. If automatic group management with {@link #subscribe(Collection)} is used,
+     * i.e. {@code nextRecordToBeProcessed.offset()} (or {@link ConsumerRecords#nextOffsets()}).
+     * You should also add the {@link ConsumerRecord#leaderEpoch()} (or {@code nextOffsets().get(...).leaderEpoch()})


Same point as above about nextOffsets().get(...).leaderEpoch().

AndrewJSchofield · 2024-12-28T08:33:44Z

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

@@ -1117,7 +1127,9 @@ public void commitAsync(OffsetCommitCallback callback) {
     * This commits offsets to Kafka. The offsets committed using this API will be used on the first fetch after every
     * 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. lastProcessedMessageOffset + 1. If automatic group management with {@link #subscribe(Collection)} is used,
+     * i.e. {@code nextRecordToBeProcessed.offset()} (or {@link ConsumerRecords#nextOffsets()}).
+     * You should also add the {@link ConsumerRecord#leaderEpoch()} (or {@code nextOffsets().get(...).leaderEpoch()})


AndrewJSchofield · 2024-12-28T08:34:08Z

clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java

-     * be the next message your application will consume, i.e. lastProcessedMessageOffset + 1.
+     * be the next message your application will consume, i.e. {@code nextRecordToBeProcessed.offset()}
+     * (or {@link ConsumerRecords#nextOffsets()}). You should also add the {@link ConsumerRecord#leaderEpoch()}
+     * (or {@code nextOffsets().get(...).leaderEpoch()}) as commit metadata.


chia7712

@mjsax thanks for this patch. one small comment is left. PTAL

chia7712 · 2025-01-01T13:28:35Z

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

@@ -266,8 +270,7 @@
 *                 for (ConsumerRecord&lt;String, String&gt; record : partitionRecords) {
 *                     System.out.println(record.offset() + &quot;: &quot; + record.value());
 *                 }
- *                 long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
- *                 consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)));
+ *                 consumer.commitSync(Collections.singletonMap(partition, partitionRecords.nextOffsets().get(partition));


typo: partitionRecords -> records

and missing parenthesis, I guess we should end with :

Suggested change

* consumer.commitSync(Collections.singletonMap(partition, partitionRecords.nextOffsets().get(partition));

* consumer.commitSync(Collections.singletonMap(partition, records.nextOffsets().get(partition)));

partitionRecords is actually correct; nextRecords() was added to ConsumerRecord[s] (not ConsumerRecord) -- that's also why we need to call get(partition) -- ConsumerRecord is already data for a single partition, and get(partition) would not make sense.

@mjsax Could you please take a look at following attachment? the type of partitionRecords is List<ConsumerRecord<String, String>> rather than ConsumerRecords

btw, it miss a ) also.

Thanks for sanity checking... I did mix up the variable names.

lianetm

Thanks for the nice update! Just couple of comments (and agree with the previous suggestions)

lianetm · 2025-01-02T15:26:20Z

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

+ * mode&mdash; transactional topics). For example, if the consumer did read a record with offset 4, but 5 is not an offset
+ * with a record, it's position might advance to 6 (or higher) directly. Similarly, if the consumer's position is 5,
+ * but there is no record with offset 5, the consumer will return the record with the next higher offset.
+ * There are actually two notions of position relevant to the user of the consumer:


Related to this but on ln 84 right below (sorry couldn't add comment there):

The {@link #commitSync() committed position} is the last offset that has been stored securely

shouldn't that refer to committed(..) instead?

The {@link #committed(Set) committed position} is the last offset that has been stored securely

Not sure... I did not modify this part... There are actually two notions of position relevant to the user of the consumer: did not change.

But I think it ok as is? Guess it works both ways... It's just a difference between "write path" vs "read path", right?

Let me know what you think. If you think using #commited(Set) is better, happy to update it, but it's orthogonal to what I want to do in this PR and would be some side improvement.

lianetm · 2025-01-02T15:28:53Z

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

@@ -266,8 +270,7 @@
 *                 for (ConsumerRecord&lt;String, String&gt; record : partitionRecords) {
 *                     System.out.println(record.offset() + &quot;: &quot; + record.value());
 *                 }
- *                 long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
- *                 consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)));
+ *                 consumer.commitSync(Collections.singletonMap(partition, partitionRecords.nextOffsets().get(partition));


and missing parenthesis, I guess we should end with :

Suggested change

* consumer.commitSync(Collections.singletonMap(partition, partitionRecords.nextOffsets().get(partition));

* consumer.commitSync(Collections.singletonMap(partition, records.nextOffsets().get(partition)));

mjsax · 2025-01-03T18:37:52Z

Thanks for all the input. Pushed an updated.

MINOR: Update Consumer and Producer JavaDocs for committing offsets

df949d3

The consumer/producer JavaDocs still contain instruction for naively computing the offset to be committed. This PR updates the JavaDocs with regard to the improvements of KIP-1094.

mjsax added docs producer consumer labels Dec 28, 2024

github-actions bot added clients small Small PRs labels Dec 28, 2024

AndrewJSchofield requested changes Dec 28, 2024

View reviewed changes

chia7712 reviewed Jan 1, 2025

View reviewed changes

lianetm reviewed Jan 2, 2025

View reviewed changes

mjsax added 2 commits January 2, 2025 13:56

review comments

56716c0

review comment

8bae23b

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

MINOR: Update Consumer and Producer JavaDocs for committing offsets #18336

MINOR: Update Consumer and Producer JavaDocs for committing offsets #18336

mjsax commented Dec 28, 2024

AndrewJSchofield left a comment

AndrewJSchofield Dec 28, 2024

AndrewJSchofield Dec 28, 2024

AndrewJSchofield Dec 28, 2024

AndrewJSchofield Dec 28, 2024

AndrewJSchofield Dec 28, 2024

AndrewJSchofield Dec 28, 2024

chia7712 left a comment

chia7712 Jan 1, 2025

lianetm Jan 2, 2025

mjsax Jan 2, 2025 •

edited

Loading

chia7712 Jan 3, 2025

mjsax Jan 3, 2025

lianetm left a comment

lianetm Jan 2, 2025

mjsax Jan 2, 2025

lianetm Jan 2, 2025

mjsax commented Jan 3, 2025

	* consumer.commitSync(Collections.singletonMap(partition, partitionRecords.nextOffsets().get(partition));
	* consumer.commitSync(Collections.singletonMap(partition, records.nextOffsets().get(partition)));

MINOR: Update Consumer and Producer JavaDocs for committing offsets #18336

Are you sure you want to change the base?

MINOR: Update Consumer and Producer JavaDocs for committing offsets #18336

Conversation

mjsax commented Dec 28, 2024

AndrewJSchofield left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chia7712 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax Jan 2, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lianetm left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax commented Jan 3, 2025

mjsax Jan 2, 2025 •

edited

Loading