fix invoke private usage

Author: Robert Kruszewski

core/src/main/scala/org/apache/spark/metrics/MetricsSystem.scala

@@ -159,6 +159,11 @@ private[spark] class MetricsSystem private (
     } else { defaultName }
   }
 
+  def getSources: Seq[Source] =
+    sourceToListeners.keySet.to[collection.immutable.Seq]
+
+  def getSinks: Seq[Sink] = sinks.to[collection.immutable.Seq]
+
   def getSourcesByName(sourceName: String): Seq[Source] =
     sourceToListeners.keySet.filter(_.sourceName == sourceName).toSeq
 

core/src/test/scala/org/apache/spark/metrics/MetricsSystemSuite.scala

@@ -17,8 +17,6 @@
 
 package org.apache.spark.metrics
 
-import scala.collection.mutable.ArrayBuffer
-
 import com.codahale.metrics.{Gauge, MetricRegistry}
 import org.scalatest.{BeforeAndAfter, PrivateMethodTester}
 
@@ -42,27 +40,23 @@ class MetricsSystemSuite extends SparkFunSuite with BeforeAndAfter with PrivateM
   test("MetricsSystem with default config") {
     val metricsSystem = MetricsSystem.createMetricsSystem("default", conf, securityMgr)
     metricsSystem.start()
-    val sources = PrivateMethod[ArrayBuffer[Source]](Symbol("sources"))
-    val sinks = PrivateMethod[ArrayBuffer[Sink]](Symbol("sinks"))
 
-    assert(metricsSystem.invokePrivate(sources()).length === StaticSources.allSources.length)
-    assert(metricsSystem.invokePrivate(sinks()).length === 0)
+    assert(metricsSystem.getSources.length === StaticSources.allSources.length)
+    assert(metricsSystem.getSinks.length === 0)
     assert(metricsSystem.getServletHandlers.nonEmpty)
   }
 
   test("MetricsSystem with sources add") {
     val metricsSystem = MetricsSystem.createMetricsSystem("test", conf, securityMgr)
     metricsSystem.start()
-    val sources = PrivateMethod[ArrayBuffer[Source]](Symbol("sources"))
-    val sinks = PrivateMethod[ArrayBuffer[Sink]](Symbol("sinks"))
 
-    assert(metricsSystem.invokePrivate(sources()).length === StaticSources.allSources.length)
-    assert(metricsSystem.invokePrivate(sinks()).length === 1)
+    assert(metricsSystem.getSources.length === StaticSources.allSources.length)
+    assert(metricsSystem.getSinks.length === 1)
     assert(metricsSystem.getServletHandlers.nonEmpty)
 
     val source = new MasterSource(null)
     metricsSystem.registerSource(source)
-    assert(metricsSystem.invokePrivate(sources()).length === StaticSources.allSources.length + 1)
+    assert(metricsSystem.getSources.length === StaticSources.allSources.length + 1)
   }
 
   test("MetricsSystem with Driver instance") {

streaming/src/test/scala/org/apache/spark/streaming/StreamingContextSuite.scala

@@ -374,13 +374,13 @@ class StreamingContextSuite
     addInputStream(ssc).register()
     ssc.start()
 
-    val sources = StreamingContextSuite.getSources(ssc.env.metricsSystem)
+    val sources = ssc.env.metricsSystem.getSources
     val streamingSource = StreamingContextSuite.getStreamingSource(ssc)
     assert(sources.contains(streamingSource))
     assert(ssc.getState() === StreamingContextState.ACTIVE)
 
     ssc.stop()
-    val sourcesAfterStop = StreamingContextSuite.getSources(ssc.env.metricsSystem)
+    val sourcesAfterStop = ssc.env.metricsSystem.getSources
     val streamingSourceAfterStop = StreamingContextSuite.getStreamingSource(ssc)
     assert(ssc.getState() === StreamingContextState.STOPPED)
     assert(!sourcesAfterStop.contains(streamingSourceAfterStop))
@@ -1026,10 +1026,6 @@ package object testPackage extends Assertions {
  * This includes methods to access private methods and fields in StreamingContext and MetricsSystem
  */
 private object StreamingContextSuite extends PrivateMethodTester {
-  private val _sources = PrivateMethod[ArrayBuffer[Source]](Symbol("sources"))
-  private def getSources(metricsSystem: MetricsSystem): ArrayBuffer[Source] = {
-    metricsSystem.invokePrivate(_sources())
-  }
   private val _streamingSource = PrivateMethod[StreamingSource](Symbol("streamingSource"))
   private def getStreamingSource(streamingContext: StreamingContext): StreamingSource = {
     streamingContext.invokePrivate(_streamingSource())