jolshan commented on a change in pull request #9590:
URL: https://github.com/apache/kafka/pull/9590#discussion_r523231421
##########
File path: core/src/main/scala/kafka/log/LogCleaner.scala
##########
@@ -711,6 +723,9 @@ private[log] class Cleaner(val id: Int,
shallowOffsetOfMaxTimestamp = result.shallowOffsetOfMaxTimestamp,
records = retained)
throttler.maybeThrottle(outputBuffer.limit())
+ if (newCanUpdateBaseOffset)
+ dest.updateBaseOffset(result.minOffset())
+ newCanUpdateBaseOffset = false
Review comment:
Yeah. I'm not totally happy with how this is handled, but I'm not sure
if there is another better way. So the new cleaned segment is created in
cleanSegments which calls this method. The issue is that multiple segments can
be compacted into a single segment and each time a segment is compacted this
method is called. Thus, this method does not always know if there is a record
in the new segment yet. This is why we have this extra boolean variable.
This is also why delaying might not work without a variable like this. The
method will not know if the new segment has been created already or not.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
