mesos/spark — pull request #527 by Reinvigorate (2013-03-15 06:15:09+00:00)
Title: KafkaInputDStream fixes and improvements

Improvements:
* fixes critical memory leak.
* makes available all the kafka consumer config properties (ie- autooffset.reset, see http://kafka.apache.org/configuration.html).
* removes dependency on private kafka class ZookeeperConsumerConnector.
* cleans up kafkaStream interface by removing initialOffets/setOffsets logic (now handled by kafkaParams /w autooffset.reset)

Example creating kafkaStream:
```scala
val kafkaParams = Map(
"zk.connect" -> "zoo01,zoo02,zoo03",
"groupid" -> "my-consumer-group",
"autooffset.reset" -> "smallest" // start at topic beginning
)

val inputStream = ssc.kafkaStream(kafkaParams, topics, StorageLevel.MEMORY_AND_DISK)
```