mimaison commented on code in PR #19064:
URL: https://github.com/apache/kafka/pull/19064#discussion_r2003090541
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1723,7 +1723,7 @@ class ReplicaManager(val config: KafkaConfig,
metadata => findPreferredReadReplica(partition, metadata,
params.replicaId, fetchInfo.fetchOffset, fetchTimeMs))
if (preferredReadReplica.isDefined) {
- replicaSelectorOpt.foreach { selector =>
+ replicaSelectorPlugin.foreach { selector =>
debug(s"Replica selector ${selector.getClass.getSimpleName}
returned preferred replica " +
Review Comment:
This needs updating, now `selector` is of type `Plugin` here.
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1921,7 +1921,7 @@ class ReplicaManager(val config: KafkaConfig,
replicaInfoSet.add(leaderReplica)
val partitionInfo = new DefaultPartitionView(replicaInfoSet.asJava,
leaderReplica)
- replicaSelector.select(partition.topicPartition, clientMetadata,
partitionInfo).toScala.collect {
+ replicaSelector.get().select(partition.topicPartition,
clientMetadata, partitionInfo).toScala.collect {
Review Comment:
`get()` -> `get`
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -6211,3 +6225,12 @@ class MockReplicaSelector extends ReplicaSelector {
Optional.of(partitionView.leader)
}
}
+
+
+class MonitorableReplicaSelector extends MockReplicaSelector with Monitorable {
+ var pluginMetrics = false
+
+ override def withPluginMetrics(metrics: PluginMetrics): Unit = {
+ pluginMetrics = true
+ }
+}
Review Comment:
Let's keep the newline at the end of the file
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -2587,11 +2587,11 @@ class ReplicaManager(val config: KafkaConfig,
new ReplicaAlterLogDirsManager(config, this, quotaManager,
brokerTopicStats, directoryEventHandler)
}
- private def createReplicaSelector(): Option[ReplicaSelector] = {
+ private def createReplicaSelector(metrics: Metrics):
Option[Plugin[ReplicaSelector]] = {
config.replicaSelectorClassName.map { className =>
val tmpReplicaSelector: ReplicaSelector = Utils.newInstance(className,
classOf[ReplicaSelector])
tmpReplicaSelector.configure(config.originals())
- tmpReplicaSelector
+ Plugin.wrapInstance(tmpReplicaSelector, metrics, className)
Review Comment:
This is not aligned correctly
--
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
