KAFKA-16726: Added share.auto.offset.reset dynamic config for share groups by chirag-wadhwa5 · Pull Request #17573 · apache/kafka

chirag-wadhwa5 · 2024-10-22T09:29:37Z

This PR adds another dynamic config share.auto.offset.reset fir share groups.

…roups in GroupConfig

….auto.offset.reset changes and pass

apoorvmittal10

Thanks for the PR, left some comments.

apoorvmittal10 · 2024-10-23T18:59:23Z

+        return ShareGroupAutoOffsetReset.forStrategy(shareAutoOffsetReset);
+    }
+
+    public enum ShareGroupAutoOffsetReset {


Why not to use existing OffsetResetStrategy?

I would keep them separate so we can evolve them separately.

Thanks for the review. It's better to use OffsetResetStrategy, so I'll remove the newly added enum.

It's better to use OffsetResetStrategy, so I'll remove the newly added enum.

Did you miss it or changed mind?

What's the final decision, are we keeping it? I would prefer to use OffsetResetStrategy as don't see any difference to one defined as new. If we need to have an independent one, in future, because of some more states then create a new one. Else it's unnecessary.

I expect that OffsetResetStrategy will evolve independently and I prefer to separate them now.

Had a word with @AndrewJSchofield. Sure, let's have separate one.

@chirag-wadhwa5 When we define just earliest and latest in this enum then do we need the explicit validations we have specified for this config?

I don't think we would require any explicit validations on top of it. I have removed the unnecessary validations now.

AndrewJSchofield · 2024-10-24T12:22:54Z

+        return ShareGroupAutoOffsetReset.forStrategy(shareAutoOffsetReset);
+    }
+
+    public enum ShareGroupAutoOffsetReset {


I would keep them separate so we can evolve them separately.

apoorvmittal10

Some comments to address.

apoorvmittal10 · 2024-10-25T09:02:11Z

+        Option<FileRecords.TimestampAndOffset> timestampAndOffset = replicaManager.fetchOffsetForTimestamp(
+            topicIdPartition.topicPartition(), ListOffsetsRequest.LATEST_TIMESTAMP, new Some<>(IsolationLevel.READ_UNCOMMITTED),
+            Optional.empty(), true).timestampAndOffsetOpt();
+        return timestampAndOffset.isEmpty() ? (long) 0 : timestampAndOffset.get().offset;


Is it safe to return 0 from this method when timestampAndOffset is empty?

Thanks for the review. We could instead throw an error saying the latest offset not found, or maybe maybe initialize to the earliest offset, in case the latest is not found. What would you suggest here @apoorvmittal10 ?

If we send 0 when timestamp is empty but if first fetch offset in log is at 10 then will the fetch throw an exception? If yes, then what will happen next? And in waht cases the the timestamp can be empty from replica manager?

I think there is the possibility of an exception in the slightly unusual situation of a new leader which is lagging behind the current epoch. While I'm sure we're not going to bump into that right now, it is a real situation in the replica manager I think and the interface here needs to be flexible enough for an exception and we need to have a plan for what to do if the exception occurs. @mumrah can you help here please? I guess we'd want to do some kind of retry, which is perhaps best achieved by aborting the share-partition initialisation and letting the next attempt to initialize the share-partition have another try. The edges of the replication behaviour are a bit challenging I think.

Hi @apoorvmittal10 , when replicaManager.fetchOffsetForTimestamp is called with LATEST_TIMESTAMP, as per the current code, only 2 things are possible, either it will throw an error or it will return a populated response (not a None object, which is possible when the same method is called with EARLIEST_TIMESTAMP). Now the error handling is there for when an error is thrown, but apart from that I don't suppose we need any other checks for whether the offset value is populated or not. But, if we want to have this check in place, I don't think we can assume any other value in place of this. The only right way would be to fail the share partition initialization. Another point to note is when offsetForEarliestTimestamp is called with EARLIEST_TIMESTAMP, the current code would return 0 if the returned value from replicaManager.fetchOffsetForTimestamp is None, which again would be a wrong assumption in case the start offset of the log is beyond 0. I believe the correct way is to fail the share partition initialization here as well.

adixitconfluent

Did a partial review. Left some comments.

apoorvmittal10

Thanks for the changes. Some comments for clarification and changes. Mostly LGTM!

apoorvmittal10 · 2024-10-28T10:45:01Z

+        return ShareGroupAutoOffsetReset.forStrategy(shareAutoOffsetReset);
+    }
+
+    public enum ShareGroupAutoOffsetReset {


It's better to use OffsetResetStrategy, so I'll remove the newly added enum.

Did you miss it or changed mind?

What's the final decision, are we keeping it? I would prefer to use OffsetResetStrategy as don't see any difference to one defined as new. If we need to have an independent one, in future, because of some more states then create a new one. Else it's unnecessary.

AndrewJSchofield · 2024-10-28T12:22:22Z

@@ -128,4 +130,17 @@ static long offsetForEarliestTimestamp(TopicIdPartition topicIdPartition, Replic
                Optional.empty(), true).timestampAndOffsetOpt();


Unfortunately, I think this is more complicated that it seems. If KIP-405 is enabled, then returning EARLIEST_TIMESTAMP is asynchronous. It may be necessary to wait before there's an answer to the earliest offset. If we put that concern aside for this PR, we do need an issue to track this.

AndrewJSchofield · 2024-10-28T12:27:14Z

+        Option<FileRecords.TimestampAndOffset> timestampAndOffset = replicaManager.fetchOffsetForTimestamp(
+            topicIdPartition.topicPartition(), ListOffsetsRequest.LATEST_TIMESTAMP, new Some<>(IsolationLevel.READ_UNCOMMITTED),
+            Optional.empty(), true).timestampAndOffsetOpt();
+        return timestampAndOffset.isEmpty() ? (long) 0 : timestampAndOffset.get().offset;


I think there is the possibility of an exception in the slightly unusual situation of a new leader which is lagging behind the current epoch. While I'm sure we're not going to bump into that right now, it is a real situation in the replica manager I think and the interface here needs to be flexible enough for an exception and we need to have a plan for what to do if the exception occurs. @mumrah can you help here please? I guess we'd want to do some kind of retry, which is perhaps best achieved by aborting the share-partition initialisation and letting the next attempt to initialize the share-partition have another try. The edges of the replication behaviour are a bit challenging I think.

AndrewJSchofield · 2024-10-28T12:32:18Z

+        return ShareGroupAutoOffsetReset.forStrategy(shareAutoOffsetReset);
+    }
+
+    public enum ShareGroupAutoOffsetReset {


I expect that OffsetResetStrategy will evolve independently and I prefer to separate them now.

…FetchAcknowledgeRequestTest

apoorvmittal10

LGTM, just I am not understanding the change with re-ordered tests and an open query on returning 0 with offsetForTimestamp method.

apoorvmittal10 · 2024-11-04T23:20:18Z

-    val send: Seq[TopicIdPartition] = Seq(topicIdPartition)
-
-    // Send the share fetch request to fetch the records produced above
-    val metadata: ShareRequestMetadata = new ShareRequestMetadata(memberId, ShareRequestMetadata.INITIAL_EPOCH)
-    val acknowledgementsMap: Map[TopicIdPartition, util.List[ShareFetchRequestData.AcknowledgementBatch]] = Map.empty
-    val shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMap)
-    val shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+    // Send the second share fetch request to fetch the records produced above
+    metadata = new ShareRequestMetadata(memberId, ShareRequestMetadata.nextEpoch(ShareRequestMetadata.INITIAL_EPOCH))
+    acknowledgementsMap = Map.empty
+    shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMap)
+    shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)


