Github user pwendell commented on a diff in the pull request:
https://github.com/apache/spark/pull/3798#discussion_r23988318
--- Diff:
external/kafka/src/main/scala/org/apache/spark/streaming/kafka/KafkaUtils.scala
---
@@ -144,4 +150,174 @@ object KafkaUtils {
createStream[K, V, U, T](
jssc.ssc, kafkaParams.toMap,
Map(topics.mapValues(_.intValue()).toSeq: _*), storageLevel)
}
+
+ /** A batch-oriented interface for consuming from Kafka.
+ * Starting and ending offsets are specified in advance,
+ * so that you can control exactly-once semantics.
+ * @param sc SparkContext object
+ * @param kafkaParams Kafka <a
href="http://kafka.apache.org/documentation.html#configuration";>
+ * configuration parameters</a>.
+ * Requires "metadata.broker.list" or "bootstrap.servers" to be set
with Kafka broker(s),
+ * NOT zookeeper servers, specified in host1:port1,host2:port2 form.
+ * @param offsetRanges Each OffsetRange in the batch corresponds to a
+ * range of offsets for a given Kafka topic/partition
+ */
+ @Experimental
+ def createRDD[
+ K: ClassTag,
+ V: ClassTag,
+ U <: Decoder[_]: ClassTag,
+ T <: Decoder[_]: ClassTag] (
+ sc: SparkContext,
+ kafkaParams: Map[String, String],
+ offsetRanges: Array[OffsetRange]
+ ): RDD[(K, V)] with HasOffsetRanges = {
--- End diff --
I've never seen a trait mixin in a return type. What does this actually
mean? I looked at the compiled byte code and the byte code signature is still
RDD.
Can we just return a `KafkaRDD` here? If this is enforced somehow by the
scala compiler, returning an interface here ties our hands in the future,
because we can't add functionality to the returned type without breaking binary
compatibility. For instance, we may want to return an RDD that has additional
methods beyond just accessing its offset ranges.
I ran a simple example and I couldn't see any byte code reference to the
mixed in trait:
```
trait Trait {}
class Class extends Trait {}
object Object {
def getTrait: Class with Trait = {new Class()}
}
> javap -v Object
public static Class getTrait();
flags: ACC_PUBLIC, ACC_STATIC
Code:
stack=1, locals=0, args_size=0
0: getstatic #16 // Field
Object$.MODULE$:LObject$;
3: invokevirtual #18 // Method
Object$.getTrait:()LClass;
6: areturn
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
