如何提取Kafka Streams中消息中嵌入的时间戳

我想提取嵌入每条消息的时间戳，并将它们作为json有效负载发送到我的数据库中 .

我想获得以下三个时间戳 .

Event-time: The point in time when an event or data record occurred, i.e. was originally created “by the source”.

Processing-time: 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.

Ingestion-time: The point in time when an event or data record is stored in a topic partition by a Kafka broker.

这是我的流应用程序代码：

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 ，但我对如何使用它感到困惑 .

请指导我如何实现这一目标 .