Cherry-pick two commits for Spark-672 (apache#661)

* Update package.scala (apache#650)

* Update package.scala

- We need to change the API access point to use `InputDStream` instead of `DirectKafkaInputDStream` since this is private, but it extends `InputDStream`
- The same functionality is there and the same principles apply: We can commit offsets directly without the boilerplate code but as always, it can only be done on the original Stream and NOT in the transformations

* Update package.scala

* Branch 2.4.4 mapr offsets fix (apache#654)

* Update package.scala

- We need to change the API access point to use `InputDStream` instead of `DirectKafkaInputDStream` since this is private, but it extends `InputDStream`
- The same functionality is there and the same principles apply: We can commit offsets directly without the boilerplate code but as always, it can only be done on the original Stream and NOT in the transformations

* Update package.scala

* Using InputDStream of ConsumerRecord

Co-authored-by: Mikhail Gorbov <[email protected]>

Co-authored-by: Nicolas A Perez <[email protected]>
Co-authored-by: Mikhail Gorbov <[email protected]>

Author: 3 people

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

@@ -20,7 +20,9 @@ package org.apache.spark.streaming
 import org.apache.spark.internal.config.ConfigBuilder
 
 import org.apache.kafka.clients.consumer.OffsetCommitCallback
-import org.apache.spark.streaming.kafka010.{CanCommitOffsets, DirectKafkaInputDStream, HasOffsetRanges}
+import org.apache.spark.streaming.kafka010.{CanCommitOffsets, HasOffsetRanges}
+import org.apache.spark.streaming.dstream.InputDStream
+import org.apache.kafka.clients.consumer.ConsumerRecord
 
 /**
  * Spark Integration for Kafka 0.9
@@ -86,10 +88,10 @@ package object kafka010 { //scalastyle:ignore
     def commitOffsetsAsync(): Unit = commitOffsetsAsync(null)
 
     def commitOffsetsAsync(callback: OffsetCommitCallback): Unit = {
-      directKafkaInputDStream.foreachRDD { rdd =>
+      inputDStream.foreachRDD { rdd =>
         val offsets = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
 
-        directKafkaInputDStream.asInstanceOf[CanCommitOffsets].commitAsync(offsets)
+        inputDStream.asInstanceOf[CanCommitOffsets].commitAsync(offsets)
       }
     }
   }