KAFKA-18014: Add duration based offset reset option for ShareConsumer by peterxcli · Pull Request #18096 · apache/kafka

peterxcli · 2024-12-07T17:40:31Z

Kafka consumer supports auto.offset.reset config option, which is used when there is no initial offset in Kafka (or) if the current offset does not exist any more on the server. This config currently supports earliest/latest/none options. Currently consumer resets might force applications to reprocess large amounts of data from earlier offsets. With infinite storage, its beneficial to have a duration based offset reset strategy. This will allow applications to consume/initialise from a fixed duration when there is no initial offset in Kafka.

As part of KIP-932, we are adding support for share consumer groups. Share consumer groups supports dynamic group configuration property share.auto.offset.reset. This is used to set the initial Share-Partition Start Offset (SPSO) based on the share.auto.offset.reset configuration. Currently share.auto.offset.reset supports earliest and latest options to automatically reset the offset

Similar to the Kafka Consumer, we will add support for by_duration: config value for share.auto.offset.reset.

from KIP-1106

Some conflict with: #17979

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

…set offset options

…group config on auto reset offset cases

peterxcli · 2024-12-08T14:11:14Z

+import java.util.Objects;
+import java.util.Optional;
+
+public class ShareGroupAutoOffsetResetStrategy {


As mentioned in #17573 (comment), let each AutoOffsetResetStrategy evolve themselves

Can we have javadocs please on the class.

Moreover the purpose of having a separate enum as ShareGroupAutoOffsetReset was to just mention that ShareGroup do no support all offset reset strategy but the ones explicitly mentioned in the ShareGroupAutoOffsetReset. Which is the sub strategies from AutoOffsetResetStrategy.

This class seems same as AutoOffsetResetStartegy, why do we need a different one? If we do need one then do we need to copy the methods as it is or could just use ShareGroupAutoOffsetResetStrategy as a wrapper over AutoOffsetResetStrategy to differentiate with self evolving startegies of ShareGroup?

I think we still need a separate ShareGroupAutoOffsetResetStrategy, because this allows us to separate the enum and validation logic. As for the wrapper part you mentioned, perhaps we could use the Delegation pattern to reuse AutoOffsetResetStrategy's logic. What do you think?

I think we still need a separate ShareGroupAutoOffsetResetStrategy

Yes, that's fine.

perhaps we could use the Delegation pattern

That could be a good option.

I wouldn't over-engineer this. It's a pretty trivial class and there's not much saved. Up to you in this case from my point of view.

@apoorvmittal10 Would like to know your opinion on this.

peterxcli · 2024-12-08T14:14:38Z

Hi @omkreddy, PR is ready, PTAL. Thanks!
Sorry for mentioning you at both side, just to make sure you do receive that 😁~

AndrewJSchofield

Thanks for the PR. A few comments to address.

AndrewJSchofield · 2024-12-08T20:21:59Z

+        Option<FileRecords.TimestampAndOffset> timestampAndOffset = replicaManager.fetchOffsetForTimestamp(
+            topicIdPartition.topicPartition(), timestampToSearch, Option.empty(), Optional.of(leaderEpoch), true).timestampAndOffsetOpt();
+        if (timestampAndOffset.isEmpty()) {
+            throw new OffsetNotAvailableException("offset for timestamp to search: " + timestampToSearch + " not found for topic partition: " + topicIdPartition);


For consistency, I would adjust the capitalisation. I suggest "Offset for timestamp " + timestamptoSearch + " not found for topic partition" instead.

Looking at the latest changes, this was not done as requested. Please can you change the wording.

AndrewJSchofield · 2024-12-08T20:31:22Z

+                fromString(offsetStrategy);
+            } catch (Exception e) {
+                throw new ConfigException(name, value, "Invalid value `" + offsetStrategy + "` for configuration " +
+                        name + ". The value must be either 'earliest', 'latest', 'none' or of the format 'by_duration:<PnDTnHnMn.nS.>'.");


none is not permitted for a share group.

You still need to remove 'none' from this exception message because it is not a valid option here.

apoorvmittal10

Thanks for the PR. I ll have a second pass later today on PR.

apoorvmittal10 · 2024-12-09T14:55:02Z

+    static long offsetForTimestamp(TopicIdPartition topicIdPartition, ReplicaManager replicaManager, long timestampToSearch, int leaderEpoch) {
+        Option<FileRecords.TimestampAndOffset> timestampAndOffset = replicaManager.fetchOffsetForTimestamp(
+            topicIdPartition.topicPartition(), timestampToSearch, Option.empty(), Optional.of(leaderEpoch), true).timestampAndOffsetOpt();
+        if (timestampAndOffset.isEmpty()) {
+            throw new OffsetNotAvailableException("Offset for timestamp: " + timestampToSearch + " not found for topic partition: " + topicIdPartition);
+        }
+        return timestampAndOffset.get().offset;
+    }
+


Should we send the the isolation level as well instead of Option.empty()? Also we can have common method and send respective timestamps from: offsetForEarliestTimestamp, offsetForLatestTimestamp and offsetForTimestamp methods.

I suggest READ_UNCOMMITTED isolation level is the best option here. When we add support for isolation levels in share groups, we can revisit. So far, share groups always use READ_UNCOMMITTED.

apoorvmittal10 · 2024-12-09T14:55:50Z

+import java.util.Objects;
+import java.util.Optional;
+
+public class ShareGroupAutoOffsetResetStrategy {


Can we have javadocs please on the class.

AndrewJSchofield

I think this PR is almost ready to merge. Just a couple of outstanding issues (isolation level, javadoc).

apoorvmittal10 · 2024-12-09T22:28:28Z


-        if (offsetResetStrategy == GroupConfig.ShareGroupAutoOffsetReset.EARLIEST)
+        if (offsetResetStrategy.timestamp().isEmpty()) {
+            throw new Exception("The timestamp is not available for the share partition: " + topicIdPartition);


This will translate to UnknownServerError to the client. Do we want to return that or specific exception?

I think I would change this to return NoOffsetForPartitionException(topicIdPartition.topicPartition()):
https://github.com/apache/kafka/pull/17972/files#diff-8a7549a9127f9c2eac443cfb35f0d7c1ee8e2884576badd3670a7f19460c59d2R293

So, currently, this could only occur when BY_DURATION is used and duration somehow is not set. Would IllegalStateException be better to throw as there shouldn't be a case when offsetResetStrategy.timestamp() should ideally be empty?

I think we can remove this exception here, because ShareGroupAutoOffsetResetStrategy already has the validation itself, so the .timestamp() must have value if its type is BY_DURATION. And we can add some comment to indicate this. Thoughts?

I would say it would be good to have a check as the method returns Optional. If it can never be empty then the method should never return Optional.

peterxcli · 2024-12-11T07:32:07Z

Thanks for your review! address some of the comments in 3d40b6e, PTAL. Thanks!

…pAutoOffsetResetStrategyTest

apoorvmittal10

Thanks for addressing the comments. I have replied to the question and took another pass.

…tegy delegator

peterxcli · 2024-12-12T12:03:37Z

Refactored ShareGroupAutoOffsetResetStrategy and remove returning optional from ShareGroupAutoOffsetResetStrategy#timestamp with cffdcf1

PTAL. Thanks!

AndrewJSchofield

A few additional comments. I think is getting close to being ready to merge.

AndrewJSchofield · 2024-12-12T18:46:43Z

+        Option<FileRecords.TimestampAndOffset> timestampAndOffset = replicaManager.fetchOffsetForTimestamp(
+            topicIdPartition.topicPartition(), timestampToSearch, Option.empty(), Optional.of(leaderEpoch), true).timestampAndOffsetOpt();
+        if (timestampAndOffset.isEmpty()) {
+            throw new OffsetNotAvailableException("offset for timestamp to search: " + timestampToSearch + " not found for topic partition: " + topicIdPartition);


Looking at the latest changes, this was not done as requested. Please can you change the wording.

AndrewJSchofield · 2024-12-12T18:55:13Z

+import java.util.Objects;
+import java.util.Optional;
+
+public class ShareGroupAutoOffsetResetStrategy {


I wouldn't over-engineer this. It's a pretty trivial class and there's not much saved. Up to you in this case from my point of view.

AndrewJSchofield · 2024-12-13T18:40:41Z

+                fromString(offsetStrategy);
+            } catch (Exception e) {
+                throw new ConfigException(name, value, "Invalid value `" + offsetStrategy + "` for configuration " +
+                        name + ". The value must be either 'earliest', 'latest', 'none' or of the format 'by_duration:<PnDTnHnMn.nS.>'.");


You still need to remove 'none' from this exception message because it is not a valid option here.

AndrewJSchofield · 2024-12-13T18:43:00Z

+        Option<FileRecords.TimestampAndOffset> timestampAndOffset = replicaManager.fetchOffsetForTimestamp(
+            topicIdPartition.topicPartition(), timestampToSearch, new Some<>(IsolationLevel.READ_UNCOMMITTED), Optional.of(leaderEpoch), true).timestampAndOffsetOpt();
+        if (timestampAndOffset.isEmpty()) {
+            throw new OffsetNotAvailableException("offset for timestamp " + timestampToSearch + " not found for topic partition: " + topicIdPartition);


nit: Please capitalise the "Offset" for consistency with the offsetForEarliestTimestamp and offsetForLatestTimestamp methods. I know this is a trivial point but we have high standards for these things.

Thank you very much for your careful review! 😁

AndrewJSchofield

Thanks for the PR. I've run the code on my local environment and it works nicely. A couple of final points since the generated HTML documentation is not valid as the code stands. But apart from that, ready to merge.

AndrewJSchofield · 2024-12-15T21:40:18Z

-    public static final String SHARE_AUTO_OFFSET_RESET_DEFAULT = ShareGroupAutoOffsetReset.LATEST.toString();
-    public static final String SHARE_AUTO_OFFSET_RESET_DOC = "The strategy to initialize the share-partition start offset.";
+    public static final String SHARE_AUTO_OFFSET_RESET_DEFAULT = ShareGroupAutoOffsetResetStrategy.LATEST.name();
+    public static final String SHARE_AUTO_OFFSET_RESET_DOC = "The strategy to initialize the share-partition start offset. " +


I was looking into what actually happens with these documentation strings and they need to be valid for HTML generation. So, I've slightly tweaked the text to replace the < and > so they're not interpreted as tags. Please can you use this string exactly.

public static final String SHARE_AUTO_OFFSET_RESET_DOC = "The strategy to initialize the share-partition start offset. " + "<ul><li>earliest: automatically reset the offset to the earliest offset</li>" + "<li>latest: automatically reset the offset to the latest offset</li>" + "<li>by_duration:<duration>: automatically reset the offset to a configured duration from the current timestamp. " + "<duration> must be specified in ISO8601 format (PnDTnHnMn.nS). " + "Negative duration is not allowed.</li>" + "<li>anything else: throw exception to the share consumer.</li></ul>";

AndrewJSchofield · 2024-12-15T21:42:46Z

+            }
+        }
+    }
+}


This validator needs a toString() method. Otherwise, the documentation will include the validator's object reference instead of a string for the valid options. I suggest:

public String toString() { return "[earliest, latest, by_duration:PnDTnHnMn.nS]"; }

That worked nicely in the generated HTML.

AndrewJSchofield · 2024-12-15T22:18:37Z

There should also be an integration test but I've created a separate task https://issues.apache.org/jira/browse/KAFKA-18260. Feel free to pick this up @peterxcli if you're interested.

AndrewJSchofield

Thanks for the PR. Looks good to me. Waiting for a green build before merging.

…apache#18096) Kafka consumer supports auto.offset.reset config option, which is used when there is no initial offset in Kafka (or) if the current offset does not exist any more on the server. This config currently supports earliest/latest/none options. Currently consumer resets might force applications to reprocess large amounts of data from earlier offsets. With infinite storage, its beneficial to have a duration based offset reset strategy. This will allow applications to consume/initialise from a fixed duration when there is no initial offset in Kafka. As part of KIP-932, we are adding support for share consumer groups. Share consumer groups supports dynamic group configuration property share.auto.offset.reset. This is used to set the initial Share-Partition Start Offset (SPSO) based on the share.auto.offset.reset configuration. Currently share.auto.offset.reset supports earliest and latest options to automatically reset the offset Similar to the Kafka Consumer, we will add support for by_duration: config value for share.auto.offset.reset. Reviewers: Andrew Schofield <[email protected]>, Apoorv Mittal <[email protected]>

peterxcli added 6 commits December 5, 2024 01:02

Introduce ShareGroupAutoOffsetResetStrategy for more flexible auto re…

1714ef4

…set offset options

Migrate ShareGroupAutoOffsetReset to ShareGroupAutoOffsetResetStrategy

efe9d6e

Add ShareGroupAutoOffsetResetStrategyTest.java

e437f36

Implement offset retrieval by timestamp in SharePartition

3f22b3c

Clarify SHARE_AUTO_OFFSET_RESET_DOC and add some additional test for …

03b10f9

…group config on auto reset offset cases

Merge remote-tracking branch 'upstream/trunk' into k18014

67b7078

github-actions Bot added core Kafka Broker KIP-932 Queues for Kafka labels Dec 7, 2024

chia7712 added the ci-approved label Dec 7, 2024

peterxcli added 2 commits December 8, 2024 02:09

Add leaderEpoch parameter to offsetForTimestamp method

efe0184

Fix default config reference in KafkaApisTest.

b9745b1

peterxcli mentioned this pull request Dec 8, 2024

KAFKA-18015: Add support for duration based offset reset strategy to Kafka Streams #17973

Merged

peterxcli commented Dec 8, 2024

View reviewed changes

AndrewJSchofield requested changes Dec 8, 2024

View reviewed changes

peterxcli requested a review from AndrewJSchofield December 9, 2024 13:56

apoorvmittal10 requested changes Dec 9, 2024

View reviewed changes

AndrewJSchofield requested changes Dec 9, 2024

View reviewed changes

Comment thread ...ator/src/main/java/org/apache/kafka/coordinator/group/ShareGroupAutoOffsetResetStrategy.java

Comment thread .../src/test/java/org/apache/kafka/coordinator/group/ShareGroupAutoOffsetResetStrategyTest.java

apoorvmittal10 requested changes Dec 9, 2024

View reviewed changes

peterxcli force-pushed the k18014 branch from 314cb09 to b9745b1 Compare December 11, 2024 05:04

peterxcli requested review from AndrewJSchofield and apoorvmittal10 December 11, 2024 07:30

Leave blank at EOL in ShareGroupAutoOffsetResetStrategy and ShareGrou…

ebf9cac

…pAutoOffsetResetStrategyTest

peterxcli force-pushed the k18014 branch 2 times, most recently from ebf9cac to 8f2ac44 Compare December 11, 2024 07:40

Address Andrew's and Apoorv's comments

3d40b6e

peterxcli force-pushed the k18014 branch from 6b87787 to 3d40b6e Compare December 11, 2024 07:54

Fix spotless error

6bdbfcf

apoorvmittal10 requested changes Dec 11, 2024

View reviewed changes

Comment thread core/src/main/java/kafka/server/share/SharePartition.java Outdated

Refactor ShareGroupAutoOffsetResetStrategy to use AutoOffsetResetStra…

cffdcf1

…tegy delegator

peterxcli requested a review from apoorvmittal10 December 12, 2024 11:59

AndrewJSchofield requested changes Dec 12, 2024

View reviewed changes

Address comments

2a3adef

peterxcli requested a review from AndrewJSchofield December 13, 2024 09:34

AndrewJSchofield requested changes Dec 13, 2024

View reviewed changes

Address comments

d819a66

peterxcli requested a review from AndrewJSchofield December 13, 2024 18:53

AndrewJSchofield requested changes Dec 15, 2024

View reviewed changes

Address Andrew's suggestion

e11bced

peterxcli requested a review from AndrewJSchofield December 16, 2024 05:28

Merge remote-tracking branch 'upstream/trunk' into pr/peterxcli/18096

faf4dfd

AndrewJSchofield approved these changes Dec 16, 2024

View reviewed changes

AndrewJSchofield merged commit 220c578 into apache:trunk Dec 16, 2024

peterxcli mentioned this pull request Dec 18, 2024

KAFKA-18260: Add integration test to ShareConsumerTest for share.auto.offset.reset=by_duration #18251

Merged

3 tasks

Conversation

peterxcli commented Dec 7, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

peterxcli Dec 8, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

apoorvmittal10 Dec 9, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

peterxcli commented Dec 8, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

AndrewJSchofield left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

apoorvmittal10 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

AndrewJSchofield left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

peterxcli Dec 11, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

peterxcli commented Dec 11, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

apoorvmittal10 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

peterxcli commented Dec 12, 2024

Uh oh!

AndrewJSchofield left a comment

Choose a reason for hiding this comment

peterxcli commented Dec 7, 2024 •

edited

Loading

peterxcli Dec 8, 2024 •

edited

Loading

apoorvmittal10 Dec 9, 2024 •

edited

Loading

peterxcli commented Dec 8, 2024 •

edited

Loading

peterxcli Dec 11, 2024 •

edited

Loading

peterxcli commented Dec 11, 2024 •

edited

Loading