Cant we define the offset reset startegy while creating KafkaShareConsumer?

…fsetForLatestTimestamp in ShareFetchUtils

AndrewJSchofield · 2024-11-05T16:09:13Z

+    @ParameterizedTest(name = "{displayName}.persister={0}")
+    @ValueSource(strings = {NO_OP_PERSISTER, DEFAULT_STATE_PERSISTER})
+    public void testShareAutoOffsetResetMultipleGroupsWithDifferentValue(String persister) {
+        KafkaShareConsumer<byte[], byte[]> shareConsumer1 = createShareConsumer(new ByteArrayDeserializer(), new ByteArrayDeserializer(), "group1");


How about shareConsumerEarliest and shareConsumerLatest? Much more descriptive than 1 or 2.

AndrewJSchofield · 2024-11-05T16:13:08Z

+    }
+
+    public enum ShareGroupAutoOffsetReset {
+        LATEST, EARLIEST, NONE;


This doesn't really have the value "NONE" does it? The KIP mentions only LATEST and EARLIEST I think.

apoorvmittal10

LGTM, agree with Andrew's comments.

adixitconfluent

Thanks for the PR. LGTM.

omkreddy · 2024-11-08T09:27:36Z

+    val send: Seq[TopicIdPartition] = Seq(topicIdPartition)
+
+    // Send the first share fetch request to initialize the share partition
+    var metadata: ShareRequestMetadata = new ShareRequestMetadata(memberId, ShareRequestMetadata.INITIAL_EPOCH)


looks like some of these lines are repeating, is it possible to extract a function and reuse the same.

omkreddy

@chirag-wadhwa5 Thanks for the PR. LGTM

…roups (apache#17573) This PR adds another dynamic config share.auto.offset.reset fir share groups. Reviewers: Andrew Schofield <[email protected]>, Apoorv Mittal <[email protected]>, Abhinav Dixit <[email protected]>, Manikumar Reddy <[email protected]>

KAFKA-16726: Added share.auto.offset.reset dynamic config for share g…

a58b079

…roups in GroupConfig

github-actions Bot added core Kafka Broker KIP-932 Queues for Kafka labels Oct 22, 2024

KAFKA-16726: Changed previous tests so they incorporate the new share…

fa79493

….auto.offset.reset changes and pass

chirag-wadhwa5 marked this pull request as ready for review October 23, 2024 18:36

apoorvmittal10 added the ci-approved label Oct 23, 2024

apoorvmittal10 requested review from AndrewJSchofield and apoorvmittal10 October 23, 2024 18:41

apoorvmittal10 requested changes Oct 23, 2024

View reviewed changes

AndrewJSchofield requested changes Oct 24, 2024

View reviewed changes

chirag-wadhwa5 added 2 commits October 24, 2024 20:46

KAFKA-16726: minor refactorings

7899138

Merge remote-tracking branch 'upstream/trunk' into KAFKA-16726

07d3a61

apoorvmittal10 requested changes Oct 25, 2024

View reviewed changes

KAFKA-16726

f5295c7

chirag-wadhwa5 requested review from AndrewJSchofield and apoorvmittal10 October 25, 2024 12:33

adixitconfluent reviewed Oct 25, 2024

View reviewed changes

Comment thread core/src/test/java/kafka/server/share/SharePartitionTest.java

Comment thread core/src/test/java/kafka/server/share/SharePartitionTest.java Outdated

Comment thread core/src/test/java/kafka/test/api/ShareConsumerTest.java Outdated

KAFKA-16726: Minor changes in SharePartitionTest

98b05a7

chirag-wadhwa5 requested a review from adixitconfluent October 28, 2024 09:00

apoorvmittal10 requested changes Oct 28, 2024

View reviewed changes

AndrewJSchofield requested review from mumrah and removed request for adixitconfluent October 28, 2024 12:00

AndrewJSchofield requested changes Oct 28, 2024

View reviewed changes

chirag-wadhwa5 added 7 commits November 4, 2024 11:29

Merge remote-tracking branch 'upstream/trunk' into KAFKA-16726

f6ba4bc

KAFKA-16726: Minor changes for better testing and code readability

b59d423

KAFKA-16726: Minor refactorings

d399a9f

KAFKA-16726: Minor changes in ShareConsumerTest

dc8fba2

KAFKA-16726: Enabled testShareFetchRequestToNonLeaderReplica in Share…

135bfc7

…FetchAcknowledgeRequestTest

KAFKA-16726: minor changes in import order

00ffa77

KAFKA-16726: minor changes in import order

37fc974

apoorvmittal10 reviewed Nov 4, 2024

View reviewed changes

chirag-wadhwa5 added 2 commits November 5, 2024 17:55

Merge remote-tracking branch 'upstream/trunk' into KAFKA-16726

94fc0cb

KAFKA-16726: error handling done in offsetForEarliestTimestamp and of…

a66bd02

…fsetForLatestTimestamp in ShareFetchUtils

AndrewJSchofield requested changes Nov 5, 2024

View reviewed changes

apoorvmittal10 approved these changes Nov 5, 2024

View reviewed changes

KAFKA-16726: Minor changes for better readability

2ee5faa

apoorvmittal10 approved these changes Nov 6, 2024

View reviewed changes

chirag-wadhwa5 requested a review from AndrewJSchofield November 6, 2024 14:16

AndrewJSchofield approved these changes Nov 7, 2024

View reviewed changes

adixitconfluent approved these changes Nov 7, 2024

View reviewed changes

omkreddy reviewed Nov 8, 2024

View reviewed changes

chirag-wadhwa5 added 2 commits November 11, 2024 06:27

Merge remote-tracking branch 'upstream/trunk' into KAFKA-16726

ed6e66b

KAFKA-16726: minor refactoring in ShareFetchAcknowledgeRequestTest

39327c1

omkreddy approved these changes Nov 11, 2024

View reviewed changes

omkreddy merged commit 9db5ed0 into apache:trunk Nov 11, 2024

peterxcli mentioned this pull request Dec 8, 2024

KAFKA-18014: Add duration based offset reset option for ShareConsumer #18096

Merged

3 tasks

		@@ -128,4 +130,17 @@ static long offsetForEarliestTimestamp(TopicIdPartition topicIdPartition, Replic
		Optional.empty(), true).timestampAndOffsetOpt();

Conversation

chirag-wadhwa5 commented Oct 22, 2024

Uh oh!

apoorvmittal10 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

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!

apoorvmittal10 Oct 28, 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!

Uh oh!

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!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

apoorvmittal10 Oct 28, 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!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

adixitconfluent left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

apoorvmittal10 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

apoorvmittal10 Oct 28, 2024 •

edited

Loading

apoorvmittal10 Oct 28, 2024 •

edited

Loading