Secondo l'errore, hai già una stringa, (hai già fatto df.selectExpr("CAST(value AS STRING)") ), quindi dovresti provare a ottenere l'evento Row come String , e non un Array[Byte]
Inizia modificando
val valueStr = new String(record.getAs[Array[Byte]]("value"))
a
val valueStr = record.getAs[String]("value")
Capisco che potresti già avere un cluster per eseguire il codice Spark, ma ti suggerirei di esaminare ancora il Connettore per lavello Kafka Connect Mongo in modo da non dover scrivere e mantenere il proprio scrittore Mongo nel codice Spark.
Oppure puoi scrivere anche i dataset Spark su mongo direttamente
