我正在使用spark结构化流媒体将记录发送到Kafka主题。Kafka主题是用配置创建的- message.timestamp.type=CreateTime 这样做是为了使目标Kafka主题记录与原始记录具有相同的时间戳。
我的Kafka流代码：

kafkaRecords.selectExpr("CAST(key AS STRING)", "CAST(value AS BINARY)","CAST(timestamp AS TIMESTAMP)")
    .write
    .format("kafka")
    .option("kafka.bootstrap.servers","IP Of kafka")
    .option("topic",targetTopic)
    .option("kafka.max.in.flight.requests.per.connection", "1")
    .option("checkpointLocation",checkPointLocation)
    .save()

但是，这并不保留原始时间戳2018/11/04，而是时间戳反映了最新日期2018/11/9。
另一方面，为了确认kafka config正在运行，当我显式地创建一个kafka生产者和生产者记录，并将其发送给其他人时，原始的时间戳将被保留。
我怎样才能在Kafka结构化流媒体中得到同样的行为。