Fix style.

lightbend · Jul 28, 2015 · 5125e60 · 5125e60
1 parent a2eb3b9
commit 5125e60
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 5 deletions.
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/dstream/ReceiverInputDStream.scala b/streaming/src/main/scala/org/apache/spark/streaming/dstream/ReceiverInputDStream.scala
@@ -44,11 +44,13 @@ abstract class ReceiverInputDStream[T: ClassTag](@transient ssc_ : StreamingCont
   /**
    * Asynchronously maintains & sends new rate limits to the receiver through the receiver tracker.
    */
-  override protected[streaming] val rateController: Option[RateController] =
-    if (RateController.isBackPressureEnabled(ssc.conf))
+  override protected[streaming] val rateController: Option[RateController] = {
+    if (RateController.isBackPressureEnabled(ssc.conf)) {
       RateEstimator.create(ssc.conf).map { new ReceiverRateController(id, _) }
-    else
+    } else {
       None
+    }
+  }
 
   /**
    * Gets the receiver object that will be sent to the worker nodes

diff --git a/streaming/src/test/scala/org/apache/spark/streaming/scheduler/RateControllerSuite.scala b/streaming/src/test/scala/org/apache/spark/streaming/scheduler/RateControllerSuite.scala
@@ -46,7 +46,7 @@ class RateControllerSuite extends TestSuiteBase {
     }
   }
 
-  test("receiver rate controller updates reach receivers") {
+  test("publish rates reach receivers") {
     val ssc = new StreamingContext(conf, batchDuration)
     withStreamingContext(ssc) { ssc =>
       val dstream = new RateLimitInputDStream(ssc) {
@@ -65,7 +65,7 @@ class RateControllerSuite extends TestSuiteBase {
     }
   }
 
-  test("multiple rate controller updates reach receivers") {
+  test("multiple publish rates reach receivers") {
     val ssc = new StreamingContext(conf, batchDuration)
     withStreamingContext(ssc) { ssc =>
       val rates = Seq(100L, 200L, 300L)