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

for example

https://github.com/koeninger/kafka-exactly-once/blob/master/src/main/scala/example/IdempotentExample.scala#L60

We also use it for doing things like e.g. starting a stream at the leader offsets before a given time