ahuang98 commented on code in PR #19664:
URL: https://github.com/apache/kafka/pull/19664#discussion_r2127073847
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -265,33 +265,52 @@ class KafkaApis(val requestChannel: RequestChannel,
}
def handleGetReplicaLogInfo(request: RequestChannel.Request): Unit = {
- authHelper.authorizeClusterOperation(request, CLUSTER_ACTION)
+ var partitionCount = 0
+ def processPartitions(topicLogInfo:
GetReplicaLogInfoResponseData.TopicPartitionLogInfo,
+ partitionIter: util.Iterator[Integer],
+ action: Integer =>
GetReplicaLogInfoResponseData.PartitionLogInfo): Unit = {
+ while (partitionIter.hasNext && partitionCount <
GetReplicaLogInfoRequest.MAX_PARTITIONS_PER_REQUEST) {
+ topicLogInfo.partitionLogInfo().add(action(partitionIter.next()))
+ partitionCount += 1
+ }
+ }
+
+ val isClusterAction = authorizeClusterOperation(request, CLUSTER_ACTION)
Review Comment:
nit: `isClusterAction` doesn't seem to describe what this val represents.
could this be renamed to something like `isAuthorizedAsClusterAction`? know
it's a bit bulky but it's a bit more accurate
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -265,33 +265,52 @@ class KafkaApis(val requestChannel: RequestChannel,
}
def handleGetReplicaLogInfo(request: RequestChannel.Request): Unit = {
- authHelper.authorizeClusterOperation(request, CLUSTER_ACTION)
+ var partitionCount = 0
+ def processPartitions(topicLogInfo:
GetReplicaLogInfoResponseData.TopicPartitionLogInfo,
+ partitionIter: util.Iterator[Integer],
+ action: Integer =>
GetReplicaLogInfoResponseData.PartitionLogInfo): Unit = {
+ while (partitionIter.hasNext && partitionCount <
GetReplicaLogInfoRequest.MAX_PARTITIONS_PER_REQUEST) {
+ topicLogInfo.partitionLogInfo().add(action(partitionIter.next()))
+ partitionCount += 1
+ }
+ }
+
+ val isClusterAction = authorizeClusterOperation(request, CLUSTER_ACTION)
+ def isAuthorized(topicName: String): Boolean =
+ isClusterAction || authHelper.authorize(request.context, DESCRIBE,
TOPIC, topicName)
Review Comment:
I mention this because I wonder if it's extraneous to check `CLUSTER_ACTION`
before `DESCRIBE`
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -265,33 +265,52 @@ class KafkaApis(val requestChannel: RequestChannel,
}
def handleGetReplicaLogInfo(request: RequestChannel.Request): Unit = {
- authHelper.authorizeClusterOperation(request, CLUSTER_ACTION)
+ var partitionCount = 0
+ def processPartitions(topicLogInfo:
GetReplicaLogInfoResponseData.TopicPartitionLogInfo,
+ partitionIter: util.Iterator[Integer],
+ action: Integer =>
GetReplicaLogInfoResponseData.PartitionLogInfo): Unit = {
+ while (partitionIter.hasNext && partitionCount <
GetReplicaLogInfoRequest.MAX_PARTITIONS_PER_REQUEST) {
+ topicLogInfo.partitionLogInfo().add(action(partitionIter.next()))
+ partitionCount += 1
+ }
+ }
+
+ val isClusterAction = authorizeClusterOperation(request, CLUSTER_ACTION)
+ def isAuthorized(topicName: String): Boolean =
+ isClusterAction || authHelper.authorize(request.context, DESCRIBE,
TOPIC, topicName)
Review Comment:
I would have thought that anyone with CLUSTER_ACTION permission would have
DESCRIBE permission (similar to how I would think that ALTER would necessitate
DESCRIBE), but guess that kind of ACL dependency is not actually enforced in
code.
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -265,33 +265,52 @@ class KafkaApis(val requestChannel: RequestChannel,
}
def handleGetReplicaLogInfo(request: RequestChannel.Request): Unit = {
- authHelper.authorizeClusterOperation(request, CLUSTER_ACTION)
+ var partitionCount = 0
+ def processPartitions(topicLogInfo:
GetReplicaLogInfoResponseData.TopicPartitionLogInfo,
+ partitionIter: util.Iterator[Integer],
+ action: Integer =>
GetReplicaLogInfoResponseData.PartitionLogInfo): Unit = {
+ while (partitionIter.hasNext && partitionCount <
GetReplicaLogInfoRequest.MAX_PARTITIONS_PER_REQUEST) {
+ topicLogInfo.partitionLogInfo().add(action(partitionIter.next()))
+ partitionCount += 1
+ }
+ }
+
+ val isClusterAction = authorizeClusterOperation(request, CLUSTER_ACTION)
+ def isAuthorized(topicName: String): Boolean =
+ isClusterAction || authHelper.authorize(request.context, DESCRIBE,
TOPIC, topicName)
+
val getReplicaLogInfoRequest = request.body[GetReplicaLogInfoRequest]
val data = getReplicaLogInfoRequest.data()
- var partitionCount = 0
- val topicPartitionIter = data.topicPartitions().iterator()
+
+ val topicIter = data.topicPartitions().iterator()
+ var maybePartitionIter: Option[util.Iterator[Integer]] = None
val responseData = new GetReplicaLogInfoResponseData()
.setBrokerEpoch(brokerEpochSupplier.get())
- while (topicPartitionIter.hasNext && partitionCount <
GetReplicaLogInfoRequest.MAX_PARTITIONS_PER_REQUEST) {
- val topic = topicPartitionIter.next()
- val maybeTopicName = metadataCache.getTopicName(topic.topicId())
- val topicPartitionLogInfo = new
GetReplicaLogInfoResponseData.TopicPartitionLogInfo()
- .setTopicId(topic.topicId())
+
+ while (topicIter.hasNext && partitionCount <
GetReplicaLogInfoRequest.MAX_PARTITIONS_PER_REQUEST) {
+ val topic = topicIter.next()
val partitionIter = topic.partitions().iterator()
+ maybePartitionIter = Some(partitionIter)
Review Comment:
ah, I was confused why we needed both the partitionIter and the wrapped
optional version of it, but I see how you use them differently.
wonder if it might be a bit more readable if we did something like the
following instead?
```
maybePartitionIter = Some(topic.partitions().iterator())
val partitionIter = maybePartitionIter.getOrElse(throw new...
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
