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 1 commit
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
4 changes: 2 additions & 2 deletions 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,8 +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),
"detected log truncation");
super("Detected log truncation with diverging offsets " + divergentOffsets,
Utils.transformMap(divergentOffsets, Function.identity(), OffsetAndMetadata::offset));
this.divergentOffsets = Collections.unmodifiableMap(divergentOffsets);
}

Expand Down
3 changes: 1 addition & 2 deletions clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -199,8 +199,7 @@ public synchronized ConsumerRecords<K, V> poll(final Duration timeout) {
long position = subscriptions.position(entry.getKey()).offset;

if (beginningOffsets.get(entry.getKey()) != null && beginningOffsets.get(entry.getKey()) > position) {
throw new OffsetOutOfRangeException(Collections.singletonMap(entry.getKey(), position),
"beginning offset is greater than the current position");
throw new OffsetOutOfRangeException(Collections.singletonMap(entry.getKey(), position));
}

if (assignment().contains(entry.getKey()) && rec.offset() >= position) {
Expand Down
8 changes: 4 additions & 4 deletions 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,12 +31,12 @@ public class OffsetOutOfRangeException extends InvalidOffsetException {
private final Map<TopicPartition, Long> offsetOutOfRangePartitions;

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

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

Expand Down
54 changes: 32 additions & 22 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,8 +772,8 @@ 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 =
private void validateOffsetsAsync(Map<TopicPartition, FetchPosition> partitionsToValidate) {
final Map<Node, Map<TopicPartition, FetchPosition>> regrouped =
regroupFetchPositionsByLeader(partitionsToValidate);

regrouped.forEach((node, fetchPositions) -> {
Expand All @@ -798,12 +800,12 @@ private void validateOffsetsAsync(Map<TopicPartition, SubscriptionState.FetchPos

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

RequestFuture<OffsetsForLeaderEpochClient.OffsetForEpochResult> future =
RequestFuture<OffsetForEpochResult> future =
offsetsForLeaderEpochClient.sendAsyncRequest(node, fetchPositions);

future.addListener(new RequestFutureListener<OffsetsForLeaderEpochClient.OffsetForEpochResult>() {
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 @@ -817,10 +819,10 @@ public void onSuccess(OffsetsForLeaderEpochClient.OffsetForEpochResult offsetsRe
// 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 = fetchPositions.get(respTopicPartition);
FetchPosition requestPosition = fetchPositions.get(respTopicPartition);

if (respEndOffset.hasUndefinedEpochOrOffset()) {
handleOffsetOutOfRange(requestPosition.offset, respTopicPartition,
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 {
Expand All @@ -832,6 +834,7 @@ public void onSuccess(OffsetsForLeaderEpochClient.OffsetForEpochResult offsetsRe
});

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);
}
}
Expand Down Expand Up @@ -1118,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 @@ -1164,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 @@ -1197,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 @@ -1270,7 +1273,7 @@ private CompletedFetch initializeCompletedFetch(CompletedFetch nextCompletedFetc
log.debug("Discarding stale fetch response for partition {} since the fetched offset {} " +
"does not match the current offset {}", tp, fetchOffset, subscriptions.position(tp));
} else {
handleOffsetOutOfRange(fetchOffset, tp, "error response in offset fetch");
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 @@ -1310,16 +1313,23 @@ private CompletedFetch initializeCompletedFetch(CompletedFetch nextCompletedFetc
return completedFetch;
}

private void handleOffsetOutOfRange(long fetchOffset,
private void handleOffsetOutOfRange(FetchPosition fetchPosition,
TopicPartition topicPartition,
String reason) {
if (subscriptions.hasDefaultOffsetResetPolicy()) {
log.info("Fetch offset {} is out of range for partition {}, resetting offset",
topicPartition, fetchOffset);
log.info("Fetch offset epoch {} is out of range for partition {}, resetting offset",
fetchPosition, topicPartition);
subscriptions.requestOffsetReset(topicPartition);
} else {
throw new OffsetOutOfRangeException(Collections.singletonMap(
topicPartition, fetchOffset), reason);
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);
}
}

Expand Down
1 change: 1 addition & 0 deletions clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3833,6 +3833,7 @@ private void testOffsetValidationWithGivenEpochOffset(final EpochEndOffset epoch
subscriptions.seekUnvalidated(tp0, new SubscriptionState.FetchPosition(0, Optional.of(epochOne), leaderAndEpoch));

fetcher.validateOffsetsIfNeeded();

consumerClient.poll(time.timer(Duration.ZERO));
assertTrue(subscriptions.awaitingValidation(tp0));
assertTrue(client.hasInFlightRequests());
Expand Down