[SPARK-4964] [Streaming] Exactly-once semantics for Kafka

It would probably be good to make this `private[spark]` and keep it as an internal utility.