[SPARK-3146][Streaming] Improve the flexibility of Spark Streaming Kafka API

Improve the flexibility of Spark Streaming Kafka API to offer user the ability to pre-process message before stored into BlockGenerator, this will potential bring some benefits, details can be seen in [SPARK-3146](https://issues.apache.org/jira/browse/SPARK-3146).