[
https://issues.apache.org/jira/browse/KAFKA-21309?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Lianet Magrans updated KAFKA-21309:
-----------------------------------
Fix Version/s: 4.5.0
> 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
> Components: clients
> Affects Versions: 4.1.0, 4.2.0, 4.3.1
> Reporter: James Palmer
> Assignee: James Palmer
> Priority: Major
> Fix For: 4.5.0
>
>
> 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. [edit:] pr
> https://github.com/apache/kafka/pull/23870
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
