我想提取嵌入每个消息的时间戳，并将它们作为json负载发送到我的数据库中。
我想得到以下三个时间戳。
活动时间： The point in time when an event or data record occurred, i.e. was originally created “by the source”. 处理时间： The point in time when the event or data record happens to be processed by the stream processing application, i.e. when the record is being consumed. 摄入时间： The point in time when an event or data record is stored in a topic partition by a Kafka broker. 这是我的streams应用程序代码：

Properties props = new Properties();

props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-pipe");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, BROKER_URL + ":9092"); // pass from env localhost:9092 ,BROKER_URL + ":9092"
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

final StreamsBuilder builder = new StreamsBuilder();

KStream<String, String> source_o365_user_activity = builder.stream("o365_user_activity");

source_o365_user_activity.flatMapValues(new ValueMapper<String, Iterable<String>>() {
    @Override
    public Iterable<String> apply(String value) {
        System.out.println("========> o365_user_activity_by_date Log:     " + value);
        ArrayList<String> keywords = new ArrayList<String>();
        try {
            JSONObject send = new JSONObject();
            JSONObject received = new JSONObject(value);

            send.put("current_date", getCurrentDate().toString()); // UTC TIME
            send.put("activity_time", received.get("CreationTime")); // CONSTANTS FINAL STATIC(Topic Names, Cassandra keys)
            send.put("user_id", received.get("UserId"));
            send.put("operation", received.get("Operation"));
            send.put("workload", received.get("Workload"));
            keywords.add(send.toString());

        } catch (Exception e) {
            // TODO: handle exception
            System.err.println("Unable to convert to json");
            e.printStackTrace();
        }

        return keywords;
    }
}).to("o365_user_activity_by_date");

在代码中，我只是获取每条记录，对其进行流处理并将其发送到不同的主题。
现在我要发送的每一张唱片 Event-time , Processing-time 以及 Ingestion-time 嵌入有效载荷中。
我已经看过了 FailOnInvalidTimestamp 以及 WallclockTimestampExtractor 但我对如何使用它们感到困惑。
请指导我如何才能做到这一点。