KAFKA-7044: Fix Fetcher.fetchOffsetsByTimes and NPE in describe consumer group #5627

apovzner · 2018-09-09T08:02:04Z

kafka-consumer-groups --describe --group ... can result in NullPointerException for two reasons:

Fetcher.fetchOffsetsByTimes() may return too early, without sending list offsets request for topic partitions that are not in cached metadata.
ConsumerGroupCommand.getLogEndOffsets() and getLogStartOffsets() assumed that endOffsets()/beginningOffsets() which eventually call Fetcher.fetchOffsetsByTimes(), would return a map with all the topic partitions passed to endOffsets()/beginningOffsets() and that values are not null. Because of (1), null values were possible if some of the topic partitions were already known (in metadata cache) and some not (metadata cache did not have entries for some of the topic partitions). However, even with fixing (1), endOffsets()/beginningOffsets() may return a map with some topic partitions missing, when list offset request returns a non-retriable error. This happens in corner cases such as message format on broker is before 0.10, or maybe in cases of some other errors.

Testing:
-- added unit test to verify fix in Fetcher.fetchOffsetsByTimes()
-- did some manual testing with kafka-consumer-groups --describe, causing NPE. Was not able to reproduce any NPE cases with DescribeConsumerGroupTest.scala,

Committer Checklist (excluded from commit message)

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

hachikuji · 2018-09-09T18:36:26Z

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


                remainingToSearch.keySet().removeAll(result.fetchedOffsets.keySet());
+                remainingToSearch.keySet().removeAll(value.partitionsWithUnknownOffset);
+                if (value.partitionsToRetry.isEmpty() && remainingToSearch.isEmpty())


There seems to be some redundance between partitionsToRetry and remainingToSearch. It might be nicer if we could get rid of remainingToSearch so that we only had to rely on ListOffsetResult to know if we should retry. I think the only thing we need is to avoid losing partitions in the call to groupListOffsetRequests.

Ah yes, we could add partitions we lose in the groupListOffsetRequest to partitionsToRetry, since sendListOffsetsRequests is called only from that one method... Let me try that.

…describe consumer group

hachikuji

Thanks for the updates. Left a couple more comments.

hachikuji · 2018-09-10T15:49:37Z

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

