flink1.11 中 使用SQL 获取 Kafka Connector 时 如何获取或提取到 kafka消息落盘kafka时间

在 Flink 1.11 中，可以使用 Flink 的 Table API 和 SQL API 来从 Kafka 中读取数据，并将其转换为 Table。假设你已经成功地将 Kafka 的数据转换为 Table，那么如何获取消息的落盘时间呢？

Flink 的 Table API 和 SQL API 为每个输入数据源都定义了一个元组类型，其中包含了所有字段的名称和类型。对于 Kafka 的元组类型，通常会包含以下字段：

- `msg`: Kafka 消息的内容。
- `offset`: Kafka 消息的偏移量。
- `partition`: Kafka 消息所在的分区。
- `timestamp`: Kafka 消息的时间戳。

要获取消息的落盘时间，可以使用 Flink SQL 中的 `PROCTIME()` 函数。`PROCTIME()` 函数返回当前处理时间，因此可以将其与 `timestamp` 字段进行比较，以确定消息是否已经落盘。

以下是一个使用 Flink SQL 从 Kafka 中读取数据并获取消息落盘时间的示例：

CREATE TABLE kafkaTable (
  msg STRING,
  offset BIGINT,
  partition INT,
  timestamp TIMESTAMP(3)
) WITH (
  'connector' = 'kafka',
  'topic' = 'myTopic',
  'properties.bootstrap.servers' = 'localhost:9092',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json'
);

SELECT
  msg,
  offset,
  partition,
  timestamp,
  PROCTIME() AS processingTime
FROM kafkaTable;

在上面的示例中，我们定义了一个名为 `kafkaTable` 的表，该表从名为 `myTopic` 的 Kafka 主题中读取数据。在 `SELECT` 子句中，我们选择了所有字段，并使用 `PROCTIME()` 函数添加了一个名为 `processingTime` 的新字段。这个新字段包含了当前处理时间，可以与 `timestamp` 字段进行比较。