KAFKA-7703: position() may return a wrong offset after seekToEnd #6407

viktorsomogyi · 2019-03-08T17:27:30Z

When poll is called which resets the offsets to the beginning, followed by a seekToEnd and a position, it could happen that the "reset to earliest" call in poll overrides the "reset to latest" initiated by seekToEnd in a very delicate way:

both request has been issued and returned to the client side (listOffsetResponse has happened)
in Fetcher.resetOffsetIfNeeded(TopicPartition, Long, OffsetData) the thread scheduler could prefer the heartbeat thread with the "reset to earliest" call, overriding the offset to the earliest and setting the SubscriptionState with that position.
The thread scheduler continues execution of the thread (application thread) with the "reset to latest" call and discards it as the "reset to earliest" already set the position - the wrong one.
The blocking position call returns with the earliest offset instead of the latest, despite it wasn't expected.

The fix makes the TopicPartitionState in SubscriptionState synchronized and starts to track the requested reset timestamp. With this we can precisely decide if the incoming offset reset is really what we want (by comparing the timestamp set when assigning for reset and the one that is actually used on seek). Therefore the latest initiated offset reset will happen only. Synchronization furthermore ensures that this is done in an atomic manner to avoid further similar bugs.

Committer Checklist (excluded from commit message)

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

ijuma · 2019-03-09T15:36:17Z

Cc @hachikuji @rajinisivaram

hachikuji

Thanks, good find. Left a couple comments.

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

clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java

viktorsomogyi · 2019-03-27T12:57:52Z

retest this please

hachikuji

@viktorsomogyi Thanks for the update. I left one suggestion to consider. Let me know what you think.

clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java

viktorsomogyi · 2019-04-12T15:00:52Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java

