Github user rezasafi commented on a diff in the pull request:
https://github.com/apache/spark/pull/22223#discussion_r213829668
--- Diff:
external/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/DirectKafkaInputDStream.scala
---
@@ -154,7 +153,8 @@ private[spark] class DirectKafkaInputDStream[K, V](
if (effectiveRateLimitPerPartition.values.sum > 0) {
val secsPerBatch = context.graph.batchDuration.milliseconds.toDouble
/ 1000
Some(effectiveRateLimitPerPartition.map {
- case (tp, limit) => tp -> Math.max((secsPerBatch * limit).toLong,
1L)
+ case (tp, limit) => tp -> Math.max((secsPerBatch * limit).toLong,
+ Math.max(ppc.minRatePerPartition(tp), 1L))
--- End diff --
I just didn't want to break the reasoning behind SPARK-18371 to have at
least 1 always. I didn't have any other reason for this. I can change it to
give the user the freedom.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
