apoorvmittal10 commented on code in PR #17739:
URL: https://github.com/apache/kafka/pull/17739#discussion_r1842154706
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1602,8 +1602,6 @@ protected void
updateFetchOffsetMetadata(Optional<LogOffsetMetadata> fetchOffset
protected Optional<LogOffsetMetadata> fetchOffsetMetadata() {
lock.readLock().lock();
try {
- if (findNextFetchOffset.get())
Review Comment:
@adixitconfluent Here is my understanding: We want to keep
`fetchOffsetMetadata` in SharePartition so when there is no change to endOffset
then existing `fetchOffsetMetadata` can be used to calculate min bytes, correc?
For this, there could be couple of scenarios
- end offset is just increasing monotonically (usually), hence next fetch
offset shall point to new latest.
- there is reset of some messages and we need to determine next fetch offset
For first I see you reset the `fetchOffsetMetadata` whenever you change the
`endOffset`. For latter you use `findNextFetchOffset` always use Optional.empty
to be sent. So the existing code seems fine to me.
The current code solves mostly all but there could be an edge case where
only single message is released from between and the batch that holds that
released message might not be able to alone satisfy min bytes criteria, is that
something we are solving?
Can you confirm on my above understanding, if everything is true then the
approach I suggest to have local copy of intermediate offset and metadata in
delayed fetch as well. And change `updateFetchOffsetMetadata` in SharePartition
to `maybeUpdateFetchOffsetMetadata(Optional<LogOffsetMetadata>
fetchOffsetMetadata, long fetchOffset). The method caches the
`Optional<LogOffsetMetadata> fetchOffsetMetadata` only if `fetchOffset` matches
`endOffset`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
