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
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
44 changes: 28 additions & 16 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 @@ -772,9 +772,9 @@ static boolean hasUsableOffsetForLeaderEpochVersion(NodeApiVersions nodeApiVersi
*/
private void validateOffsetsAsync(Map<TopicPartition, SubscriptionState.FetchPosition> partitionsToValidate) {
final Map<Node, Map<TopicPartition, SubscriptionState.FetchPosition>> regrouped =
regroupFetchPositionsByLeader(partitionsToValidate);
regroupFetchPositionsByLeader(partitionsToValidate);

regrouped.forEach((node, fetchPostitions) -> {
regrouped.forEach((node, fetchPositions) -> {
if (node.isEmpty()) {
metadata.requestUpdate();
return;
Expand All @@ -788,18 +788,19 @@ 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);
offsetsForLeaderEpochClient.sendAsyncRequest(node, fetchPositions);

future.addListener(new RequestFutureListener<OffsetsForLeaderEpochClient.OffsetForEpochResult>() {
@Override
public void onSuccess(OffsetsForLeaderEpochClient.OffsetForEpochResult offsetsResult) {
Expand All @@ -812,13 +813,25 @@ 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(
SubscriptionState.FetchPosition requestPosition = fetchPositions.get(respTopicPartition);

if (respEndOffset.hasUndefinedEpochOrOffset()) {
if (subscriptions.hasDefaultOffsetResetPolicy()) {
Comment thread
abbccdda marked this conversation as resolved.
Outdated
log.info("Fetch offset {} is out of range for partition {}, resetting offset", requestPosition, respTopicPartition);
Comment thread
abbccdda marked this conversation as resolved.
Outdated
subscriptions.requestOffsetReset(respTopicPartition);
} else {
throw new OffsetOutOfRangeException(Collections.singletonMap(respTopicPartition, requestPosition.offset));
Comment thread
abbccdda marked this conversation as resolved.
Outdated
}
} 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.
Expand All @@ -828,7 +841,7 @@ public void onSuccess(OffsetsForLeaderEpochClient.OffsetForEpochResult offsetsRe

@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
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;
}
}
40 changes: 30 additions & 10 deletions clients/src/main/java/org/apache/kafka/common/requests/MetadataResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -371,7 +371,7 @@ public PartitionMetadata withoutLeaderEpoch() {
@Override
public String toString() {
return "PartitionMetadata(" +
", error=" + error +
"error=" + error +
", partition=" + topicPartition +
", leader=" + leaderId +
", leaderEpoch=" + leaderEpoch +
Expand Down Expand Up @@ -429,7 +429,8 @@ private Collection<TopicMetadata> createTopicMetadata(MetadataResponseData data)

public static MetadataResponse prepareResponse(int throttleTimeMs, Collection<Node> brokers, String clusterId,
int controllerId, List<TopicMetadata> topicMetadataList,
int clusterAuthorizedOperations) {
int clusterAuthorizedOperations,
short responseVersion) {
MetadataResponseData responseData = new MetadataResponseData();
responseData.setThrottleTimeMs(throttleTimeMs);
brokers.forEach(broker ->
Expand Down Expand Up @@ -464,22 +465,41 @@ public static MetadataResponse prepareResponse(int throttleTimeMs, Collection<No
}
responseData.topics().add(metadataResponseTopic);
});
return new MetadataResponse(responseData);
return new MetadataResponse(responseData.toStruct(responseVersion), responseVersion);

@abbccdda abbccdda Apr 22, 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.

Change the function signature to be able to pass in response version.

}

public static MetadataResponse prepareResponse(int throttleTimeMs, Collection<Node> brokers, String clusterId,
int controllerId, List<TopicMetadata> topicMetadataList) {
public static MetadataResponse prepareResponse(int throttleTimeMs,
Collection<Node> brokers,
String clusterId,
int controllerId,
List<TopicMetadata> topicMetadataList,
short responseVersion) {
return prepareResponse(throttleTimeMs, brokers, clusterId, controllerId, topicMetadataList,
MetadataResponse.AUTHORIZED_OPERATIONS_OMITTED);
MetadataResponse.AUTHORIZED_OPERATIONS_OMITTED, responseVersion);
}

public static MetadataResponse prepareResponse(Collection<Node> brokers,
String clusterId,
int controllerId,
List<TopicMetadata> topicMetadata,
short responseVersion) {
return prepareResponse(AbstractResponse.DEFAULT_THROTTLE_TIME, brokers, clusterId, controllerId,
topicMetadata, responseVersion);
}

public static MetadataResponse prepareResponse(Collection<Node> brokers, String clusterId, int controllerId,
public static MetadataResponse prepareResponse(Collection<Node> brokers,
String clusterId,
int controllerId,
List<TopicMetadata> topicMetadata) {
return prepareResponse(AbstractResponse.DEFAULT_THROTTLE_TIME, brokers, clusterId, controllerId, topicMetadata);
return prepareResponse(AbstractResponse.DEFAULT_THROTTLE_TIME, brokers, clusterId, controllerId,
topicMetadata, ApiKeys.METADATA.latestVersion());
}

public static MetadataResponse prepareResponse(int throttleTimeMs, List<MetadataResponseTopic> topicMetadataList,
Collection<Node> brokers, String clusterId, int controllerId,
public static MetadataResponse prepareResponse(int throttleTimeMs,
List<MetadataResponseTopic> topicMetadataList,
Collection<Node> brokers,
String clusterId,
int controllerId,
int clusterAuthorizedOperations) {
MetadataResponseData responseData = new MetadataResponseData();
responseData.setThrottleTimeMs(throttleTimeMs);
Expand Down
Loading