James Palmer created KAFKA-21309:
------------------------------------
Summary: Consumer using SubscriptionPattern throws
IllegalStateException on Metadata response for a deleted assigned topic"
Key: KAFKA-21309
URL: https://issues.apache.org/jira/browse/KAFKA-21309
Project: Kafka
Issue Type: Bug
Affects Versions: 4.3.1, 4.2.0, 4.1.0
Reporter: James Palmer
Assignee: James Palmer
A Java consumer on kafka-clients 4.1 or later, using
\{{group.protocol=consumer}} and \{{subscribe(SubscriptionPattern)}}, throws
\{{IllegalStateException("Use errorsByTopicId() when managing topic using topic
id")}} when a topic assigned to it is deleted.
* Since 4.1 (KAFKA-18117), this consumer asks for metadata by topic ID.
* The broker answers a deleted topic's ID with \{{UNKNOWN_TOPIC_ID}} and no
name, which the schema allows from v12.
* \{{NetworkClient}} then reads the errors by topic name, and
\{{MetadataResponse.errors()}} throws on the row without a name.
As a consequence, the client never marks the Metadata request as finished, so
it sends no more Metadata requests and never consumes new matching topics. This
happens with default settings. With auto-commit disabled, we also saw it send
\{{ConsumerGroupHeartbeat}} continuously once a new matching topic was assigned.
kafka-clients 4.0.0 is not affected, nor are consumers subscribed by topic
names. The code is unchanged on trunk.
Reproduce:
https://gist.github.com/jpalmerr/b653005afeb74d50346f8bf085d11ac5/ec78cbbbba3b10035858bd9d5f57e8e11effb36b
The consumer logs this once, shortly after the delete:
{noformat}
ERROR Unexpected error caught in consumer network thread
java.lang.IllegalStateException: Use errorsByTopicId() when managing topic
using topic id
at
org.apache.kafka.common.requests.MetadataResponse.errors(MetadataResponse.java:99)
at
org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.handleSuccessfulResponse(NetworkClient.java:1293)
at
org.apache.kafka.clients.NetworkClient.handleCompletedReceives(NetworkClient.java:1011)
at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:654)
at
org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.poll(NetworkClientDelegate.java:166)
at
org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.poll(NetworkClientDelegate.java:148)
at
org.apache.kafka.clients.consumer.internals.ConsumerNetworkThread.runOnce(ConsumerNetworkThread.java:228)
at
org.apache.kafka.clients.consumer.internals.ConsumerNetworkThread.run(ConsumerNetworkThread.java:162)
{noformat}
A failing unit test in \{{NetworkClientTest}} will follow as a PR.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
