Bagaimana untuk menghantar data json ke kafka menggunakan Java

Question

Di Internet, saya hanya melihat beberapa berita tentang Java yang menghasilkan jenis STRING.
Mengikut kelas pengeluar Java, anda boleh menyesuaikan jenis mesej yang dihantar, seperti producer.send(new KeyedMessage<String, HashMap<String, String>>(topik, mesej);
Walau bagaimanapun, jika anda menjalankan cara ini, ralat akan dilaporkan. Seperti berikut, sila berikan jawapan pakar:
Pengecualian dalam utas "Thread-4" java.lang.ClassCastException: java.util.HashMap tidak boleh dihantar ke java.lang.String

at kafka.serializer.StringEncoder.toBytes(Encoder.scala:46)
at kafka.producer.async.DefaultEventHandler$$anonfun$serialize.apply(DefaultEventHandler.scala:130)
at kafka.producer.async.DefaultEventHandler$$anonfun$serialize.apply(DefaultEventHandler.scala:125)
at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
at scala.collection.mutable.WrappedArray.foreach(WrappedArray.scala:34)
at kafka.producer.async.DefaultEventHandler.serialize(DefaultEventHandler.scala:125)
at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:52)
at kafka.producer.Producer.send(Producer.scala:77)
at kafka.javaapi.producer.Producer.send(Producer.scala:33)
at com.Model.Producer.kafkaProducer.run(kafkaProducer.java:35)

Answer 1

Dokumen, hanya bersiri ke dalam String