[jira] [Commented] (KAFKA-13194) LogCleaner may clean past highwatermark

["Justine Olshan (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-13194?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17397601#comment-17397601 ] 

Justine Olshan commented on KAFKA-13194:
----------------------------------------

There was code to change this in [https://github.com/apache/kafka/pull/9590,] but if we want to fix faster, we should just make a new PR. 

Replacing
 // we do not clean beyond the first unstable offset
 log.firstUnstableOffset,

with
 // we do not clean beyond the lastStableOffset (and therefore the high watermark)
 Option(log.lastStableOffset),

> LogCleaner may clean past highwatermark
> ---------------------------------------
>
>                 Key: KAFKA-13194
>                 URL: https://issues.apache.org/jira/browse/KAFKA-13194
>             Project: Kafka
>          Issue Type: Bug
>            Reporter: Lucas Bradstreet
>            Priority: Minor
>
> Here we have the cleaning point being bounded to the active segment base offset and the first unstable offset. Which makes sense:
>  
> {code:java}
>    // find first segment that cannot be cleaned
>     // neither the active segment, nor segments with any messages closer to the head of the log than the minimum compaction lag time
>     // may be cleaned
>     val firstUncleanableDirtyOffset: Long = Seq(      // we do not clean beyond the first unstable offset
>       log.firstUnstableOffset,      // the active segment is always uncleanable
>       Option(log.activeSegment.baseOffset),      // the first segment whose largest message timestamp is within a minimum time lag from now
>       if (minCompactionLagMs > 0) {
>         // dirty log segments
>         val dirtyNonActiveSegments = log.localNonActiveLogSegmentsFrom(firstDirtyOffset)
>         dirtyNonActiveSegments.find { s =>
>           val isUncleanable = s.largestTimestamp > now - minCompactionLagMs
>           debug(s"Checking if log segment may be cleaned: log='${log.name}' segment.baseOffset=${s.baseOffset} " +
>             s"segment.largestTimestamp=${s.largestTimestamp}; now - compactionLag=${now - minCompactionLagMs}; " +
>             s"is uncleanable=$isUncleanable")
>           isUncleanable
>         }.map(_.baseOffset)
>       } else None
>     ).flatten.min
> {code}
>  
> But LSO starts out as None.
> {code:java}
> @volatile private var firstUnstableOffsetMetadata: Option[LogOffsetMetadata] = None
> private[log] def firstUnstableOffset: Option[Long] = firstUnstableOffsetMetadata.map(_.messageOffset){code}
> For most code depending on the LSO, fetchLastStableOffsetMetadata is used to default it to the hwm if it's not set.
>  
> {code:java}
>   private def fetchLastStableOffsetMetadata: LogOffsetMetadata = {
>     checkIfMemoryMappedBufferClosed()    // cache the current high watermark to avoid a concurrent update invalidating the range check
>     val highWatermarkMetadata = fetchHighWatermarkMetadata    firstUnstableOffsetMetadata match {
>       case Some(offsetMetadata) if offsetMetadata.messageOffset < highWatermarkMetadata.messageOffset =>
>         if (offsetMetadata.messageOffsetOnly) {
>           lock synchronized {
>             val fullOffset = convertToOffsetMetadataOrThrow(offsetMetadata.messageOffset)
>             if (firstUnstableOffsetMetadata.contains(offsetMetadata))
>               firstUnstableOffsetMetadata = Some(fullOffset)
>             fullOffset
>           }
>         } else {
>           offsetMetadata
>         }
>       case _ => highWatermarkMetadata
>     }
>   }
> {code}
>  
>  
> This means that in the case where the hwm is prior to the active segment base, the log cleaner may clean past the hwm. This is most likely to occur after a broker restart when the log cleaner may start cleaning prior to replication becoming active.



--
This message was sent by Atlassian Jira
(v8.3.4#803005)