Skip to content

KAFKA-16452: Bound high-watermark offset to range between LLSO and LEO #15634

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
kamalcph wants to merge 1 commit into from
Closed

KAFKA-16452: Bound high-watermark offset to range between LLSO and LEO #15634

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 15 additions & 9 deletions core/src/main/scala/kafka/log/UnifiedLog.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,17 +136,17 @@ class UnifiedLog(@volatile var logStartOffset: Long,
*/
@volatile private var firstUnstableOffsetMetadata: Option[LogOffsetMetadata] = None

@volatile private[kafka] var _localLogStartOffset: Long = logStartOffset

/* Keep track of the current high watermark in order to ensure that segments containing offsets at or above it are
* not eligible for deletion. This means that the active segment is only eligible for deletion if the high watermark
* equals the log end offset (which may never happen for a partition under consistent load). This is needed to
* prevent the log start offset (which is exposed in fetch responses) from getting ahead of the high watermark.
*/
@volatile private var highWatermarkMetadata: LogOffsetMetadata = new LogOffsetMetadata(logStartOffset)
@volatile private var highWatermarkMetadata: LogOffsetMetadata = new LogOffsetMetadata(_localLogStartOffset)

@satishd satishd Apr 1, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There won't be any effect with this change as _localLogStartOffset is initialized with logStartOffset. But it is good to keep _localLogStartOffset for consistency and relevance of this field.


@volatile var partitionMetadataFile: Option[PartitionMetadataFile] = None

@volatile private[kafka] var _localLogStartOffset: Long = logStartOffset

def localLogStartOffset(): Long = _localLogStartOffset

// This is the offset(inclusive) until which segments are copied to the remote storage.
Expand Down Expand Up @@ -269,7 +269,7 @@ class UnifiedLog(@volatile var logStartOffset: Long,

/**
* Update the high watermark to a new offset. The new high watermark will be lower
* bounded by the log start offset and upper bounded by the log end offset.
* bounded by the local-log-start-offset and upper bounded by the log-end-offset.
*
* This is intended to be called by the leader when initializing the high watermark.
*
Expand All @@ -282,15 +282,15 @@ class UnifiedLog(@volatile var logStartOffset: Long,

/**
* Update high watermark with offset metadata. The new high watermark will be lower
* bounded by the log start offset and upper bounded by the log end offset.
* bounded by the local-log-start-offset and upper bounded by the log-end-offset.
*
* @param highWatermarkMetadata the suggested high watermark with offset metadata
* @return the updated high watermark offset
*/
def updateHighWatermark(highWatermarkMetadata: LogOffsetMetadata): Long = {
val endOffsetMetadata = localLog.logEndOffsetMetadata
val newHighWatermarkMetadata = if (highWatermarkMetadata.messageOffset < logStartOffset) {
new LogOffsetMetadata(logStartOffset)
val newHighWatermarkMetadata = if (highWatermarkMetadata.messageOffset < _localLogStartOffset) {

@junrao junrao Apr 8, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm, when will we set HWM to be lower than _localLogStartOffset?

In UnifiedLog.deletableSegments(), we have the following code that bounds the retention based deletion by highWatermark. When updating highWatermark, the value typically increases.
val predicateResult = highWatermark >= upperBoundOffset && predicate(segment, nextSegmentOpt)

@kamalcph kamalcph Apr 9, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

when will we set HWM to be lower than _localLogStartOffset?

This can happen when recovering the partition due to ungraceful shutdown and the replication-offset-checkpoint file is missing/corrupted. When the broker comes online, HWM is set to to localLogStartOffset in UnifiedLog#updateLocalLogStartOffset, then we load the HWM from the checkpoint file in Partition#createLog.

If the HWM checkpoint file is missing / does not contain the entry for partition, then the default value of 0 is taken. If 0 < LogStartOffset (LSO), then LSO is assumed as HWM . Thus, the non-monotonic update of highwatermark from LLSO to LSO can happen.

@junrao junrao Apr 12, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, if replication-offset-checkpoint is corrupted, HWM could temporarily be set to below local-log-start-offset. I am still trying to understand the impact of that. In the common case, the restarted broker can't become the leader or serve reads until it's caught up. At that time, the HWM will be up to date. In the rare case, the restarted broker is elected as the leader before caught up through unclean election. Is this the case that you want to address?

The jira also says:

If the high watermark is less than the local-log-start-offset, then the UnifiedLog#fetchHighWatermarkMetadata method will throw the OFFSET_OUT_OF_RANGE error when it converts the offset to metadata. Once this error happens, the followers will receive out-of-range exceptions and the producers won't be able to produce messages since the leader cannot move the high watermark.

However, the follower read is bounded by logEndOffset, not HWM? Where does the follower read need to convert HWM to metadata?

@kamalcph kamalcph Apr 13, 2024 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In the rare case, the restarted broker is elected as the leader before caught up through unclean election. Is this the case that you want to address?

yes, we want to address this case too. And, the issue can also happen during clean preferred-leader-election:

Call stack: The replica (1002) has full data but HW is invalid, then the fetch-offset will be equal to LeaderLog(1001).highWatermark

Leader (1001):
    KafkaApis.handleFetchRequest
        ReplicaManager.fetchMessages
            ReplicaManager.readFromLocalLog
                Partition.fetchRecords
                    Partition.updateFollowerFetchState
                        Partition.maybeExpandIsr
                            Partition.submitAlterPartition
                            ...
                            ...
                            ...
        # If there is not enough data to respond and there is no remote data, we will let the fetch request wait for new data.
        # parks the request in the DelayedFetchPurgatory


Another thread, runs Preferred-Leader-Election in controller (1003), since the replica 1002 joined the ISR list, it can be elected as the preferred leader. The controller sends LeaderAndIsr requests to all the brokers.

    KafkaController.processReplicaLeaderElection
        KafkaController.onReplicaElection
            PartitionStateMachine.handleStateChanges
                PartitionStateMachine.doHandleStateChanges
                    PartitionStateMachine.electLeaderForPartitions
                ControllerChannelManager.sendRequestsToBrokers


Replica 1002 got elected as Leader and have invalid highWatermark since it didn't process the fetch-response from the previous leader 1001, throws OFFSET_OUT_OF_RANGE error when processing the LeaderAndIsr request. Note that in LeaderAndIsr request even if one partition fails, then the remaining partitions in that request won't be processed.

    KafkaApis.handleLeaderAndIsrRequest
        ReplicaManager.becomeLeaderOrFollower
            ReplicaManager.makeLeaders
                Partition.makeLeader
                    Partition.maybeIncrementLeaderHW
                        UnifiedLog.maybeIncrementHighWatermark (LeaderLog)
                            UnifiedLog.fetchHighWatermarkMetadata


The controller assumes that the current-leader for the tp0 is 1002, but the broker 1002 couldn't process the LISR. The controller retries the LISR until the broker 1002 becomes leader for tp0. During this time, the producers won't be able to send messages, as the node 1002, sends NOT_LEADER_FOR_PARTITION error-code to the producer.

During this time, if a follower sends the FETCH request to read from the current-leader 1002, then OFFSET_OUT_OF_RANGE error will be returned by the leader:

 KafkaApis.handleFetchRequest
      ReplicaManager.fetchMessages
          ReplicaManager.readFromLog
              Partition.fetchRecords
                  # readFromLocalLog
                  Partition.updateFollowerFetchState
                      Partition.maybeIncrementLeaderHW
                           LeaderLog.maybeIncrementHighWatermark
                              UnifiedLog.fetchHighWatermarkMetadata
                                  UnifiedLog.convertToOffsetMetadataOrThrow
                                      LocalLog.convertToOffsetMetadataOrThrow
                                          LocalLog.read
                                              # OffsetOutOfRangeException exception

@junrao junrao Apr 16, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the detailed explanation.

For the makeLeaders path, it will call UnifiedLog.convertToOffsetMetadataOrThrow. Within it, checkLogStartOffset(offset) shouldn't throw OFFSET_OUT_OF_RANGE since we are comparing the offset with logStartOffset. Do you know which part throws OFFSET_OUT_OF_RANGE error?

For the follower fetch path, it's bounded by LogEndOffset. So it shouldn't need to call UnifiedLog.fetchHighWatermarkMetadata, right? The regular consumer will call UnifiedLog.fetchHighWatermarkMetadata.

@junrao junrao Apr 17, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@kamalcph : Thanks for the explanation. I understand the problem now.

As for the fix, it seems that it could work for HWM. However, I am not sure that we could always do the same thing of LastStableOffset. For example, if we lose the local data in all replicas, the lastStableOffset could still be in the middle of a tiered segment and moving it to localLogStartOffset immediately will be incorrect.

Here is another potential approach. Note that OffsetMetadata (segmentBaseOffset and relativePositionInSegment) is only used in DelayedFetch for estimating the amount of available bytes. If occasionally OffsetMetadata is not available, we don't have to force an exception in convertToOffsetMetadataOrThrow(). Instead, we can leave the OffsetMetadata as empty and just use a conservative 1 byte for estimating the amount of available bytes. This approach will apply to both HWM and LSO. The inaccurate byte estimate will be ok as long as it's infrequent. What do you think?

@kamalcph kamalcph Apr 18, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for suggesting the alternative approach. I'll check and comeback on this.

@kamalcph kamalcph Apr 28, 2024 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For example, if we lose the local data in all replicas, the lastStableOffset could still be in the middle of a tiered segment and moving it to localLogStartOffset immediately will be incorrect.

I'm not clear on this:

  1. Segments that are eligible for upload to remote storage only when the lastStableOffset moves beyond the segment-to-be-uploaded-end-offset.
  2. When all the replicas loses local data (offline partition), then we consider the data in remote storage also lost. Currently, for this case, we don't have provision to serve the remote data.
  3. When firstUnstableOffsetMetadata is empty, we return highWatermark. With this patch, the highWatermark lower boundary is set to localLogStartOffset so there won't be an issue.

Note that OffsetMetadata (segmentBaseOffset and relativePositionInSegment) is only used in DelayedFetch for estimating the amount of available bytes.

The LogOffsetMetadata#onOlderSegment method is used in the hot-path of incrementing the high-watermark and expects the full metadata, otherwise it throws an error. Is it ok to remove the throwable from LogOffsetMetadata#onOlderSegment method and return false when messageOffsetOnly available?

@kamalcph kamalcph Apr 28, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Opened #15825 a draft PR with the suggested approach. PTAL.

@junrao junrao May 1, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not clear on this:

  1. Segments that are eligible for upload to remote storage only when the lastStableOffset moves beyond the segment-to-be-uploaded-end-offset.
  2. When all the replicas loses local data (offline partition), then we consider the data in remote storage also lost. Currently, for this case, we don't have provision to serve the remote data.
  3. When firstUnstableOffsetMetadata is empty, we return highWatermark. With this patch, the highWatermark lower boundary is set to localLogStartOffset so there won't be an issue.

That's true. It's just that that is yet another offset that we need to bound. I am also not sure if there are other side effects of adjusting HWM and LSO.

Left some comments on #15825.

new LogOffsetMetadata(_localLogStartOffset)
} else if (highWatermarkMetadata.messageOffset >= endOffsetMetadata.messageOffset) {
endOffsetMetadata
} else {
Expand Down Expand Up @@ -332,7 +332,7 @@ class UnifiedLog(@volatile var logStartOffset: Long,

/**
* Update high watermark with a new value. The new high watermark will be lower
* bounded by the log start offset and upper bounded by the log end offset.
* bounded by the local-log-start-offset and upper bounded by the log-end-offset.
*
* This method is intended to be used by the follower to update its high watermark after
* replication from the leader.
Expand Down Expand Up @@ -1224,6 +1224,12 @@ class UnifiedLog(@volatile var logStartOffset: Long,
s"but we only have log segments starting from offset: $logStartOffset.")
}

private def checkLocalLogStartOffset(offset: Long): Unit = {

@chia7712 chia7712 Apr 16, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It seems reading records between [logStartOffset, localLogStartOffset] is dangerous since the segment won't be in local-disk. That is a bit chaos to me as UnifiedLog presents a unified view of local and tiered log segment (

kafka/core/src/main/scala/kafka/log/UnifiedLog.scala

Line 59 in fccd7fe

* A log which presents a unified view of local and tiered log segments.
). The check looks like a limit that we can't "view" data from tiered log segment.

@kamalcph kamalcph Apr 17, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agree on this. The checkLocalLogStartOffset is used only in the convertToOffsetMetadataOrThrow method which reads from local-disk.

if (offset < localLogStartOffset())
throw new OffsetOutOfRangeException(s"Received request for offset $offset for partition $topicPartition, " +
s"but we only have local-log segments starting from offset: ${localLogStartOffset()}.")
}

/**
* Read messages from the log.
*
Expand Down Expand Up @@ -1428,7 +1434,7 @@ class UnifiedLog(@volatile var logStartOffset: Long,
* If the message offset is out of range, throw an OffsetOutOfRangeException
*/
private def convertToOffsetMetadataOrThrow(offset: Long): LogOffsetMetadata = {
checkLogStartOffset(offset)
checkLocalLogStartOffset(offset)
localLog.convertToOffsetMetadataOrThrow(offset)
}

Expand Down
74 changes: 74 additions & 0 deletions core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -318,6 +318,80 @@ class UnifiedLogTest {
assertHighWatermark(4L)
}

@Test
def testHighWatermarkMaintenanceForRemoteTopic(): Unit = {
val logConfig = LogTestUtils.createLogConfig(segmentBytes = 1024 * 1024, remoteLogStorageEnable = true)
val log = createLog(logDir, logConfig, remoteStorageSystemEnable = true)
val leaderEpoch = 0

def assertHighWatermark(offset: Long): Unit = {
assertEquals(offset, log.highWatermark)
assertValidLogOffsetMetadata(log, log.fetchOffsetSnapshot.highWatermark)
}

// High watermark initialized to 0
assertHighWatermark(0L)

var offset = 0L
for(_ <- 0 until 50) {
val records = TestUtils.singletonRecords("test".getBytes())
val info = log.appendAsLeader(records, leaderEpoch)
offset = info.lastOffset
if (offset != 0 && offset % 10 == 0)
log.roll()
}
assertEquals(5, log.logSegments.size)

// High watermark not changed by append
assertHighWatermark(0L)

// Update high watermark as leader
log.maybeIncrementHighWatermark(new LogOffsetMetadata(50L))
assertHighWatermark(50L)
assertEquals(50L, log.logEndOffset)

// Cannot update high watermark past the log end offset
log.updateHighWatermark(60L)
assertHighWatermark(50L)

// simulate calls to upload 3 segments to remote storage and remove them from local-log.
log.updateHighestOffsetInRemoteStorage(30)
log.maybeIncrementLocalLogStartOffset(31L, LogStartOffsetIncrementReason.SegmentDeletion)
log.deleteOldSegments()
assertEquals(2, log.logSegments.size)
assertEquals(31L, log.localLogStartOffset())
assertHighWatermark(50L)

// simulate one remote-log segment deletion
val logStartOffset = 11L
log.maybeIncrementLogStartOffset(logStartOffset, LogStartOffsetIncrementReason.SegmentDeletion)
assertEquals(11, log.logStartOffset)

// Updating the HW below the log-start-offset / local-log-start-offset is not allowed. HW should reset to local-log-start-offset.
log.updateHighWatermark(new LogOffsetMetadata(5L))
assertHighWatermark(31L)
// Updating the HW between log-start-offset and local-log-start-offset is not allowed. HW should reset to local-log-start-offset.

@junrao junrao Apr 12, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is moving HW below local-log-start-offset, not log-start-offset.

log.updateHighWatermark(new LogOffsetMetadata(25L))
assertHighWatermark(31L)
// Updating the HW between local-log-start-offset and log-end-offset is allowed.
log.updateHighWatermark(new LogOffsetMetadata(32L))
assertHighWatermark(32L)
assertEquals(11L, log.logStartOffset)
assertEquals(31L, log.localLogStartOffset())

// Truncating the logs to below the local-log-start-offset, should update the high watermark

@satishd satishd Apr 1, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good to see covering the truncation scenarios also.

log.truncateTo(15L)
assertHighWatermark(15L)
assertEquals(15L, log.logStartOffset)
assertEquals(15L, log.localLogStartOffset())

// Truncate the logs to below the log-start-offset, should update the high watermark
log.truncateFullyAndStartAt(10L)
assertHighWatermark(10L)
assertEquals(10L, log.logStartOffset)
assertEquals(10L, log.localLogStartOffset())
}

private def assertNonEmptyFetch(log: UnifiedLog, offset: Long, isolation: FetchIsolation): Unit = {
val readInfo = log.read(startOffset = offset,
maxLength = Int.MaxValue,
Expand Down