Skip to content

Commit

Permalink
Renames and other nitpicks.
Browse files Browse the repository at this point in the history
  • Loading branch information
dragos committed Jul 22, 2015
1 parent 162d9e5 commit 8941cf9
Show file tree
Hide file tree
Showing 2 changed files with 20 additions and 12 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,12 @@ private[receiver] abstract class RateLimiter(conf: SparkConf) extends Logging {
def getCurrentLimit: Long =
rateLimiter.getRate.toLong

/**
* Set the rate limit to `newRate`. The new rate will not exceed the maximum rate configured by
* {{{spark.streaming.receiver.maxRate}}}, even if `newRate` is higher than that.
*
* @param newRate A new rate in events per second. It has no effect if it's 0 or negative.
*/
private[receiver] def updateRate(newRate: Long): Unit =
if (newRate > 0) {
if (maxRateLimit > 0) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,7 @@ class ReceiverTrackerSuite extends TestSuiteBase {
}

test("Receiver tracker - propagates rate limit") {
object streamingListener extends StreamingListener {
object ReceiverStartedWaiter extends StreamingListener {
@volatile
var started = false

Expand All @@ -89,32 +89,32 @@ class ReceiverTrackerSuite extends TestSuiteBase {
}
}

ssc.addStreamingListener(streamingListener)
ssc.addStreamingListener(ReceiverStartedWaiter)
ssc.scheduler.listenerBus.start(ssc.sc)

val newRateLimit = 100L
val ids = new TestReceiverInputDStream(ssc)
val inputDStream = new RateLimitInputDStream(ssc)
val tracker = new ReceiverTracker(ssc)
tracker.start()

// we wait until the Receiver has registered with the tracker,
// otherwise our rate update is lost
eventually(timeout(5 seconds)) {
assert(streamingListener.started)
assert(ReceiverStartedWaiter.started)
}
tracker.sendRateUpdate(ids.id, newRateLimit)
tracker.sendRateUpdate(inputDStream.id, newRateLimit)
// this is an async message, we need to wait a bit for it to be processed
eventually(timeout(3 seconds)) {
assert(ids.getCurrentRateLimit.get === newRateLimit)
assert(inputDStream.getCurrentRateLimit.get === newRateLimit)
}
}
}

/** An input DStream with a hard-coded receiver that gives access to internals for testing. */
private class TestReceiverInputDStream(@transient ssc_ : StreamingContext)
private class RateLimitInputDStream(@transient ssc_ : StreamingContext)
extends ReceiverInputDStream[Int](ssc_) {

override def getReceiver(): DummyReceiver = TestDummyReceiver
override def getReceiver(): DummyReceiver = SingletonDummyReceiver

def getCurrentRateLimit: Option[Long] = {
invokeExecutorMethod.getCurrentRateLimit
Expand All @@ -124,15 +124,17 @@ private class TestReceiverInputDStream(@transient ssc_ : StreamingContext)
val c = classOf[Receiver[_]]
val ex = c.getDeclaredMethod("executor")
ex.setAccessible(true)
ex.invoke(TestDummyReceiver).asInstanceOf[ReceiverSupervisor]
ex.invoke(SingletonDummyReceiver).asInstanceOf[ReceiverSupervisor]
}
}

/**
* We need the receiver to be an object, otherwise serialization will create another one
* and we won't be able to read its rate limit.
* A Receiver as an object so we can read its rate limit.
*
* @note It's necessary to be a top-level object, or else serialization would create another
* one on the executor side and we won't be able to read its rate limit.
*/
private object TestDummyReceiver extends DummyReceiver
private object SingletonDummyReceiver extends DummyReceiver

/**
* Dummy receiver implementation
Expand Down

0 comments on commit 8941cf9

Please sign in to comment.