-                boolean match = this.subscription.contains(topicPartition.topic());
-                if (!match) {
-                    log.info("Assigned partition {} for non-subscribed topic; subscription is {}", topicPartition, this.subscription);
+            synchronized (SubscriptionState.this) {


Spotbugs throw an error for this but I don't think it's accurate since we synchronize the whole method and this predicate will be executed there. Should I add this to the excludes?

Actually I'm not sure why we need this additional block if we are already holding the lock. The predicate wouldn't be used after this method returns.

Basically this is why I added it, I think spotbugs incorrectly detects this. I hope I can make an exception in spotbugs for this method.

hachikuji

Thanks for the updates. Left a few more comments.

clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java

hachikuji · 2019-04-12T17:03:39Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java

            return !hasValidPosition() && !awaitingReset();
        }

-        private boolean isPaused() {
+        private synchronized boolean isPaused() {


Do we still need synchronization here if SubscriptionState is synchronized?

Not all method in SubscriptionState is synchronized. Basically when I don't need to modify assignment I only synchronize until I get the TopicPartitionState and then lock that object to avoid locking the whole collection when it's not necessary.
If you look at isPaused there:

public boolean isPaused(TopicPartition tp) { TopicPartitionState assignedOrNull = assignedStateOrNull(tp); return assignedOrNull != null && assignedOrNull.isPaused(); }

the first line would only get the state from the collection, release the lock of SubscriptionState and only lock the given TopicPartitionState.
When I write the collection then I lock for instance in assignFromUser then I lock the whole method as I think we should keep writes consistent.
If you think this is unnecessary or complicated I can just lock SubscriptionState every time.

Yeah, I have a slight preference to just lock SubscriptionState every time since it is the simplest option. I don't think contention is a major problem since there's only the heartbeat thread which is sleeping most of the time. Unless there's some reason to think the cost of lock acquisition itself is a concern.

Ok. I don't think lock acquisition would be expensive in this case.

hachikuji · 2019-04-12T17:10:46Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java

-                boolean match = this.subscription.contains(topicPartition.topic());
-                if (!match) {
-                    log.info("Assigned partition {} for non-subscribed topic; subscription is {}", topicPartition, this.subscription);
+            synchronized (SubscriptionState.this) {


Actually I'm not sure why we need this additional block if we are already holding the lock. The predicate wouldn't be used after this method returns.

clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java

hachikuji

Sorry for the delay. This is looking good. I had just a few additional comments.

clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java

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

clients/src/main/java/org/apache/kafka/common/internals/PartitionStates.java

gradle/spotbugs-exclude.xml

clients/src/main/java/org/apache/kafka/common/internals/PartitionStates.java

hachikuji · 2019-05-24T23:12:19Z

retest this please

hachikuji · 2019-05-24T23:16:32Z

Compilation failing:

16:14:09 > Task :clients:compileJava
16:14:09 /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.12@2/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java:108: error: partitionStateValues() is not public in PartitionStates; cannot be accessed from outside package
16:14:09             ", assignment=" + assignment.partitionStateValues() + " (id=" + assignmentId + ")}";
16:14:09                                         ^
16:14:09   where S is a type-variable:
16:14:09     S extends Object declared in class PartitionStates
16:14:11 1 error
16:14:12

… called

viktorsomogyi · 2019-05-27T15:28:32Z

Fixed the compilation error (it was due to a conflict with another commit) but the test failures seem suspicious. Please hold on for now.

hachikuji

Thanks, just a couple more comments.

hachikuji · 2019-05-28T17:52:19Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java

-    public void resetGroupSubscription() {
-        this.groupSubscription.retainAll(subscription);
+    synchronized void resetGroupSubscription() {
+        groupSubscription = new HashSet<>(groupSubscription);


Since we also treat subscription as immutable once created, could we change this to groupSubscription = subscription?

Makes sense, will do.

hachikuji · 2019-05-28T18:05:56Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java

-        this.subscription = topicsToSubscribe;
-        this.groupSubscription.addAll(topicsToSubscribe);
+        subscription = topicsToSubscribe;
+        groupSubscription = topicsToSubscribe;


This logic looks a little different. The "group subscription" is a little hard to understand. It's intended to be the union of the subscriptions of all consumers in the group. When we change the local subscription, we should still keep the subscription from the rest of the group. Perhaps separately we can consider how to simplify this bookkeeping.

Thanks for pointing this out. I'll fix this.

hachikuji

LGTM. Thanks @viktorsomogyi .

hachikuji · 2019-05-29T22:50:29Z

I will go ahead and merge. The failing test is a known flake: https://issues.apache.org/jira/browse/KAFKA-8122.

When poll is called which resets the offsets to the beginning, followed by a seekToEnd and a position, it could happen that the "reset to earliest" call in poll overrides the "reset to latest" initiated by seekToEnd in a very delicate way: 1. both request has been issued and returned to the client side (listOffsetResponse has happened) 2. in Fetcher.resetOffsetIfNeeded(TopicPartition, Long, OffsetData) the thread scheduler could prefer the heartbeat thread with the "reset to earliest" call, overriding the offset to the earliest and setting the SubscriptionState with that position. 3. The thread scheduler continues execution of the thread (application thread) with the "reset to latest" call and discards it as the "reset to earliest" already set the position - the wrong one. 4. The blocking position call returns with the earliest offset instead of the latest, despite it wasn't expected. The fix makes SubscriptionState synchronized so that we can verify that the reset is expected while holding the lock. Reviewers: Jason Gustafson <jason@confluent.io>

viktorsomogyi · 2019-05-30T10:59:10Z

@hachikuji thanks a lot for reviewing this!

…che#6407) When poll is called which resets the offsets to the beginning, followed by a seekToEnd and a position, it could happen that the "reset to earliest" call in poll overrides the "reset to latest" initiated by seekToEnd in a very delicate way: 1. both request has been issued and returned to the client side (listOffsetResponse has happened) 2. in Fetcher.resetOffsetIfNeeded(TopicPartition, Long, OffsetData) the thread scheduler could prefer the heartbeat thread with the "reset to earliest" call, overriding the offset to the earliest and setting the SubscriptionState with that position. 3. The thread scheduler continues execution of the thread (application thread) with the "reset to latest" call and discards it as the "reset to earliest" already set the position - the wrong one. 4. The blocking position call returns with the earliest offset instead of the latest, despite it wasn't expected. The fix makes SubscriptionState synchronized so that we can verify that the reset is expected while holding the lock. Reviewers: Jason Gustafson <jason@confluent.io>

viktorsomogyi force-pushed the kafka-consumer-seek-bug branch 3 times, most recently from ed52b04 to 10acb07 Compare March 9, 2019 15:19

hachikuji self-assigned this Mar 15, 2019

hachikuji reviewed Mar 19, 2019

View reviewed changes

clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java Outdated Show resolved Hide resolved

clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java Outdated Show resolved Hide resolved

viktorsomogyi force-pushed the kafka-consumer-seek-bug branch from fecccf5 to 45ed81e Compare March 26, 2019 13:01

hachikuji reviewed Apr 4, 2019

View reviewed changes

clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java Outdated Show resolved Hide resolved

viktorsomogyi force-pushed the kafka-consumer-seek-bug branch 2 times, most recently from 808ab74 to 31eecaa Compare April 12, 2019 14:58

viktorsomogyi commented Apr 12, 2019

View reviewed changes

clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java Show resolved Hide resolved

viktorsomogyi commented Apr 12, 2019

View reviewed changes

hachikuji reviewed Apr 12, 2019

View reviewed changes

viktorsomogyi force-pushed the kafka-consumer-seek-bug branch 3 times, most recently from 3aeaed9 to 9705b17 Compare April 25, 2019 14:04

viktorsomogyi force-pushed the kafka-consumer-seek-bug branch from 9705b17 to e1568af Compare May 21, 2019 08:46

hachikuji reviewed May 22, 2019

View reviewed changes

viktorsomogyi added 4 commits May 27, 2019 10:27

KAFKA-7703: position() may return a wrong offset after "seekToEnd" is…

9acfc8c

… called

Synchronize SubscriptionState instead of TopicPartitionState

d2d4cca

Synchronize preferred replica methods

f01800d

Fix review comments

818389b

viktorsomogyi force-pushed the kafka-consumer-seek-bug branch from 9e6e24a to 818389b Compare May 27, 2019 09:09

Fixes after rebase

0f089d0

Fix test issues

ae33505

hachikuji reviewed May 28, 2019

View reviewed changes

Address subscription related comments

c6ef427

hachikuji approved these changes May 29, 2019

View reviewed changes

hachikuji merged commit e82e2e7 into apache:trunk May 29, 2019

viktorsomogyi deleted the kafka-consumer-seek-bug branch May 30, 2019 11:32

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-7703: position() may return a wrong offset after seekToEnd #6407

KAFKA-7703: position() may return a wrong offset after seekToEnd #6407

viktorsomogyi commented Mar 8, 2019 •

edited

Loading

ijuma commented Mar 9, 2019

hachikuji left a comment

viktorsomogyi commented Mar 27, 2019

hachikuji left a comment

viktorsomogyi Apr 12, 2019

hachikuji Apr 12, 2019

viktorsomogyi Apr 18, 2019

hachikuji left a comment

hachikuji Apr 12, 2019

viktorsomogyi Apr 18, 2019

hachikuji Apr 24, 2019

viktorsomogyi Apr 25, 2019 •

edited

Loading

hachikuji Apr 12, 2019

hachikuji left a comment

hachikuji commented May 24, 2019

hachikuji commented May 24, 2019

viktorsomogyi commented May 27, 2019

hachikuji left a comment

hachikuji May 28, 2019

viktorsomogyi May 29, 2019

hachikuji May 28, 2019

viktorsomogyi May 29, 2019

hachikuji left a comment

hachikuji commented May 29, 2019

viktorsomogyi commented May 30, 2019 •

edited

Loading

KAFKA-7703: position() may return a wrong offset after seekToEnd #6407

KAFKA-7703: position() may return a wrong offset after seekToEnd #6407

Conversation

viktorsomogyi commented Mar 8, 2019 • edited Loading

Committer Checklist (excluded from commit message)

ijuma commented Mar 9, 2019

hachikuji left a comment

Choose a reason for hiding this comment

viktorsomogyi commented Mar 27, 2019

hachikuji left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hachikuji left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

viktorsomogyi Apr 25, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hachikuji left a comment

Choose a reason for hiding this comment

hachikuji commented May 24, 2019

hachikuji commented May 24, 2019

viktorsomogyi commented May 27, 2019

hachikuji left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hachikuji left a comment

Choose a reason for hiding this comment

hachikuji commented May 29, 2019

viktorsomogyi commented May 30, 2019 • edited Loading

viktorsomogyi commented Mar 8, 2019 •

edited

Loading

viktorsomogyi Apr 25, 2019 •

edited

Loading

viktorsomogyi commented May 30, 2019 •

edited

Loading