jolshan commented on a change in pull request #9944:
URL: https://github.com/apache/kafka/pull/9944#discussion_r651331247
##########
File path: core/src/main/scala/kafka/server/ReplicaFetcherThread.scala
##########
@@ -216,12 +217,21 @@ class ReplicaFetcherThread(name: String,
try {
val clientResponse = leaderEndpoint.sendRequest(fetchRequest)
val fetchResponse =
clientResponse.responseBody.asInstanceOf[FetchResponse]
- if (!fetchSessionHandler.handleResponse(fetchResponse)) {
- Map.empty
+ if (!fetchSessionHandler.handleResponse(fetchResponse,
clientResponse.requestHeader().apiVersion())) {
+ if (fetchResponse.error() == Errors.UNKNOWN_TOPIC_ID)
+ throw new UnknownTopicIdException("There was a topic ID in the
request that was unknown to the server.")
+ else if (fetchResponse.error() == Errors.FETCH_SESSION_TOPIC_ID_ERROR)
+ throw new FetchSessionTopicIdException("There was a topic ID in the
request that was inconsistent with the session.")
+ else
+ Map.empty
} else {
- fetchResponse.responseData.asScala
+ fetchResponse.responseData(fetchSessionHandler.sessionTopicNames,
clientResponse.requestHeader().apiVersion()).asScala
}
} catch {
+ case unknownId: UnknownTopicIdException =>
+ throw unknownId
+ case sessionUnknownId: FetchSessionTopicIdException =>
+ throw sessionUnknownId
Review comment:
This happens inside of `fetchSessionHandler.handleResponse`. We set the
session to close upon the next request. The code path for Fetcher is slightly
different so it made sense for that code to have it there.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
