ijuma commented on a change in pull request #9755:
URL: https://github.com/apache/kafka/pull/9755#discussion_r546398998
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -3530,3 +3494,14 @@ class KafkaApis(val requestChannel: RequestChannel,
}
}
+
+object KafkaApis {
+ // Traffic from both in-sync and out of sync replicas are accounted for in
replication quota to ensure total replication
Review comment:
Indent is not right.
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -3530,3 +3494,14 @@ class KafkaApis(val requestChannel: RequestChannel,
}
}
+
+object KafkaApis {
+ // Traffic from both in-sync and out of sync replicas are accounted for in
replication quota to ensure total replication
+ // traffic doesn't exceed quota.
+ private[server] def sizeOfThrottledPartitions(versionId: Short,
+ unconvertedResponse:
FetchResponse[Records],
+ quota:
ReplicationQuotaManager): Int = {
+ FetchResponse.sizeOf(versionId, unconvertedResponse.responseData.entrySet()
+ .iterator().asScala.filter(element =>
quota.isThrottled(element.getKey)).asJava)
Review comment:
A couple of unnecessary ().
##########
File path: core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
##########
@@ -3067,4 +3067,31 @@ class KafkaApisTest {
Errors.LOG_DIR_NOT_FOUND -> 1,
Errors.INVALID_TOPIC_EXCEPTION -> 1).asJava, response.errorCounts)
}
+
+ @Test
+ def testSizeOfThrottledPartitions(): Unit = {
+ def fetchResponse(data: Map[TopicPartition, String]):
FetchResponse[Records] = {
+ val responseData = new util.LinkedHashMap[TopicPartition,
FetchResponse.PartitionData[Records]](data.map {
+ case (tp, raw) =>
+ tp -> new FetchResponse.PartitionData(Errors.NONE,
+ 105, 105, 0, Optional.empty(), Collections.emptyList(),
Optional.empty(),
+ MemoryRecords.withRecords(CompressionType.NONE,
+ new SimpleRecord(100,
raw.getBytes(StandardCharsets.UTF_8))).asInstanceOf[Records])
+ }.toMap.asJava)
+ new FetchResponse(Errors.NONE, responseData, 100, 100)
+ }
+
+ val throttledPartition = new TopicPartition("throttledData", 0)
+ val throttledData = Map(throttledPartition -> "throttledData")
+ val expectedSize =
FetchResponse.sizeOf(FetchResponseData.HIGHEST_SUPPORTED_VERSION,
+ fetchResponse(throttledData).responseData().entrySet().iterator)
Review comment:
Unnecessary () here and in the inner method.
##########
File path: core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
##########
@@ -3067,4 +3067,31 @@ class KafkaApisTest {
Errors.LOG_DIR_NOT_FOUND -> 1,
Errors.INVALID_TOPIC_EXCEPTION -> 1).asJava, response.errorCounts)
}
+
+ @Test
+ def testSizeOfThrottledPartitions(): Unit = {
+ def fetchResponse(data: Map[TopicPartition, String]):
FetchResponse[Records] = {
+ val responseData = new util.LinkedHashMap[TopicPartition,
FetchResponse.PartitionData[Records]](data.map {
Review comment:
I think I'd move data to the next line to make it a bit more readable.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
