dajac commented on code in PR #12005:
URL: https://github.com/apache/kafka/pull/12005#discussion_r845476205
##########
core/src/main/scala/kafka/server/ReplicaFetcherThread.scala:
##########
@@ -213,62 +174,6 @@ class ReplicaFetcherThread(name: String,
"equal or larger than your settings for max.message.bytes, both at a
broker and topic level.")
}
-
- override protected def fetchFromLeader(fetchRequest: FetchRequest.Builder):
Map[TopicPartition, FetchData] = {
- val clientResponse = try {
- leaderEndpoint.sendRequest(fetchRequest)
- } catch {
- case t: Throwable =>
- fetchSessionHandler.handleError(t)
- throw t
- }
- val fetchResponse = clientResponse.responseBody.asInstanceOf[FetchResponse]
- if (!fetchSessionHandler.handleResponse(fetchResponse,
clientResponse.requestHeader().apiVersion())) {
- // If we had a session topic ID related error, throw it, otherwise
return an empty fetch data map.
- if (fetchResponse.error == Errors.FETCH_SESSION_TOPIC_ID_ERROR) {
- throw Errors.forCode(fetchResponse.error().code()).exception()
- } else {
- Map.empty
- }
- } else {
- fetchResponse.responseData(fetchSessionHandler.sessionTopicNames,
clientResponse.requestHeader().apiVersion()).asScala
- }
- }
-
- override protected def fetchEarliestOffsetFromLeader(topicPartition:
TopicPartition, currentLeaderEpoch: Int): Long = {
- fetchOffsetFromLeader(topicPartition, currentLeaderEpoch,
ListOffsetsRequest.EARLIEST_TIMESTAMP)
- }
-
- override protected def fetchLatestOffsetFromLeader(topicPartition:
TopicPartition, currentLeaderEpoch: Int): Long = {
- fetchOffsetFromLeader(topicPartition, currentLeaderEpoch,
ListOffsetsRequest.LATEST_TIMESTAMP)
- }
-
- private def fetchOffsetFromLeader(topicPartition: TopicPartition,
currentLeaderEpoch: Int, earliestOrLatest: Long): Long = {
- val topic = new ListOffsetsTopic()
- .setName(topicPartition.topic)
- .setPartitions(Collections.singletonList(
- new ListOffsetsPartition()
- .setPartitionIndex(topicPartition.partition)
- .setCurrentLeaderEpoch(currentLeaderEpoch)
- .setTimestamp(earliestOrLatest)))
- val requestBuilder =
ListOffsetsRequest.Builder.forReplica(listOffsetRequestVersion, replicaId)
- .setTargetTimes(Collections.singletonList(topic))
-
- val clientResponse = leaderEndpoint.sendRequest(requestBuilder)
- val response =
clientResponse.responseBody.asInstanceOf[ListOffsetsResponse]
- val responsePartition = response.topics.asScala.find(_.name ==
topicPartition.topic).get
- .partitions.asScala.find(_.partitionIndex ==
topicPartition.partition).get
-
- Errors.forCode(responsePartition.errorCode) match {
- case Errors.NONE =>
- if (brokerConfig.interBrokerProtocolVersion >= KAFKA_0_10_1_IV2)
- responsePartition.offset
- else
- responsePartition.oldStyleOffsets.get(0)
- case error => throw error.exception
- }
- }
-
override def buildFetch(partitionMap: Map[TopicPartition,
PartitionFetchState]): ResultWithPartitions[Option[ReplicaFetch]] = {
Review Comment:
Have we considered pushing this one to the trait as well? It is
implementation is very dependant on the type of the endpoint so that might be
interesting. This is also the only place where `fetchSessionHandler` is used in
this class. We could have it all handled in the endpoint.
--
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.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
