KAFKA-7831; Do not modify subscription state from background thread #6221

hachikuji · 2019-02-02T00:15:46Z

Metadata may be updated from the background thread, so we need to protect access to SubscriptionState. This patch restructures the metadata handling so that we only check pattern subscriptions in the foreground. Additionally, it improves the following:

SubscriptionState is now the source of truth for the topics that will be fetched. We had a lot of messy logic previously to try and keep the the topic set in Metadata consistent with the subscription, so this simplifies the logic.
The metadata needs for the producer and consumer are quite different, so it made sense to separate the custom logic into separate extensions of Metadata. For example, only the producer requires topic expiration.
We've always had an edge case in which a metadata change with an inflight request may cause us to effectively miss an expected update. This patch implements a separate version inside Metadata which is bumped when the needed topics changes.
This patch removes the MetadataListener, which was the cause of https://issues.apache.org/jira/browse/KAFKA-7764.

Committer Checklist (excluded from commit message)

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

rajinisivaram · 2019-02-07T17:01:24Z

clients/src/main/java/org/apache/kafka/clients/Metadata.java

    /**
     * Updates the cluster metadata. If topic expiry is enabled, expiry time
     * is set for topics if required and expired topics are removed from the metadata.
     *
-     * @param metadataResponse metadata response received from the broker
+     * @param response metadata response received from the broker


nit: javadoc for new param requestVersion

rajinisivaram

@hachikuji Thanks for the PR. Made one pass through the implementation changes and it is looking really good. I will go through the tests and do another pass through the implementation tomorrow.

clients/src/main/java/org/apache/kafka/clients/Metadata.java

clients/src/main/java/org/apache/kafka/clients/NetworkClient.java

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

rajinisivaram

Did another pass through the implementation and left some minor comments. Still need to go through the unit tests.

clients/src/main/java/org/apache/kafka/clients/Metadata.java

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

rajinisivaram

@hachikuji I have gone through the tests as well. The PR is looking good, I have left only some minor comments.

clients/src/test/java/org/apache/kafka/clients/MetadataTest.java

clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java

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

rajinisivaram · 2019-02-12T10:05:33Z

clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java


-        for (List<ConsumerRecord<byte[], byte[]>> records: fetcherNoAutoReset.fetchedRecords().values())
+        Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> recordsByPartition = fetchedRecords();
+        for (List<ConsumerRecord<byte[], byte[]>> records : recordsByPartition.values())


Could do recordsByPartition.values().forEach(fetchedRecords::addAll)?

rajinisivaram · 2019-02-12T10:16:42Z

clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java

    private <T> List<Long> collectRecordOffsets(List<ConsumerRecord<T, T>> records) {
        List<Long> res = new ArrayList<>(records.size());
        for (ConsumerRecord<?, ?> record : records)
            res.add(record.offset());
        return res;
    }
+


nit: unnecessary newline?

rajinisivaram · 2019-02-12T10:31:10Z

clients/src/test/java/org/apache/kafka/common/utils/SystemTimeTest.java

+    protected Time createTime() {
+        return Time.SYSTEM;
+    }
+}


nit: newline?

rajinisivaram · 2019-02-12T10:32:41Z

clients/src/test/java/org/apache/kafka/common/utils/TimeTest.java

+
+        assertNull(caughtException.get());
+    }
+}


nit: newline?

rajinisivaram · 2019-02-12T10:33:33Z

clients/src/test/java/org/apache/kafka/common/utils/TimeTest.java

+    public void testWaitObjectConditionSatisfied() throws InterruptedException {
+        Object obj = new Object();
+        Time time = createTime();
+        long timeoutMs = 100;


Perhaps use a large timeout here and test that join returns before timeout elapses?

rajinisivaram · 2019-02-12T10:41:15Z

clients/src/main/java/org/apache/kafka/common/utils/Time.java

+     *
+     * @throws org.apache.kafka.common.errors.TimeoutException if the timeout expires before the condition is satisfied
+     */
+    void waitObject(Object obj, Predicate<Void> condition, long timeoutMs) throws InterruptedException;


Could we add description of the params to the javadoc? In particular the fact that we synchronize on obj and that we expect the caller to notify when condition is satisfied?

