Hangleton commented on code in PR #13535:
URL: https://github.com/apache/kafka/pull/13535#discussion_r1182493022
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1160,48 +1171,100 @@ class ReplicaManager(val config: KafkaConfig,
fetchPartitionStatus += (topicIdPartition ->
FetchPartitionStatus(logOffsetMetadata, partitionData))
})
}
- val delayedFetch = new DelayedFetch(
- params = params,
- fetchPartitionStatus = fetchPartitionStatus,
- replicaManager = this,
- quota = quota,
- responseCallback = responseCallback
- )
-
- // create a list of (topic, partition) pairs to use as keys for this
delayed fetch operation
- val delayedFetchKeys = fetchPartitionStatus.map { case (tp, _) =>
TopicPartitionOperationKey(tp) }
-
- // try to complete the request immediately, otherwise put it into the
purgatory;
- // this is because while the delayed fetch operation is being created,
new requests
- // may arrive and hence make this operation completable.
- delayedFetchPurgatory.tryCompleteElseWatch(delayedFetch,
delayedFetchKeys)
+
+ if (remoteFetchInfo.isPresent) {
+ val key = new
TopicPartitionOperationKey(remoteFetchInfo.get.topicPartition.topic(),
remoteFetchInfo.get.topicPartition.partition())
+ val remoteFetchResult = new CompletableFuture[RemoteLogReadResult]
+ var remoteFetchTask: Future[Void] = null
+ try {
+ remoteFetchTask =
remoteLogManager.get.asyncRead(remoteFetchInfo.get, (result:
RemoteLogReadResult) => {
+ remoteFetchResult.complete(result)
+ delayedRemoteFetchPurgatory.checkAndComplete(key)
+ })
+ } catch {
+ // if the task queue of remote storage reader thread pool is full,
return what we currently have
+ // (the data read from local log segment for the other
topic-partitions) and an error for the topic-partition that
+ // we couldn't read from remote storage
+ case e: RejectedExecutionException =>
+ val fetchPartitionData = logReadResults.map { case (tp, result) =>
+ val r = {
+ if
(tp.topicPartition().equals(remoteFetchInfo.get.topicPartition))
+ createLogReadResult(e)
Review Comment:
Should we add a log if the nature of the error is not propagated?
--
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
