Consumer can block indefinitely when resetting offsets for a deleted topic

[royantman]

Hi, bumped into the following edge case which results in the fetcher getting stuck indefinitely
inside fetcher._retrieve_offsets():

• Consumer is running in a stable group, subscribed to multiple topics
• A topic that is part of the subscription is deleted (no commits for these TPs)
• A fetch request will get error_code=3 (Unknown) and a MetadataRequest will be made
• Subscription state is being updated due after the MetadataResponse
• Rebalance is triggered due to subscription change
• Revoke, group leader election, sync assignment, and assignment are taking place
• During that time, another topic that was part of the subscription is deleted (no commits for these TPs either)
• KafkaConsumer._coordinator.poll() did not trigger a rebalance
• KafkaConsumer._update_fetch_positions(self._subscription.missing_fetch_positions()) is causing the client to get the reset positions for the assignment (which still holds a topic that no longer exists due to deletion)
• Errors.UnknownTopicOrPartitionError is thrown from fetcher._handle_offset_response() since the partition does no exist
• A metadata refresh is taking place and the subscription is updated, but we are caught in an infinite loop inside fetcher._retrieve_offsets() since the timestamps are not updated and timeout is infinity and we now get an early exit in fetcher._send_offset_requests() that forces metadata update over and over again

        timestamps_by_node = collections.defaultdict(dict)
        for partition, timestamp in six.iteritems(timestamps):
            node_id = self._client.cluster.leader_for_partition(partition)
            if node_id is None:
                self._client.add_topic(partition.topic)
                log.debug("Partition %s is unknown for fetching offset,"
                          " wait for metadata refresh", partition)
                return Future().failure(Errors.StaleMetadata(partition))
...
...

If you guide me to what you think would be a good way to tackle this, i'll work on a PR.

Cheers.

[royantman]

The first thing I wanted to do is to remove the unknown partition from the timestamps list since the metadata and subscription are updated.

[jeffwidman]

What versions of Kafka broker and kafka-python are you using?

[royantman]

1.4.4, 1.4.5, 1.4.6, master

Kafka 0.10.2.1 / 2.0.0

[royantman]

Patched 1.4.6 version (PR #1782) is running on our busy lower environments where topic creations and deletions are happening rapidly for a month now, no issues.
FYI

[dpkp]

Is this the same issue as https://issues.apache.org/jira/browse/KAFKA-6397 ? If so, I'd prefer to just implement the upstream solution.

[royantman]

@dpkp @jeffwidman 1.4.6 code seems to be aligned with Kafka 1.0
KAFKA-6397 (Kafka 1.1) might solve all this but will demand pretty big refactoring.
Besides many method names refactoring, all the reset partitions path was changed.
Fetcher._retrieve_offsets()(python) or Fetcher.retrieveOffsetsByTimes()(java) is being completely avoided and is now being called only by Consumer public API such as offsetsForTimes, beginningOffsets and endOffsets. This also means that Fetcher._retrieve_offsets() cannot be called with timeout=float("inf") (apart from tests)

I need time to research if the new reset offsets approach of KAFKA-6397 indeed solves the issue.
Also seems like Kafka 2.x overrides some of the changes made in Kafka 1.1, so if the goal is to perform a similar implementation, should we look into getting closer to the latest version?

let me know what you think.

[jeffwidman]

I think everyone would be two 👍 on updating to match newer Java version... the only reason we haven't is lack of man hours on a volunteer project.

[royantman]

Will see when I can find time for this. Probably only in around 1 month