[SPARK-12177][Streaming][Kafka] limit LocationStrategy api surface area

external/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/LocationStrategy.scala

@@ -36,42 +36,47 @@ import org.apache.spark.annotation.Experimental
 @Experimental
 sealed trait LocationStrategy
 
-/**
- *  :: Experimental ::
- * Use this only if your executors are on the same nodes as your Kafka brokers.
- */
-@Experimental
-case object PreferBrokers extends LocationStrategy {
-  def create: PreferBrokers.type = this
-}
+private case object PreferBrokers extends LocationStrategy
 
-/**
- *  :: Experimental ::
- * Use this in most cases, it will consistently distribute partitions across all executors.
- */
-@Experimental
-case object PreferConsistent extends LocationStrategy {
-  def create: PreferConsistent.type = this
-}
+private case object PreferConsistent extends LocationStrategy
 
-/**
- *  :: Experimental ::
- * Use this to place particular TopicPartitions on particular hosts if your load is uneven.
- * Any TopicPartition not specified in the map will use a consistent location.
- */
-@Experimental
-case class PreferFixed private(hostMap: ju.Map[TopicPartition, String]) extends LocationStrategy
+private case class PreferFixed(hostMap: ju.Map[TopicPartition, String]) extends LocationStrategy
 
 /**
- *  :: Experimental ::
- * Use this to place particular TopicPartitions on particular hosts if your load is uneven.
- * Any TopicPartition not specified in the map will use a consistent location.
+ * :: Experimental :: object to obtain instances of [[LocationStrategy]]
+ *
  */
 @Experimental
-object PreferFixed {
-  def apply(hostMap: collection.Map[TopicPartition, String]): PreferFixed = {
+object LocationStrategies {
+  /**
+   *  :: Experimental ::
+   * Use this only if your executors are on the same nodes as your Kafka brokers.
+   */
+  @Experimental
+  def preferBrokers: LocationStrategy = PreferBrokers
+
+  /**
+   *  :: Experimental ::
+   * Use this in most cases, it will consistently distribute partitions across all executors.
+   */
+  @Experimental
+  def preferConsistent: LocationStrategy = PreferConsistent
+
+  /**
+   *  :: Experimental ::
+   * Use this to place particular TopicPartitions on particular hosts if your load is uneven.
+   * Any TopicPartition not specified in the map will use a consistent location.
+   */
+  @Experimental
+  def preferFixed(hostMap: collection.Map[TopicPartition, String]): LocationStrategy =
     PreferFixed(new ju.HashMap[TopicPartition, String](hostMap.asJava))
-  }
-  def create(hostMap: ju.Map[TopicPartition, String]): PreferFixed =
+
+  /**
+   *  :: Experimental ::
+   * Use this to place particular TopicPartitions on particular hosts if your load is uneven.
+   * Any TopicPartition not specified in the map will use a consistent location.
+   */
+  @Experimental
+  def preferFixed(hostMap: ju.Map[TopicPartition, String]): LocationStrategy =
     PreferFixed(hostMap)
 }

external/kafka-0-10/src/test/java/org/apache/spark/streaming/kafka010/JavaDirectKafkaStreamSuite.java

@@ -90,7 +90,7 @@ public void testKafkaStream() throws InterruptedException {
 
     JavaInputDStream<ConsumerRecord<String, String>> istream1 = KafkaUtils.createDirectStream(
         ssc,
-        PreferConsistent.create(),
+        LocationStrategies.preferConsistent(),
         Subscribe.<String, String>create(Arrays.asList(topic1), kafkaParams)
     );
 
@@ -123,7 +123,7 @@ public String call(ConsumerRecord<String, String> r) {
 
     JavaInputDStream<ConsumerRecord<String, String>> istream2 = KafkaUtils.createDirectStream(
         ssc,
-        PreferConsistent.create(),
+        LocationStrategies.preferConsistent(),
         Subscribe.<String, String>create(Arrays.asList(topic2), kafkaParams2)
     );
 

external/kafka-0-10/src/test/java/org/apache/spark/streaming/kafka010/JavaKafkaRDDSuite.java

@@ -96,14 +96,14 @@ public String call(ConsumerRecord<String, String> r) {
         sc,
         kafkaParams,
         offsetRanges,
-        PreferFixed.create(leaders)
+        LocationStrategies.preferFixed(leaders)
     ).map(handler);
 
     JavaRDD<String> rdd2 = KafkaUtils.<String, String>createRDD(
         sc,
         kafkaParams,
         offsetRanges,
-        PreferConsistent.create()
+        LocationStrategies.preferConsistent()
     ).map(handler);
 
     // just making sure the java user apis work; the scala tests handle logic corner cases

external/kafka-0-10/src/test/java/org/apache/spark/streaming/kafka010/JavaLocationStrategySuite.java

@@ -41,18 +41,19 @@ public void testLocationStrategyConstructors() {
       JavaConverters.mapAsScalaMapConverter(hosts).asScala();
 
     // make sure constructors can be called from java
-    final LocationStrategy c1 = PreferConsistent.create();
-    final LocationStrategy c2 = PreferConsistent$.MODULE$;
-    Assert.assertEquals(c1, c2);
+    final LocationStrategy c1 = LocationStrategies.preferConsistent();
+    final LocationStrategy c2 = LocationStrategies.preferConsistent();
+    Assert.assertSame(c1, c2);
 
-    final LocationStrategy c3 = PreferBrokers.create();
-    final LocationStrategy c4 = PreferBrokers$.MODULE$;
-    Assert.assertEquals(c3, c4);
+    final LocationStrategy c3 = LocationStrategies.preferBrokers();
+    final LocationStrategy c4 = LocationStrategies.preferBrokers();
+    Assert.assertSame(c3, c4);
 
-    final LocationStrategy c5 = PreferFixed.create(hosts);
-    final LocationStrategy c6 = PreferFixed.apply(sHosts);
-    Assert.assertEquals(c5, c6);
+    Assert.assertNotSame(c1, c3);
 
+    final LocationStrategy c5 = LocationStrategies.preferFixed(hosts);
+    final LocationStrategy c6 = LocationStrategies.preferFixed(sHosts);
+    Assert.assertEquals(c5, c6);
   }
 
 }

external/kafka-0-10/src/test/scala/org/apache/spark/streaming/kafka010/DirectKafkaStreamSuite.scala

@@ -93,7 +93,7 @@ class DirectKafkaStreamSuite
     kp
   }
 
-  val preferredHosts = PreferConsistent
+  val preferredHosts = LocationStrategies.preferConsistent
 
   test("basic stream receiving with multiple topics and smallest starting offset") {
     val topics = List("basic1", "basic2", "basic3")

external/kafka-0-10/src/test/scala/org/apache/spark/streaming/kafka010/KafkaRDDSuite.scala

@@ -62,7 +62,7 @@ class KafkaRDDSuite extends SparkFunSuite with BeforeAndAfterAll {
     "group.id" -> s"test-consumer-${Random.nextInt}-${System.currentTimeMillis}"
   ).asJava
 
-  private val preferredHosts = PreferConsistent
+  private val preferredHosts = LocationStrategies.preferConsistent
 
   test("basic usage") {
     val topic = s"topicbasic-${Random.nextInt}-${System.currentTimeMillis}"