KAFKA-9840: Skip End Offset validation when the leader epoch is not reliable by abbccdda · Pull Request #8486 · apache/kafka

abbccdda · 2020-04-14T17:03:56Z

This PR provides two fixes:

Complete offset validation if the leader epoch is not reliable
Refresh metadata when the return offset or epoch is not defined

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

abbccdda · 2020-04-22T02:56:27Z

Move the function closer to its caller.

abbccdda · 2020-04-22T02:56:58Z

Change one: immediately complete the validation when the leader epoch is not reliable.

Could you not move the function while changing it so that the diff is easier to check -- e.g. here I'd have to blindly trust you that there are only changes of this func :)

Np, will do

abbccdda · 2020-04-22T02:57:44Z

Change two: do not complete the validation as the returned epoch or offset is invalid.

abbccdda · 2020-04-22T02:58:58Z

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

abbccdda · 2020-04-22T03:01:36Z

Added test coverage L203 and L214, other changes are just side cleanups and signature refactoring in this file.

guozhangwang

Made a pass over non-testing part. I'd suggest we revert the function movement so that it helps with more clear reviews.

guozhangwang · 2020-04-23T00:07:38Z

Could you not move the function while changing it so that the diff is easier to check -- e.g. here I'd have to blindly trust you that there are only changes of this func :)

guozhangwang · 2020-04-23T00:10:48Z