@@ -414,7 +415,9 @@ private ListOffsetResult fetchOffsetsByTimes(Map<TopicPartition, Long> timestamp
                if (value.partitionsToRetry.isEmpty())
                    return result;

-                remainingToSearch.keySet().removeAll(result.fetchedOffsets.keySet());
+                remainingToSearch = timestampsToSearch.entrySet().stream()


I think this can be replaced with remainingToSearch.keySet().retainAll(value.partitionsToRetry).

hachikuji · 2018-09-10T15:54:31Z

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

@@ -632,9 +635,11 @@ public void onFailure(RuntimeException e) {
        final RequestFuture<ListOffsetResult> listOffsetRequestsFuture = new RequestFuture<>();
        final Map<TopicPartition, OffsetData> fetchedTimestampOffsets = new HashMap<>();
        final Set<TopicPartition> partitionsToRetry = new HashSet<>();
+        final Set<TopicPartition> partitionsRequireMetadataUpdate = new HashSet<>(timestampsToSearch.keySet());


Another idea might be to pass partitionsToRetry into groupListOffsetRequests.

apovzner · 2018-09-10T22:02:42Z

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

            } else if (client.isUnavailable(info.leader())) {
                client.maybeThrowAuthFailure(info.leader());

                // The connection has failed and we need to await the blackout period before we can
                // try again. No need to request a metadata update since the disconnect will have
                // done so already.
                log.debug("Leader {} for partition {} is unavailable for fetching offset until reconnect backoff expires",
-                        info.leader(), tp);
+                          info.leader(), tp);
+                partitionsToRetry.add(tp);


I moved the logic for getting the list of partitions that do not have available leader to groupListOffsetRequests . I was contemplating whether we should add partition for which we have a failed connection to leader to partitionsToRetry. Seems like we should, because we may be able to reconnect before the list offsets timeout.

hachikuji · 2018-09-11T06:07:32Z

core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

-          logEndOffsetResult._2 match {
-            case LogOffsetResult.LogOffset(logEndOffset) => getDescribePartitionResult(logEndOffsetResult._1, Some(logEndOffset))
-            case LogOffsetResult.Unknown => getDescribePartitionResult(logEndOffsetResult._1, None)
-            case LogOffsetResult.Ignore => null


As far as I can tell, this was the only use of LogOffsetResult.Ignore. Maybe we can get rid of it?

If we did that, then we could probably also get rid of LogOffsetResult and replace it with a simple Option[Long], but we can leave that for a separate PR.

hachikuji

LGTM. We can do the cleanup of LogOffsetResult separately.

…mer group (#5627) A call to `kafka-consumer-groups --describe --group ...` can result in NullPointerException for two reasons: 1) `Fetcher.fetchOffsetsByTimes()` may return too early, without sending list offsets request for topic partitions that are not in cached metadata. 2) `ConsumerGroupCommand.getLogEndOffsets()` and `getLogStartOffsets()` assumed that endOffsets()/beginningOffsets() which eventually call Fetcher.fetchOffsetsByTimes(), would return a map with all the topic partitions passed to endOffsets()/beginningOffsets() and that values are not null. Because of (1), null values were possible if some of the topic partitions were already known (in metadata cache) and some not (metadata cache did not have entries for some of the topic partitions). However, even with fixing (1), endOffsets()/beginningOffsets() may return a map with some topic partitions missing, when list offset request returns a non-retriable error. This happens in corner cases such as message format on broker is before 0.10, or maybe in cases of some other errors. Testing: -- added unit test to verify fix in Fetcher.fetchOffsetsByTimes() -- did some manual testing with `kafka-consumer-groups --describe`, causing NPE. Was not able to reproduce any NPE cases with DescribeConsumerGroupTest.scala, Reviewers: Jason Gustafson <jason@confluent.io>

…mer group (apache#5627) A call to `kafka-consumer-groups --describe --group ...` can result in NullPointerException for two reasons: 1) `Fetcher.fetchOffsetsByTimes()` may return too early, without sending list offsets request for topic partitions that are not in cached metadata. 2) `ConsumerGroupCommand.getLogEndOffsets()` and `getLogStartOffsets()` assumed that endOffsets()/beginningOffsets() which eventually call Fetcher.fetchOffsetsByTimes(), would return a map with all the topic partitions passed to endOffsets()/beginningOffsets() and that values are not null. Because of (1), null values were possible if some of the topic partitions were already known (in metadata cache) and some not (metadata cache did not have entries for some of the topic partitions). However, even with fixing (1), endOffsets()/beginningOffsets() may return a map with some topic partitions missing, when list offset request returns a non-retriable error. This happens in corner cases such as message format on broker is before 0.10, or maybe in cases of some other errors. Testing: -- added unit test to verify fix in Fetcher.fetchOffsetsByTimes() -- did some manual testing with `kafka-consumer-groups --describe`, causing NPE. Was not able to reproduce any NPE cases with DescribeConsumerGroupTest.scala, Reviewers: Jason Gustafson <jason@confluent.io>

hachikuji self-assigned this Sep 9, 2018

hachikuji reviewed Sep 9, 2018

View reviewed changes

apovzner added 4 commits September 9, 2018 20:03

KAFKA-7044: Fix early return from fetchOffsetsByTimes and fix NPE in …

474bfb4

…describe consumer group

New test exceeded default max number of lines in the FetcherTest class

3306849

addressed review comments

4f1a6a0

cleanup

aec7385

apovzner force-pushed the kafka-7044 branch from 8af4aff to aec7385 Compare September 10, 2018 03:09

hachikuji reviewed Sep 10, 2018

View reviewed changes

passing partitions to retry to groupListOffsetRequests

e71e6f4

apovzner commented Sep 10, 2018

View reviewed changes

removed printing partitions to retry in resetOffsetsAsync

1f8fbd8

hachikuji reviewed Sep 11, 2018

View reviewed changes

hachikuji approved these changes Sep 11, 2018

View reviewed changes

hachikuji merged commit e2ec2d7 into apache:trunk Sep 11, 2018

tr4r3x mentioned this pull request Nov 6, 2019

Update kafka dependency to 2.0.1 zalando-incubator/remora#74

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-7044: Fix Fetcher.fetchOffsetsByTimes and NPE in describe consumer group #5627

KAFKA-7044: Fix Fetcher.fetchOffsetsByTimes and NPE in describe consumer group #5627

apovzner commented Sep 9, 2018 •

edited

Loading

hachikuji Sep 9, 2018

apovzner Sep 10, 2018

hachikuji left a comment

hachikuji Sep 10, 2018

hachikuji Sep 10, 2018

apovzner Sep 10, 2018

hachikuji Sep 11, 2018

hachikuji left a comment

KAFKA-7044: Fix Fetcher.fetchOffsetsByTimes and NPE in describe consumer group #5627

KAFKA-7044: Fix Fetcher.fetchOffsetsByTimes and NPE in describe consumer group #5627

Conversation

apovzner commented Sep 9, 2018 • edited Loading

Committer Checklist (excluded from commit message)

hachikuji Sep 9, 2018

Choose a reason for hiding this comment

apovzner Sep 10, 2018

Choose a reason for hiding this comment

hachikuji left a comment

Choose a reason for hiding this comment

hachikuji Sep 10, 2018

Choose a reason for hiding this comment

hachikuji Sep 10, 2018

Choose a reason for hiding this comment

apovzner Sep 10, 2018

Choose a reason for hiding this comment

hachikuji Sep 11, 2018

Choose a reason for hiding this comment

hachikuji left a comment

Choose a reason for hiding this comment

apovzner commented Sep 9, 2018 •

edited

Loading