Skip to content

KAFKA-9840: Skip End Offset validation when the leader epoch is not reliable #8486

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

Merged
hachikuji merged 8 commits into from
Jun 5, 2020
Merged

KAFKA-9840: Skip End Offset validation when the leader epoch is not reliable #8486

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
fddb49e
avoid offset validation when the leader epoch is not reliable, and ch…
Apr 10, 2020
dbaa133
address comments
Jun 3, 2020
94e0911
address last comment
Jun 3, 2020
03407c4
fix further comments
Jun 3, 2020
a6c0785
address logging comments
Jun 3, 2020
6a66b1b
propagate exception in onSuccess callback
Jun 3, 2020
1a97ccc
fix KafkaConsumerTest
Jun 4, 2020
87ca234
remove unnecessary complete validation logic
Jun 5, 2020
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
6 changes: 4 additions & 2 deletions clients/src/main/java/org/apache/kafka/clients/Metadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -234,7 +234,9 @@ public synchronized void bootstrap(List<InetSocketAddress> addresses) {
}

/**
* Update metadata assuming the current request version. This is mainly for convenience in testing.
* Update metadata assuming the current request version.
*
* For testing only.
*/
public synchronized void updateWithCurrentRequestVersion(MetadataResponse response, boolean isPartialUpdate, long nowMs) {
this.update(this.requestVersion, response, isPartialUpdate, nowMs);
Expand Down Expand Up @@ -401,7 +403,7 @@ public synchronized void maybeThrowAnyException() {
* the producer to abort waiting for metadata if there were fatal exceptions (e.g. authentication failures)
* in the last metadata update.
*/
public synchronized void maybeThrowFatalException() {
protected synchronized void maybeThrowFatalException() {
KafkaException metadataException = this.fatalException;
if (metadataException != null) {
fatalException = null;
Expand Down
3 changes: 2 additions & 1 deletion clients/src/main/java/org/apache/kafka/clients/consumer/LogTruncationException.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,8 @@ public class LogTruncationException extends OffsetOutOfRangeException {
private final Map<TopicPartition, OffsetAndMetadata> divergentOffsets;

public LogTruncationException(Map<TopicPartition, OffsetAndMetadata> divergentOffsets) {
super(Utils.transformMap(divergentOffsets, Function.identity(), OffsetAndMetadata::offset));
super("Detected log truncation with diverging offsets " + divergentOffsets,
Utils.transformMap(divergentOffsets, Function.identity(), OffsetAndMetadata::offset));
this.divergentOffsets = Collections.unmodifiableMap(divergentOffsets);
}

Expand Down
7 changes: 6 additions & 1 deletion clients/src/main/java/org/apache/kafka/clients/consumer/OffsetOutOfRangeException.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,12 @@ public class OffsetOutOfRangeException extends InvalidOffsetException {
private final Map<TopicPartition, Long> offsetOutOfRangePartitions;

public OffsetOutOfRangeException(Map<TopicPartition, Long> offsetOutOfRangePartitions) {
super("Offsets out of range with no configured reset policy for partitions: " + offsetOutOfRangePartitions);
this("Offsets out of range with no configured reset policy for partitions: " +
offsetOutOfRangePartitions, offsetOutOfRangePartitions);
}

public OffsetOutOfRangeException(String message, Map<TopicPartition, Long> offsetOutOfRangePartitions) {
super(message);
this.offsetOutOfRangePartitions = offsetOutOfRangePartitions;
}

Expand Down
97 changes: 63 additions & 34 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,8 @@
import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
import org.apache.kafka.clients.consumer.OffsetOutOfRangeException;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
import org.apache.kafka.clients.consumer.internals.OffsetsForLeaderEpochClient.OffsetForEpochResult;
import org.apache.kafka.clients.consumer.internals.SubscriptionState.FetchPosition;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.IsolationLevel;
import org.apache.kafka.common.KafkaException;
Expand Down Expand Up @@ -476,7 +478,7 @@ public void resetOffsetsIfNeeded() {
}

/**
* Validate offsets for all assigned partitions for which a leader change has been detected.
* Validate offsets for all assigned partitions for which a leader change has been detected.
*/
public void validateOffsetsIfNeeded() {
RuntimeException exception = cachedOffsetForLeaderException.getAndSet(null);
Expand All @@ -490,7 +492,7 @@ public void validateOffsetsIfNeeded() {
});

// Collect positions needing validation, with backoff
Map<TopicPartition, SubscriptionState.FetchPosition> partitionsToValidate = subscriptions
Map<TopicPartition, FetchPosition> partitionsToValidate = subscriptions
.partitionsNeedingValidation(time.milliseconds())
.stream()
.collect(Collectors.toMap(Function.identity(), subscriptions::position));
Expand Down Expand Up @@ -672,15 +674,15 @@ private List<ConsumerRecord<K, V>> fetchRecords(CompletedFetch completedFetch, i
log.debug("Not returning fetched records for assigned partition {} since it is no longer fetchable",
completedFetch.partition);
} else {
SubscriptionState.FetchPosition position = subscriptions.position(completedFetch.partition);
FetchPosition position = subscriptions.position(completedFetch.partition);
if (completedFetch.nextFetchOffset == position.offset) {
List<ConsumerRecord<K, V>> partRecords = completedFetch.fetchRecords(maxRecords);

log.trace("Returning {} fetched records at offset {} for assigned partition {}",
partRecords.size(), position, completedFetch.partition);

if (completedFetch.nextFetchOffset > position.offset) {
SubscriptionState.FetchPosition nextPosition = new SubscriptionState.FetchPosition(
FetchPosition nextPosition = new FetchPosition(
completedFetch.nextFetchOffset,
completedFetch.lastEpoch,
position.currentLeader);
Expand Down Expand Up @@ -713,7 +715,7 @@ private List<ConsumerRecord<K, V>> fetchRecords(CompletedFetch completedFetch, i
}

private void resetOffsetIfNeeded(TopicPartition partition, OffsetResetStrategy requestedResetStrategy, ListOffsetData offsetData) {
SubscriptionState.FetchPosition position = new SubscriptionState.FetchPosition(
FetchPosition position = new FetchPosition(
offsetData.offset, offsetData.leaderEpoch, metadata.currentLeader(partition));
offsetData.leaderEpoch.ifPresent(epoch -> metadata.updateLastSeenEpochIfNewer(partition, epoch));
subscriptions.maybeSeekUnvalidated(partition, position.offset, requestedResetStrategy);
Expand Down Expand Up @@ -770,11 +772,11 @@ static boolean hasUsableOffsetForLeaderEpochVersion(NodeApiVersions nodeApiVersi
*
* Requests are grouped by Node for efficiency.
*/
private void validateOffsetsAsync(Map<TopicPartition, SubscriptionState.FetchPosition> partitionsToValidate) {
final Map<Node, Map<TopicPartition, SubscriptionState.FetchPosition>> regrouped =
regroupFetchPositionsByLeader(partitionsToValidate);
private void validateOffsetsAsync(Map<TopicPartition, FetchPosition> partitionsToValidate) {
final Map<Node, Map<TopicPartition, FetchPosition>> regrouped =
regroupFetchPositionsByLeader(partitionsToValidate);

regrouped.forEach((node, fetchPostitions) -> {
regrouped.forEach((node, fetchPositions) -> {
if (node.isEmpty()) {
metadata.requestUpdate();
return;
Expand All @@ -788,21 +790,22 @@ private void validateOffsetsAsync(Map<TopicPartition, SubscriptionState.FetchPos

if (!hasUsableOffsetForLeaderEpochVersion(nodeApiVersions)) {
log.debug("Skipping validation of fetch offsets for partitions {} since the broker does not " +
"support the required protocol version (introduced in Kafka 2.3)",
fetchPostitions.keySet());
for (TopicPartition partition : fetchPostitions.keySet()) {
"support the required protocol version (introduced in Kafka 2.3)",
fetchPositions.keySet());
for (TopicPartition partition : fetchPositions.keySet()) {
subscriptions.completeValidation(partition);
}
return;
}

subscriptions.setNextAllowedRetry(fetchPostitions.keySet(), time.milliseconds() + requestTimeoutMs);
subscriptions.setNextAllowedRetry(fetchPositions.keySet(), time.milliseconds() + requestTimeoutMs);

RequestFuture<OffsetsForLeaderEpochClient.OffsetForEpochResult> future =
offsetsForLeaderEpochClient.sendAsyncRequest(node, fetchPostitions);
future.addListener(new RequestFutureListener<OffsetsForLeaderEpochClient.OffsetForEpochResult>() {
RequestFuture<OffsetForEpochResult> future =
offsetsForLeaderEpochClient.sendAsyncRequest(node, fetchPositions);

future.addListener(new RequestFutureListener<OffsetForEpochResult>() {
@Override
public void onSuccess(OffsetsForLeaderEpochClient.OffsetForEpochResult offsetsResult) {
public void onSuccess(OffsetForEpochResult offsetsResult) {
Map<TopicPartition, OffsetAndMetadata> truncationWithoutResetPolicy = new HashMap<>();
if (!offsetsResult.partitionsToRetry().isEmpty()) {
subscriptions.setNextAllowedRetry(offsetsResult.partitionsToRetry(), time.milliseconds() + retryBackoffMs);
Expand All @@ -812,23 +815,33 @@ public void onSuccess(OffsetsForLeaderEpochClient.OffsetForEpochResult offsetsRe
// For each OffsetsForLeader response, check if the end-offset is lower than our current offset
// for the partition. If so, it means we have experienced log truncation and need to reposition
// that partition's offset.
//
// In addition, check whether the returned offset and epoch are valid. If not, then we should reset
// its offset if reset policy is configured, or throw out of range exception.
offsetsResult.endOffsets().forEach((respTopicPartition, respEndOffset) -> {
SubscriptionState.FetchPosition requestPosition = fetchPostitions.get(respTopicPartition);
Optional<OffsetAndMetadata> divergentOffsetOpt = subscriptions.maybeCompleteValidation(
FetchPosition requestPosition = fetchPositions.get(respTopicPartition);

if (respEndOffset.hasUndefinedEpochOrOffset()) {
handleOffsetOutOfRange(requestPosition, respTopicPartition,

@hachikuji hachikuji Jun 3, 2020

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.

Didn't notice this before, but this is handled inside a loop. If one partition hits an error, then the raised exception will prevent us from completing the validation for other partitions.

"Failed leader offset epoch validation for " + respEndOffset
Comment thread
abbccdda marked this conversation as resolved.
Outdated
+ " since no end offset larger than current fetch epoch was reported");
} else {
Optional<OffsetAndMetadata> divergentOffsetOpt = subscriptions.maybeCompleteValidation(
respTopicPartition, requestPosition, respEndOffset);
divergentOffsetOpt.ifPresent(divergentOffset -> {
truncationWithoutResetPolicy.put(respTopicPartition, divergentOffset);
});
divergentOffsetOpt.ifPresent(
divergentOffset -> truncationWithoutResetPolicy.put(respTopicPartition, divergentOffset));
}
});

if (!truncationWithoutResetPolicy.isEmpty()) {
Comment thread
abbccdda marked this conversation as resolved.
log.error("Detected log truncation with diverging offsets " + truncationWithoutResetPolicy);
Comment thread
abbccdda marked this conversation as resolved.
Outdated
throw new LogTruncationException(truncationWithoutResetPolicy);
}
}

@Override
public void onFailure(RuntimeException e) {
subscriptions.requestFailed(fetchPostitions.keySet(), time.milliseconds() + retryBackoffMs);
subscriptions.requestFailed(fetchPositions.keySet(), time.milliseconds() + retryBackoffMs);
metadata.requestUpdate();

if (!(e instanceof RetriableException) && !cachedOffsetForLeaderException.compareAndSet(null, e)) {
Expand Down Expand Up @@ -966,7 +979,6 @@ public void onSuccess(ClientResponse response, RequestFuture<ListOffsetResult> f
* value of each partition may be null only for v0. In v1 and later the ListOffset API would not
* return a null timestamp (-1 is returned instead when necessary).
*/
@SuppressWarnings("deprecation")
private void handleListOffsetResponse(Map<TopicPartition, ListOffsetRequest.PartitionData> timestampsToSearch,
ListOffsetResponse listOffsetResponse,
RequestFuture<ListOffsetResult> future) {
Expand Down Expand Up @@ -1051,12 +1063,12 @@ static class ListOffsetResult {
private final Map<TopicPartition, ListOffsetData> fetchedOffsets;
private final Set<TopicPartition> partitionsToRetry;

public ListOffsetResult(Map<TopicPartition, ListOffsetData> fetchedOffsets, Set<TopicPartition> partitionsNeedingRetry) {
ListOffsetResult(Map<TopicPartition, ListOffsetData> fetchedOffsets, Set<TopicPartition> partitionsNeedingRetry) {
this.fetchedOffsets = fetchedOffsets;
this.partitionsToRetry = partitionsNeedingRetry;
}

public ListOffsetResult() {
ListOffsetResult() {
this.fetchedOffsets = new HashMap<>();
this.partitionsToRetry = new HashSet<>();
}
Expand Down Expand Up @@ -1109,7 +1121,7 @@ private Map<Node, FetchSessionHandler.FetchRequestData> prepareFetchRequests() {

for (TopicPartition partition : fetchablePartitions()) {
// Use the preferred read replica if set, or the position's leader
SubscriptionState.FetchPosition position = this.subscriptions.position(partition);
FetchPosition position = this.subscriptions.position(partition);
Optional<Node> leaderOpt = position.currentLeader.leader;
if (!leaderOpt.isPresent()) {
metadata.requestUpdate();
Expand Down Expand Up @@ -1155,8 +1167,8 @@ private Map<Node, FetchSessionHandler.FetchRequestData> prepareFetchRequests() {
return reqs;
}

private Map<Node, Map<TopicPartition, SubscriptionState.FetchPosition>> regroupFetchPositionsByLeader(
Map<TopicPartition, SubscriptionState.FetchPosition> partitionMap) {
private Map<Node, Map<TopicPartition, FetchPosition>> regroupFetchPositionsByLeader(
Map<TopicPartition, FetchPosition> partitionMap) {
return partitionMap.entrySet()
.stream()
.filter(entry -> entry.getValue().currentLeader.leader.isPresent())
Expand Down Expand Up @@ -1188,7 +1200,7 @@ private CompletedFetch initializeCompletedFetch(CompletedFetch nextCompletedFetc
} else if (error == Errors.NONE) {
// we are interested in this fetch only if the beginning offset matches the
// current consumed position
SubscriptionState.FetchPosition position = subscriptions.position(tp);
FetchPosition position = subscriptions.position(tp);
if (position == null || position.offset != fetchOffset) {
log.debug("Discarding stale fetch response for partition {} since its offset {} does not match " +
"the expected offset {}", tp, fetchOffset, position);
Expand Down Expand Up @@ -1260,11 +1272,8 @@ private CompletedFetch initializeCompletedFetch(CompletedFetch nextCompletedFetc
if (fetchOffset != subscriptions.position(tp).offset) {
log.debug("Discarding stale fetch response for partition {} since the fetched offset {} " +
"does not match the current offset {}", tp, fetchOffset, subscriptions.position(tp));
} else if (subscriptions.hasDefaultOffsetResetPolicy()) {
log.info("Fetch offset {} is out of range for partition {}, resetting offset", fetchOffset, tp);
subscriptions.requestOffsetReset(tp);
} else {
throw new OffsetOutOfRangeException(Collections.singletonMap(tp, fetchOffset));
handleOffsetOutOfRange(subscriptions.position(tp), tp, "error response in offset fetch");
}
} else {
log.debug("Unset the preferred read replica {} for partition {} since we got {} when fetching {}",
Expand Down Expand Up @@ -1304,6 +1313,26 @@ private CompletedFetch initializeCompletedFetch(CompletedFetch nextCompletedFetc
return completedFetch;
}

private void handleOffsetOutOfRange(FetchPosition fetchPosition,
TopicPartition topicPartition,
String reason) {
if (subscriptions.hasDefaultOffsetResetPolicy()) {
log.info("Fetch offset epoch {} is out of range for partition {}, resetting offset",
fetchPosition, topicPartition);
subscriptions.requestOffsetReset(topicPartition);
} else {
Map<TopicPartition, Long> offsetOutOfRangePartitions =
Collections.singletonMap(topicPartition, fetchPosition.offset);
String errorMessage = String.format("Offsets out of range " +
"with no configured reset policy for partitions: %s" +
", for fetch offset: %d, " +
"root cause: %s",
offsetOutOfRangePartitions, fetchPosition.offset, reason);
log.error(errorMessage);
Comment thread
abbccdda marked this conversation as resolved.
Outdated
throw new OffsetOutOfRangeException(errorMessage, offsetOutOfRangePartitions);
}
}

/**
* Parse the record entry, deserializing the key / value fields if necessary
*/
Expand Down
4 changes: 0 additions & 4 deletions ...rc/main/java/org/apache/kafka/clients/consumer/internals/OffsetsForLeaderEpochClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,10 +85,6 @@ protected OffsetForEpochResult handleResponse(
case KAFKA_STORAGE_ERROR:
case OFFSET_NOT_AVAILABLE:
case LEADER_NOT_AVAILABLE:
logger().debug("Attempt to fetch offsets for partition {} failed due to {}, retrying.",
Comment thread
abbccdda marked this conversation as resolved.
Outdated
topicPartition, error);
partitionsToRetry.add(topicPartition);
break;
case FENCED_LEADER_EPOCH:
case UNKNOWN_LEADER_EPOCH:
logger().debug("Attempt to fetch offsets for partition {} failed due to {}, retrying.",
Expand Down
15 changes: 9 additions & 6 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -786,7 +786,14 @@ private boolean maybeValidatePosition(Metadata.LeaderAndEpoch currentLeaderAndEp
return false;
}

if (!currentLeaderAndEpoch.leader.isPresent() && !currentLeaderAndEpoch.epoch.isPresent()) {
if (!currentLeaderAndEpoch.leader.isPresent()) {
return false;
}

// If the leader id is known but the epoch is unknown, we must either have an older API or the epoch has been
// discarded. If so, skip the validation and begin fetching.
if (!currentLeaderAndEpoch.epoch.isPresent()) {
completeValidation();
return false;
}

Expand Down Expand Up @@ -818,9 +825,7 @@ private void validatePosition(FetchPosition position) {
*/
private void completeValidation() {
if (hasPosition()) {
transitionState(FetchStates.FETCHING, () -> {
this.nextRetryTimeMs = null;
});
transitionState(FetchStates.FETCHING, () -> this.nextRetryTimeMs = null);
}
}

Expand Down Expand Up @@ -1011,8 +1016,6 @@ public boolean hasValidPosition() {
*
* This includes the offset and epoch from the last record in
* the batch from a FetchResponse. It also includes the leader epoch at the time the batch was consumed.
*
* The last fetch epoch is used to
*/
public static class FetchPosition {
public final long offset;
Expand Down
9 changes: 6 additions & 3 deletions clients/src/main/java/org/apache/kafka/common/requests/EpochEndOffset.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,9 +40,7 @@ public EpochEndOffset(Errors error, int leaderEpoch, long endOffset) {
}

public EpochEndOffset(int leaderEpoch, long endOffset) {
this.error = Errors.NONE;
this.leaderEpoch = leaderEpoch;
this.endOffset = endOffset;
this(Errors.NONE, leaderEpoch, endOffset);
}

public Errors error() {
Expand Down Expand Up @@ -86,4 +84,9 @@ public boolean equals(Object o) {
public int hashCode() {
return Objects.hash(error, leaderEpoch, endOffset);
}

public boolean hasUndefinedEpochOrOffset() {
return this.endOffset == UNDEFINED_EPOCH_OFFSET ||

@guozhangwang guozhangwang Apr 23, 2020

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.

For my own understanding: if endOffset is UNDEFINED the epoch should always be UNDEFINED too? If that's the case we can just rely on leaderEpoch alone?

@abbccdda abbccdda Apr 23, 2020

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

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

I just feel this check is stronger if later we change the behavior on broker.

@hachikuji hachikuji Jun 2, 2020

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.

Older versions did not return the epoch, so it was possible to see an offset defined without an epoch. However, the version that the consumer relies on should always have both or neither. Anyway, I think it is reasonable to be a little stricter here.

this.leaderEpoch == UNDEFINED_EPOCH;
}
}
Loading