You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "xijiu (via GitHub)" <gi...@apache.org> on 2024/10/21 09:05:05 UTC
[PR] KAFKA-17804: Optimize ReplicaManager.completeDelayedOperationsWhenNotPartitionLeader [kafka]
xijiu opened a new pull request, #17563:
URL: https://github.com/apache/kafka/pull/17563
As title
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17804: (DO NOT MERGE)Optimize ReplicaManager.completeDelayedOperationsWhenNotPartitionLeader [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #17563:
URL: https://github.com/apache/kafka/pull/17563#discussion_r1813912519
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -657,7 +661,7 @@ class ReplicaManager(val config: KafkaConfig,
}
// If we were the leader, we may have some operations still waiting for completion.
// We force completion to prevent them from timing out.
- completeDelayedOperationsWhenNotPartitionLeader(topicPartition, topicId)
+ completeDelayedOperationsWhenNotPartitionLeader(topicPartition, topicId, isStopReplica = true)
Review Comment:
We can move this check inside `if (stopPartition.deleteLocalLog) `.
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -3072,7 +3076,7 @@ class ReplicaManager(val config: KafkaConfig,
stateChangeLogger.info(s"Started fetchers as part of become-follower for ${partitionsToStartFetching.size} partitions")
partitionsToStartFetching.foreach{ case (topicPartition, partition) =>
- completeDelayedOperationsWhenNotPartitionLeader(topicPartition, partition.topicId)}
+ completeDelayedOperationsWhenNotPartitionLeader(topicPartition, partition.topicId, isStopReplica = false)}
Review Comment:
We can optimize this a bit more. We only need to make this call if the replica is now a follower and previously a leader.
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -469,14 +469,18 @@ class ReplicaManager(val config: KafkaConfig,
})
}
- private def completeDelayedOperationsWhenNotPartitionLeader(topicPartition: TopicPartition, topicId: Option[Uuid]): Unit = {
+ private def completeDelayedOperationsWhenNotPartitionLeader(topicPartition: TopicPartition, topicId: Option[Uuid],
Review Comment:
Perhaps it's clearer to split this into 2 methods, one for NotLeader and another NotReplica?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17804: Optimize ReplicaManager.completeDelayedOperationsWhenNotPartitionLeader [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #17563:
URL: https://github.com/apache/kafka/pull/17563#discussion_r1809214823
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -469,14 +469,18 @@ class ReplicaManager(val config: KafkaConfig,
})
}
- private def completeDelayedOperationsWhenNotPartitionLeader(topicPartition: TopicPartition, topicId: Option[Uuid]): Unit = {
+ private def completeDelayedOperationsWhenNotPartitionLeader(topicPartition: TopicPartition, topicId: Option[Uuid],
+ isStopReplica: Boolean): Unit = {
val topicPartitionOperationKey = TopicPartitionOperationKey(topicPartition)
delayedProducePurgatory.checkAndComplete(topicPartitionOperationKey)
- delayedFetchPurgatory.checkAndComplete(topicPartitionOperationKey)
- delayedRemoteFetchPurgatory.checkAndComplete(topicPartitionOperationKey)
- delayedRemoteListOffsetsPurgatory.checkAndComplete(topicPartitionOperationKey)
- if (topicId.isDefined) delayedShareFetchPurgatory.checkAndComplete(
- new DelayedShareFetchPartitionKey(topicId.get, topicPartition.partition()))
+ // We support fetch from follower now, So only 'stop reply' requires the following things
+ if (isStopReplica) {
+ delayedFetchPurgatory.checkAndComplete(topicPartitionOperationKey)
Review Comment:
Pardon me, I have a question. Some tests case are checking that the produce failed because it changed to follower before replicating, if we won't completion this fetch purgatory, and we can't observe the error about `NOT_LEADER_OR_FOLLOWER`. WDYT?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17804: (DO NOT MERGE)Optimize ReplicaManager.completeDelayedOperationsWhenNotPartitionLeader [kafka]
Posted by "xijiu (via GitHub)" <gi...@apache.org>.
xijiu commented on code in PR #17563:
URL: https://github.com/apache/kafka/pull/17563#discussion_r1810103650
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -469,14 +469,18 @@ class ReplicaManager(val config: KafkaConfig,
})
}
- private def completeDelayedOperationsWhenNotPartitionLeader(topicPartition: TopicPartition, topicId: Option[Uuid]): Unit = {
+ private def completeDelayedOperationsWhenNotPartitionLeader(topicPartition: TopicPartition, topicId: Option[Uuid],
+ isStopReplica: Boolean): Unit = {
val topicPartitionOperationKey = TopicPartitionOperationKey(topicPartition)
delayedProducePurgatory.checkAndComplete(topicPartitionOperationKey)
- delayedFetchPurgatory.checkAndComplete(topicPartitionOperationKey)
- delayedRemoteFetchPurgatory.checkAndComplete(topicPartitionOperationKey)
- delayedRemoteListOffsetsPurgatory.checkAndComplete(topicPartitionOperationKey)
- if (topicId.isDefined) delayedShareFetchPurgatory.checkAndComplete(
- new DelayedShareFetchPartitionKey(topicId.get, topicPartition.partition()))
+ // We support fetch from follower now, So only 'stop reply' requires the following things
+ if (isStopReplica) {
+ delayedFetchPurgatory.checkAndComplete(topicPartitionOperationKey)
Review Comment:
@m1a2st Thanks very much for code review, I will check the failed test cases and make some modifications.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17804: (DO NOT MERGE)Optimize ReplicaManager.completeDelayedOperationsWhenNotPartitionLeader [kafka]
Posted by "xijiu (via GitHub)" <gi...@apache.org>.
xijiu closed pull request #17563: KAFKA-17804: (DO NOT MERGE)Optimize ReplicaManager.completeDelayedOperationsWhenNotPartitionLeader
URL: https://github.com/apache/kafka/pull/17563
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17804: Optimize ReplicaManager.completeDelayedOperationsWhenNotPartitionLeader [kafka]
Posted by "xijiu (via GitHub)" <gi...@apache.org>.
xijiu commented on PR #17563:
URL: https://github.com/apache/kafka/pull/17563#issuecomment-2426066572
@junrao @chia7712 PTAL
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org