+                    offsetsResult.endOffsets().forEach((respTopicPartition, respEndOffset) -> {
+                        if (respEndOffset.hasUndefinedEpochOrOffset()) {
+                            // Should attempt to find the new leader in the next try.
+                            log.debug("Requesting metadata update for partition {} due to undefined epoch or offset {}",


nit: ... or offset {} from OffsetsForLeaderEpoch response

guozhangwang · 2020-04-23T00:12:45Z

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


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?

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

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.

guozhangwang

I think @hachikuji can take another look at this.

ijuma · 2020-05-17T16:46:19Z

Where are with this?

abbccdda · 2020-05-21T17:41:00Z

@ijuma will ping Jason for another review

hachikuji · 2020-05-27T21:25:02Z

I was waiting for #8376 to get merged. I will review this shortly.

hachikuji

@abbccdda Sorry for the delay. I think what we decided is that if the current leader epoch is not known, then we would not attempt to send OffsetsForLeaderEpoch because we cannot trust the result. What I had in mind was a little bit simpler, but I could be missing something. Inside SubscriptionState.TopicPartitionState.maybeValidatePosition, we have the following check:

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

What I was thinking is that we could do something like the following:

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

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

That is, if we know the leader, but we don't have an epoch, then we must either have an older api or we must have discarded the value. Either way, we skip validation and begin fetching. Would that not work?

…eck epoch/offset in response

hachikuji · 2020-06-02T22:23:36Z

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


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.

hachikuji · 2020-06-03T01:23:00Z

retest this please

hachikuji · 2020-06-03T01:23:10Z

retest this please

hachikuji · 2020-06-03T01:23:24Z

retest this please

hachikuji · 2020-06-03T16:05:31Z

retest this please

hachikuji · 2020-06-03T16:05:43Z

retest this please

hachikuji · 2020-06-03T16:14:37Z


                        if (respEndOffset.hasUndefinedEpochOrOffset()) {
-                            handleOffsetOutOfRange(requestPosition.offset, respTopicPartition,
+                            handleOffsetOutOfRange(requestPosition, respTopicPartition,


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.

hachikuji · 2020-06-03T17:10:02Z

+                                    "Failed leader offset epoch validation for " + requestPosition
+                                        + " since no end offset larger than current fetch epoch was reported");
+                            } catch (OffsetOutOfRangeException e) {
+                                // Swallow the OffsetOutOfRangeException to finish all partitions validation.


I don't feel great about having this in the code, even if it's supposed to be temporary. I think we should just fix the exception propagation bug in this patch. It seems like it would be straightforward to do something similar to what is done in the onFailure path.

if (!(e instanceof RetriableException) && !cachedOffsetForLeaderException.compareAndSet(null, e)) { log.error("Discarding error in OffsetsForLeaderEpoch because another error is pending", e); }

Above may not be ideal, but at least it provides a way to propagate individual errors.

hachikuji · 2020-06-03T23:24:38Z

ok to test

hachikuji

LGTM. Thanks for the patch!

abbccdda · 2020-06-04T19:41:21Z


    private void initMetadata(MockClient mockClient, Map<String, Integer> partitionCounts) {
-        MetadataResponse initialMetadata = TestUtils.metadataUpdateWith(1, partitionCounts);
+        int leaderEpoch = 1;


This is the actual fix, the other parts are mostly cleanup

hachikuji · 2020-06-04T20:35:14Z

ok to test

hachikuji · 2020-06-05T01:56:34Z

ok to test

abbccdda · 2020-06-05T05:22:35Z

retest please

hachikuji · 2020-06-05T06:30:43Z

retest this please

hachikuji · 2020-06-05T06:30:55Z

retest this please

hachikuji · 2020-06-05T16:16:47Z

retest this please

…eliable (#8486) This PR provides two fixes: 1. Skip offset validation if the current leader epoch cannot be reliably determined. 2. Raise an out of range error if the leader returns an undefined offset in response to the OffsetsForLeaderEpoch request. Reviewers: Guozhang Wang <wangguoz@gmail.com>, Jason Gustafson <jason@confluent.io>

* 'trunk' of github.com:apache/kafka: (46 commits) MINOR: improve code encapsulation between StreamThread and TaskManager (apache#8819) Fixing KAFKA-10094 (apache#8797) KAFKA-9851: Revoking Connect tasks due to connectivity issues should also clear the running assignment (apache#8804) KAFKA-9840; Skip End Offset validation when the leader epoch is not reliable (apache#8486) HOT_FIX: Update javadoc since imports added (apache#8817) KAFKA-8011: Fix flaky RegexSourceIntegrationTest (apache#8799) KAFKA-9570: Define SSL configs in all worker config classes, not just distributed (apache#8135) KAFKA-10111: Make SinkTaskContext.errantRecordReporter() a default method (apache#8814) KAFKA-10110: Corrected potential NPE when null label value added to KafkaMetricsContext (apache#8811) MINOR: Change the order that Connect calls `config()` and `validate()` to avoid validating if the required ConfigDef is null (apache#8810) MINOR: fix backwards incompatibility in JmxReporter introduced by KIP-606 MINOR: Fix javadoc warnings (apache#8809) KAFKA-9441: Improve Kafka Streams task management (apache#8776) fix the broken links of streams javadoc (apache#8789) KAFKA-10040; Make computing the PreferredReplicaImbalanceCount metric more efficient (apache#8724) KAFKA-10066: TestOutputTopic should pass record headers into deserializers (apache#8759) MINOR: Add explanation for disabling forwarding from value transformers (apache#8771) KAFKA-10033: Throw UnknownTopicOrPartitionException if altering configs of non-existing topic KAFKA-9434: automated protocol for alterReplicaLogDirs (apache#8311) KAFKA-9313: Set `use_all_dns_ips` as the new default for `client.dns.lookup` (KIP-602) (apache#8644) ...

Conflicts: * build.gradle: take upstream changes regarding heap memory configuration for the build. * apache-github/trunk: (33 commits) MINOR: fix HTML markup (apache#8823) KAFKA-10012; Reduce overhead of strings in SelectorMetrics (apache#8684) KAFKA-9216: Enforce internal config topic settings for Connect workers during startup (apache#8270) KAFKA-10097: Internalize checkpoint data (apache#8820) KAFKA-10033: Throw UnknownTopicOrPartitionException when modifying a non-existent topic's config MINOR: improve code encapsulation between StreamThread and TaskManager (apache#8819) Fixing KAFKA-10094 (apache#8797) KAFKA-9851: Revoking Connect tasks due to connectivity issues should also clear the running assignment (apache#8804) KAFKA-9840; Skip End Offset validation when the leader epoch is not reliable (apache#8486) HOT_FIX: Update javadoc since imports added (apache#8817) KAFKA-8011: Fix flaky RegexSourceIntegrationTest (apache#8799) KAFKA-9570: Define SSL configs in all worker config classes, not just distributed (apache#8135) KAFKA-10111: Make SinkTaskContext.errantRecordReporter() a default method (apache#8814) KAFKA-10110: Corrected potential NPE when null label value added to KafkaMetricsContext (apache#8811) MINOR: Change the order that Connect calls `config()` and `validate()` to avoid validating if the required ConfigDef is null (apache#8810) MINOR: fix backwards incompatibility in JmxReporter introduced by KIP-606 MINOR: Fix javadoc warnings (apache#8809) KAFKA-9441: Improve Kafka Streams task management (apache#8776) fix the broken links of streams javadoc (apache#8789) KAFKA-10040; Make computing the PreferredReplicaImbalanceCount metric more efficient (apache#8724) ...

abbccdda marked this pull request as draft April 14, 2020 17:04

abbccdda force-pushed the KAFKA-9840 branch 3 times, most recently from 0910d36 to 22d7e20 Compare April 21, 2020 15:42

abbccdda force-pushed the KAFKA-9840 branch from 22d7e20 to f4c87f5 Compare April 21, 2020 19:42

abbccdda marked this pull request as ready for review April 22, 2020 02:55

abbccdda commented Apr 22, 2020

View reviewed changes

Comment thread ...s/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetsForLeaderEpochClient.java Outdated

abbccdda commented Apr 22, 2020

View reviewed changes

Comment thread clients/src/test/java/org/apache/kafka/clients/consumer/internals/SubscriptionStateTest.java Outdated

abbccdda force-pushed the KAFKA-9840 branch from c77993e to 84e5e90 Compare April 22, 2020 03:19

guozhangwang reviewed Apr 23, 2020

View reviewed changes

abbccdda force-pushed the KAFKA-9840 branch 2 times, most recently from 18d2e3a to cf73388 Compare April 23, 2020 16:46

guozhangwang reviewed Apr 23, 2020

View reviewed changes

abbccdda force-pushed the KAFKA-9840 branch from cf73388 to 3254a05 Compare May 27, 2020 21:50

hachikuji reviewed May 29, 2020

View reviewed changes

abbccdda force-pushed the KAFKA-9840 branch 3 times, most recently from 8b5d33a to 1d08929 Compare June 1, 2020 21:58

avoid offset validation when the leader epoch is not reliable, and ch…

fddb49e

…eck epoch/offset in response

abbccdda force-pushed the KAFKA-9840 branch from 1d08929 to fddb49e Compare June 2, 2020 04:58

hachikuji reviewed Jun 2, 2020

View reviewed changes

address comments

dbaa133

hachikuji reviewed Jun 3, 2020

View reviewed changes

fix further comments

03407c4

abbccdda force-pushed the KAFKA-9840 branch from 0be71f7 to 03407c4 Compare June 3, 2020 03:32

hachikuji reviewed Jun 3, 2020

View reviewed changes

address logging comments

a6c0785

hachikuji reviewed Jun 3, 2020

View reviewed changes

Comment thread clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java Outdated

abbccdda force-pushed the KAFKA-9840 branch from ed7321b to 5d5b6d4 Compare June 3, 2020 20:32

propagate exception in onSuccess callback

6a66b1b

abbccdda force-pushed the KAFKA-9840 branch from 5d5b6d4 to 6a66b1b Compare June 3, 2020 20:34

hachikuji approved these changes Jun 4, 2020

View reviewed changes

fix KafkaConsumerTest

1a97ccc

abbccdda commented Jun 4, 2020

View reviewed changes

remove unnecessary complete validation logic

87ca234

hachikuji merged commit 910f317 into apache:trunk Jun 5, 2020

Uh oh!

Conversation

abbccdda commented Apr 14, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

guozhangwang left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

guozhangwang left a comment

Choose a reason for hiding this comment

Uh oh!

ijuma commented May 17, 2020

Uh oh!

abbccdda commented May 21, 2020

Uh oh!

hachikuji commented May 27, 2020

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

hachikuji commented Jun 3, 2020

Uh oh!

hachikuji commented Jun 3, 2020

Uh oh!

hachikuji commented Jun 3, 2020

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

hachikuji commented Jun 3, 2020

Uh oh!

hachikuji commented Jun 3, 2020

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

abbccdda commented Apr 14, 2020 •

edited

Loading