KAFKA-2334 Guard against non-monotonic offsets in the client #5991
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package org.apache.kafka.common.errors; | ||
|
|
||
| /** | ||
| * Indicates that the leader is not able to guarantee monotonically increasing offsets | ||
| * due to a recent leader election and high-water mark lag | ||
| */ | ||
| public class OffsetNotAvailableException extends RetriableException { | ||
| private static final long serialVersionUID = 1L; | ||
|
|
||
| public OffsetNotAvailableException(String message) { | ||
| super(message); | ||
| } | ||
|
|
||
| public OffsetNotAvailableException(String message, Throwable cause) { | ||
| super(message, cause); | ||
| } | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -805,7 +805,8 @@ class Partition(val topicPartition: TopicPartition, | |
| def fetchOffsetForTimestamp(timestamp: Long, | ||
| isolationLevel: Option[IsolationLevel], | ||
| currentLeaderEpoch: Optional[Integer], | ||
| fetchOnlyFromLeader: Boolean, | ||
| isFromClient: Boolean): Option[TimestampAndOffset] = inReadLock(leaderIsrUpdateLock) { | ||
|
Contributor
There was a problem hiding this comment. I think we may be able to drop this argument. When the isolation level is read_uncommitted, then we are limited by the high watermark. For followers, isolation level will be |
||
| // decide whether to only fetch from leader | ||
| val localReplica = localReplicaWithEpochOrException(currentLeaderEpoch, fetchOnlyFromLeader) | ||
|
|
||
|
|
@@ -815,6 +816,25 @@ class Partition(val topicPartition: TopicPartition, | |
| case None => localReplica.logEndOffset.messageOffset | ||
| } | ||
|
|
||
| // Only actually check the HW if this is a client request and _not_ while unclean leader | ||
| // election is enabled | ||
| val shouldCheckHW = { | ||
| !logManager.currentDefaultConfig.uncleanLeaderElectionEnable && | ||
|
Contributor
There was a problem hiding this comment. I think we'd want to do this validation even if unclean leader election is enabled. The point that the KIP was making is that high watermark monotonicity can be violated if unclean leader election is enabled. But we'd still want to minimize such violations to cases when an unclean leader election has actually happened.
Member
Author
There was a problem hiding this comment. Thanks for the clarification, I misunderstood the KIP. I'll take this check out. |
||
| isFromClient && | ||
| leaderEpochStartOffsetOpt.isDefined | ||
| } | ||
|
|
||
| if(shouldCheckHW) { | ||
|
Contributor
There was a problem hiding this comment. nit: typically we prefer a space after |
||
| // Wait until the HW has caught up with the start offset from this epoch | ||
| if(leaderEpochStartOffsetOpt.get > localReplica.highWatermark.messageOffset) { | ||
| throw Errors.OFFSET_NOT_AVAILABLE.exception(s"Failed to fetch offsets for " + | ||
|
Contributor
There was a problem hiding this comment. Does this exception apply for all cases or only when the offset for latest timestamp is requested?
Contributor
There was a problem hiding this comment. That's a good question. Presumably it is possible for a timestamp query to find an offset between the log end offset and the high watermark as well. I guess the point is that we cannot trust the high watermark upper bound until we can ensure that it cannot have gone backwards. Maybe the only offset you can safely query without this validation is the earliest?
Member
Author
There was a problem hiding this comment. I think refusing to return any offset until the HW has caught up is a reasonable solution.
Contributor
There was a problem hiding this comment. Hmm seems like this behavior was changed then in 6a56493?
Member
Author
There was a problem hiding this comment. Indeed! After talking through the various cases with @hachikuji, we decided to only enforce the check on latest timestamp offset requests.
Contributor
There was a problem hiding this comment. Thanks! Would be good to document the change in the KIP, given the following is now out-dated: |
||
| s"partition $topicPartition with leader epoch ${currentLeaderEpoch.get} as this partition's " + | ||
| s"high-water mark (${localReplica.highWatermark.messageOffset}) is lagging behind its " + | ||
| s"LEO (${leaderEpochStartOffsetOpt.get}).") | ||
| } | ||
| } | ||
|
|
||
|
|
||
| if (timestamp == ListOffsetRequest.LATEST_TIMESTAMP) { | ||
| Some(new TimestampAndOffset(RecordBatch.NO_TIMESTAMP, lastFetchableOffset, Optional.of(leaderEpoch))) | ||
| } else { | ||
|
|
||
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Let's be specific and say that the new error code is OFFSET_NOT_AVAILABLE