如何在Spark Streaming 2.3.1中将每条记录写入多个kafka主题？

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.spark.streaming.kafka010._

// 创建KafkaProducer实例
val kafkaProps = new Properties()
kafkaProps.put("bootstrap.servers", "kafka-broker1:9092,kafka-broker2:9092")
kafkaProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
kafkaProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
val producer = new KafkaProducer[String, String](kafkaProps)

// 创建Spark Streaming上下文
val sparkConf = new SparkConf().setAppName("SparkStreamingExample")
val ssc = new StreamingContext(sparkConf, Seconds(1))

// 创建DStream
val stream = KafkaUtils.createDirectStream[String, String](
  ssc,
  PreferConsistent,
  Subscribe[String, String](Array("input-topic"), kafkaParams)
)

// 将每条记录写入多个Kafka主题
stream.foreachRDD { rdd =>
  rdd.foreach { record =>
    // 将记录写入第一个Kafka主题
    val topic1Record = new ProducerRecord[String, String]("topic1", record.key(), record.value())
    producer.send(topic1Record)

    // 将记录写入第二个Kafka主题
    val topic2Record = new ProducerRecord[String, String]("topic2", record.key(), record.value())
    producer.send(topic2Record)
  }
}

// 启动Spark Streaming
ssc.start()
ssc.awaitTermination()