ijuma commented on code in PR #13056:
URL: https://github.com/apache/kafka/pull/13056#discussion_r1059261143
##########
core/src/main/scala/kafka/utils/Throttler.scala:
##########
@@ -27,32 +27,35 @@ import scala.math._
/**
* A class to measure and throttle the rate of some process. The throttler
takes a desired rate-per-second
- * (the units of the process don't matter, it could be bytes or a count of
some other thing), and will sleep for
- * an appropriate amount of time when maybeThrottle() is called to attain the
desired rate.
+ * (the units of the process don't matter, it could be bytes or a count of
some other thing), and will sleep for
+ * an appropriate amount of time when {@link maybeThrottle(double)} is called
to attain the desired rate.
*
* @param desiredRatePerSec: The rate we want to hit in units/sec
* @param checkIntervalMs: The interval at which to check our rate
* @param throttleDown: Does throttling increase or decrease our rate?
+ * @param metricName: The metric name for this throttler
* @param time: The time implementation to use
+ * @param units: The units of the {@code time}
*/
@threadsafe
class Throttler(@volatile var desiredRatePerSec: Double,
- checkIntervalMs: Long = 100L,
+ checkIntervalMs: Long = 300L,
throttleDown: Boolean = true,
- metricName: String = "throttler",
- units: String = "entries",
+ metricName: String = "cleaner-io",
+ units: String = "bytes",
time: Time = Time.SYSTEM) extends Logging with
KafkaMetricsGroup {
-
+
private val lock = new Object
private val meter = newMeter(metricName, units, TimeUnit.SECONDS)
private val checkIntervalNs = TimeUnit.MILLISECONDS.toNanos(checkIntervalMs)
- private var periodStartNs: Long = time.nanoseconds
+ private var periodStartNs: Long = 0
private var observedSoFar: Double = 0.0
-
+ private val msPerSec = TimeUnit.SECONDS.toMillis(1)
+ private val nsPerSec = TimeUnit.SECONDS.toNanos(1)
+
+ // Before calling maybeThrottle(), remember to run initStartTimeNs() once to
make sure the periodStartNs is correct
def maybeThrottle(observed: Double): Unit = {
- val msPerSec = TimeUnit.SECONDS.toMillis(1)
- val nsPerSec = TimeUnit.SECONDS.toNanos(1)
Review Comment:
They should go in the `object` since they are not associated with an
instance.
--
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
