flink设置从某个时间点开始度kafka

可以使用 Flink 的 `assignTimestampsAndWatermarks` 方法来设置从某个时间点开始消费 Kafka。

具体来说，可以使用 `KafkaConsumer` 的 `assign` 方法手动分配 Kafka 分区，在分配分区后，可以通过设置 `KafkaConsumer` 的 `timestamp` 参数来指定从哪个时间点开始消费消息。

以下是一个示例代码，展示如何从 2021 年 7 月 1 日开始消费 Kafka 消息：

Properties props = new Properties();
props.setProperty("bootstrap.servers", "localhost:9092");
props.setProperty("group.id", "test-group");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

TopicPartition partition = new TopicPartition("test", 0);
consumer.assign(Collections.singletonList(partition));

long timestamp = LocalDateTime.of(2021, 7, 1, 0, 0, 0)
        .atZone(ZoneId.systemDefault())
        .toInstant()
        .toEpochMilli();

consumer.seek(partition, timestamp);

DataStream<String> stream = env.addSource(new FlinkKafkaConsumer<>("test", new SimpleStringSchema(), properties))
        .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<String>() {
            @Override
            public long extractAscendingTimestamp(String element) {
                // 设置事件时间
                return System.currentTimeMillis();
            }
        });

在上面的代码中，通过 `KafkaConsumer` 的 `assign` 方法将分区 `0` 手动分配给消费者，并且通过 `seek` 方法设置从 2021 年 7 月 1 日开始消费消息。最后使用 `assignTimestampsAndWatermarks` 方法设置事件时间。