[SPARK-4964] settle on "Direct" as a naming convention for the new st…

…ream

external/kafka/src/main/scala/org/apache/spark/streaming/kafka/DeterministicKafkaInputDStream.scala → external/kafka/src/main/scala/org/apache/spark/streaming/kafka/DirectKafkaInputDStream.scala

@@ -53,7 +53,7 @@ import org.apache.spark.streaming.dstream._
  * @param maxRetries maximum number of times in a row to retry getting leaders' offsets
  */
 private[streaming]
-class DeterministicKafkaInputDStream[
+class DirectKafkaInputDStream[
   K: ClassTag,
   V: ClassTag,
   U <: Decoder[_]: ClassTag,
@@ -68,7 +68,7 @@ class DeterministicKafkaInputDStream[
     "spark.streaming.kafka.maxRetries", 1)
 
   protected[streaming] override val checkpointData =
-    new DeterministicKafkaInputDStreamCheckpointData
+    new DirectKafkaInputDStreamCheckpointData
 
   protected val kc = new KafkaCluster(kafkaParams)
 
@@ -129,7 +129,7 @@ class DeterministicKafkaInputDStream[
   }
 
   private[streaming]
-  class DeterministicKafkaInputDStreamCheckpointData extends DStreamCheckpointData(this) {
+  class DirectKafkaInputDStreamCheckpointData extends DStreamCheckpointData(this) {
     def batchForTime = data.asInstanceOf[mutable.HashMap[
       Time, Array[OffsetRange.OffsetRangeTuple]]]
 

external/kafka/src/main/scala/org/apache/spark/streaming/kafka/KafkaUtils.scala

@@ -246,7 +246,7 @@ object KafkaUtils {
    *  starting point of the stream
    */
   @Experimental
-  def createNewStream[
+  def createDirectStream[
     K: ClassTag,
     V: ClassTag,
     U <: Decoder[_]: ClassTag,
@@ -257,7 +257,7 @@ object KafkaUtils {
       fromOffsets: Map[TopicAndPartition, Long],
       messageHandler: MessageAndMetadata[K, V] => R
   ): InputDStream[R] = {
-    new DeterministicKafkaInputDStream[K, V, U, T, R](
+    new DirectKafkaInputDStream[K, V, U, T, R](
       ssc, kafkaParams, fromOffsets, messageHandler)
   }
 
@@ -289,7 +289,7 @@ object KafkaUtils {
    * @param topics names of the topics to consume
    */
   @Experimental
-  def createNewStream[
+  def createDirectStream[
     K: ClassTag,
     V: ClassTag,
     U <: Decoder[_]: ClassTag,
@@ -313,7 +313,7 @@ object KafkaUtils {
       val fromOffsets = leaderOffsets.map { case (tp, lo) =>
           (tp, lo.offset)
       }
-      new DeterministicKafkaInputDStream[K, V, U, T, (K, V)](
+      new DirectKafkaInputDStream[K, V, U, T, (K, V)](
         ssc, kafkaParams, fromOffsets, messageHandler)
     }).fold(
       errs => throw new SparkException(errs.mkString("\n")),

external/kafka/src/test/scala/org/apache/spark/streaming/kafka/KafkaNewStreamSuite.scala → external/kafka/src/test/scala/org/apache/spark/streaming/kafka/KafkaDirectStreamSuite.scala

@@ -29,7 +29,7 @@ import org.apache.spark.SparkConf
 import org.apache.spark.storage.StorageLevel
 import org.apache.spark.streaming.{Milliseconds, StreamingContext}
 
-class KafkaNewStreamSuite extends KafkaStreamSuiteBase with BeforeAndAfter with Eventually {
+class KafkaDirectStreamSuite extends KafkaStreamSuiteBase with BeforeAndAfter with Eventually {
   val sparkConf = new SparkConf()
     .setMaster("local[4]")
     .setAppName(this.getClass.getSimpleName)
@@ -63,7 +63,7 @@ class KafkaNewStreamSuite extends KafkaStreamSuiteBase with BeforeAndAfter with
       createTopic(t)
       produceAndSendMessage(t, data)
     }
-    val stream = KafkaUtils.createNewStream[String, String, StringDecoder, StringDecoder](
+    val stream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
       ssc, kafkaParams, topics)
     var total = 0L;