Según el error, ya tiene una cadena (ya hizo df.selectExpr("CAST(value AS STRING)") ), por lo que debería intentar obtener el evento Row como String , y no un Array[Byte]
Comience por cambiar
val valueStr = new String(record.getAs[Array[Byte]]("value"))
a
val valueStr = record.getAs[String]("value")
Entiendo que es posible que ya tenga un clúster para ejecutar el código Spark, pero le sugiero que siga investigando Conector de lavabo Kafka Connect Mongo para que no tenga que escribir y mantener su propio escritor Mongo en código Spark.
O bien, puede escribir conjuntos de datos de Spark en mongo directamente también
