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

a cleaner logic is to use min(curr +mmp, v)