Right now Kafka source v2 doesn't support null keys or values.
- When processing a null key, all of the following keys in the same partition will be null. This is a correctness bug.
- When processing a null value, it will throw NPE.
The workaround is setting sql conf "spark.sql.streaming.disabledV2MicroBatchReaders" to "org.apache.spark.sql.kafka010.KafkaSourceProvider" to use the v1 source.