Instead of Predicate<Void> what about Supplier<Boolean>?

ijuma · 2019-02-17T00:05:54Z

In KAFKA-7565, Rajini added a comment about this PR. Including it here so that @hachikuji can take it into consideration:

Jason Gustafson I am not sure if this is covered by your PR for KAFKA-7831.

We have a check for whether the partitions returned in a fetch response are still in the fetch session and we return if that is not the case:

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

Line 234 in b02b5b6

if (!handler.handleResponse(response)) {

Further down in the code, we assume that the partitions do exist in the session and access the partitions without checking for null:

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

Line 243 in b02b5b6

long fetchOffset = data.sessionPartitions().get(partition).fetchOffset;

Presumably fetch session partitions can change in between lines 234 aand 243 since the response could be processed on the heartbeat thread. With the changes from your PR for KAFKA-7831, can we guarantee that the session partitions wont change

hachikuji · 2019-02-20T17:12:03Z

@rajinisivaram Thanks for the review. I will address your comments today.

hachikuji · 2019-02-22T08:09:46Z

@rajinisivaram Thanks for reviewing. I pushed an update. I had a couple questions in response to your comments as well.

rajinisivaram · 2019-03-04T18:43:14Z

@hachikuji Thanks for the updates, looks good. Left a comment on inProgressRequestVersion. Also, the PR needs rebasing since there is a conflict.

ijuma · 2019-03-06T13:43:14Z

Tests are passing

mumrah

LGTM. Few questions/comments inline

mumrah · 2019-03-06T01:42:09Z

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

@@ -94,7 +102,7 @@ public SubscriptionState(LogContext logContext, OffsetResetStrategy defaultReset
        this.defaultResetStrategy = defaultResetStrategy;
        this.subscription = Collections.emptySet();
        this.assignment = new PartitionStates<>();
-        this.groupSubscription = new HashSet<>();
+        this.groupSubscription = Collections.newSetFromMap(new ConcurrentHashMap<>());


While looking this up (hadn't seen newSetFromMap before), I found that ConcurrentHashMap has a newKeySet method which is basically equivalent to this. Not suggesting we use it, just thought I'd share

mumrah · 2019-03-06T01:48:13Z

clients/src/main/java/org/apache/kafka/common/utils/Time.java

+     *
+     * @throws org.apache.kafka.common.errors.TimeoutException if the timeout expires before the condition is satisfied
+     */
+    void waitObject(Object obj, Predicate<Void> condition, long timeoutMs) throws InterruptedException;


Instead of Predicate<Void> what about Supplier<Boolean>?

mumrah · 2019-03-06T01:52:25Z

clients/src/main/java/org/apache/kafka/clients/Metadata.java

    private long lastRefreshMs;
    private long lastSuccessfulRefreshMs;
    private AuthenticationException authenticationException;
+    private KafkaException metadataException;


Do we care about overwriting this value if get-and-clear hasn't been called?

mumrah · 2019-03-06T20:04:09Z

clients/src/main/java/org/apache/kafka/clients/Metadata.java

        this.lastRefreshMs = 0L;
        this.lastSuccessfulRefreshMs = 0L;
-        this.version = 0;
+        this.requestVersion = 0;
+        this.updateVersion = 0;


I wonder if having the broker manage a metadata version would make this any easier?

I think it would. It would be nice to have a simple version in the metadata response that we could use to stale metadata. I suspect the partition leader epoch is the closest we can get to that without introducing server metadata bottlenecks. That said, I think some amount of bookkeeping is unavoidable in the client since the set of topics that it is interested in changes over time. You need some way of telling whether pending metadata fetches reflect changes to the interested topic set.

rajinisivaram

@hachikuji Thanks for the PR, LGTM

hachikuji · 2019-03-08T00:27:05Z

@rajinisivaram @mumrah Thanks much for reviews. Merging to trunk only. Apologies in advance for anyone who has to fix conflicts in their PR.

…pache#6221) Metadata may be updated from the background thread, so we need to protect access to SubscriptionState. This patch restructures the metadata handling so that we only check pattern subscriptions in the foreground. Additionally, it improves the following: 1. SubscriptionState is now the source of truth for the topics that will be fetched. We had a lot of messy logic previously to try and keep the the topic set in Metadata consistent with the subscription, so this simplifies the logic. 2. The metadata needs for the producer and consumer are quite different, so it made sense to separate the custom logic into separate extensions of Metadata. For example, only the producer requires topic expiration. 3. We've always had an edge case in which a metadata change with an inflight request may cause us to effectively miss an expected update. This patch implements a separate version inside Metadata which is bumped when the needed topics changes. 4. This patch removes the MetadataListener, which was the cause of https://issues.apache.org/jira/browse/KAFKA-7764. Reviewers: David Arthur <mumrah@gmail.com>, Rajini Sivaram <rajinisivaram@googlemail.com>

hachikuji force-pushed the KAFKA-7831 branch 6 times, most recently from d87b815 to 03cbbc1 Compare February 4, 2019 17:21

rajinisivaram reviewed Feb 7, 2019

View reviewed changes

rajinisivaram reviewed Feb 11, 2019

View reviewed changes

rajinisivaram reviewed Feb 12, 2019

View reviewed changes

hachikuji force-pushed the KAFKA-7831 branch from 03cbbc1 to 11e2842 Compare February 22, 2019 08:08

hachikuji force-pushed the KAFKA-7831 branch 2 times, most recently from 8a7a45d to cefe5df Compare February 23, 2019 18:39

hachikuji force-pushed the KAFKA-7831 branch from cefe5df to b8757e2 Compare March 4, 2019 23:41

mumrah reviewed Mar 6, 2019

View reviewed changes

hachikuji added 2 commits March 7, 2019 11:43

KAFKA-7831; Do not modify subscription state from background thread

ba727b2

Address review comments

ac7e43e

hachikuji force-pushed the KAFKA-7831 branch 2 times, most recently from d3d0945 to 84366c0 Compare March 7, 2019 20:57

rajinisivaram approved these changes Mar 7, 2019

View reviewed changes

Use Supplier<Boolean> instead of Predictate<Void> for wait condition

546247c

hachikuji force-pushed the KAFKA-7831 branch from 84366c0 to 546247c Compare March 7, 2019 21:10

hachikuji merged commit 460e46c into apache:trunk Mar 8, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-7831; Do not modify subscription state from background thread #6221

KAFKA-7831; Do not modify subscription state from background thread #6221

hachikuji commented Feb 2, 2019

rajinisivaram Feb 7, 2019

rajinisivaram left a comment

rajinisivaram left a comment

rajinisivaram left a comment

rajinisivaram Feb 12, 2019

rajinisivaram Feb 12, 2019

rajinisivaram Feb 12, 2019

rajinisivaram Feb 12, 2019

rajinisivaram Feb 12, 2019

rajinisivaram Feb 12, 2019

mumrah Mar 6, 2019

ijuma commented Feb 17, 2019

hachikuji commented Feb 20, 2019

hachikuji commented Feb 22, 2019

rajinisivaram commented Mar 4, 2019

ijuma commented Mar 6, 2019

mumrah left a comment

mumrah Mar 6, 2019

mumrah Mar 6, 2019

mumrah Mar 6, 2019

mumrah Mar 6, 2019

hachikuji Mar 7, 2019

rajinisivaram left a comment

hachikuji commented Mar 8, 2019

KAFKA-7831; Do not modify subscription state from background thread #6221

KAFKA-7831; Do not modify subscription state from background thread #6221

Conversation

hachikuji commented Feb 2, 2019

Committer Checklist (excluded from commit message)

Choose a reason for hiding this comment

rajinisivaram left a comment

Choose a reason for hiding this comment

rajinisivaram left a comment

Choose a reason for hiding this comment

rajinisivaram 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ijuma commented Feb 17, 2019

hachikuji commented Feb 20, 2019

hachikuji commented Feb 22, 2019

rajinisivaram commented Mar 4, 2019

ijuma commented Mar 6, 2019

mumrah 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

Choose a reason for hiding this comment

rajinisivaram left a comment

Choose a reason for hiding this comment

hachikuji commented Mar 8, 2019