使用KafkaSource的projectionFn时，如何通过timestamp.extractor获取消息时间戳？

你提到用projectionFn只能访问key和value，其实在新版Flink的KafkaSource API里，projectionFn是可以直接拿到完整的Kafka ConsumerRecord对象的，里面就包含了消息的时间戳信息，不需要额外依赖timestamp.extractor来间接获取。下面分两种场景给你具体说明：

一、使用新版Flink KafkaSource API（Flink 1.13+） #

新版的KafkaSource采用了Flink统一的Source API，setProjection方法接收的ProjectionFunction参数，输入就是完整的ConsumerRecord<K, V>。你可以直接从这个对象里提取时间戳，同时保留key和value的解析逻辑。

举个Java代码示例：

// 定义一个自定义POJO来封装消息的key、value和时间戳
public class KafkaMessage {
    private String key;
    private YourParsedValue value; // 这里是你解析JSON后的对象
    private long timestamp;

    // 构造方法、getter/setter省略
}

// 构建KafkaSource时配置projectionFn
KafkaSource<String, String> kafkaSource = KafkaSource.<String, String>builder()
        .setBootstrapServers("your-broker-address")
        .setTopics("target-topic")
        .setGroupId("consumer-group-id")
        // 先指定反序列化key和value为字符串（后续在projection里解析JSON）
        .setDeserializer(KafkaRecordDeserializationSchema.valueOnly(StringDeserializer.class))
        .setProjection(consumerRecord -> {
            // 直接获取Kafka消息的时间戳
            long msgTimestamp = consumerRecord.timestamp();
            // 获取key和原始value字符串
            String msgKey = consumerRecord.key();
            String rawValue = consumerRecord.value();
            
            // 这里执行你的JSON解析逻辑，把rawValue转成自定义对象
            YourParsedValue parsedValue = parseJsonToObject(rawValue);
            
            // 返回封装了所有信息的自定义对象
            return new KafkaMessage(msgKey, parsedValue, msgTimestamp);
        })
        .build();

这里要注意：consumerRecord.timestamp()返回的时间戳，取决于Kafka消息的时间戳类型（是生产者设置的CreateTime，还是Broker写入的LogAppendTime），你可以通过consumerRecord.timestampType()来查看具体类型。

二、使用旧版FlinkKafkaConsumer API（Flink 1.12及以前） #

如果还在使用旧的消费者API，你需要先自定义KafkaDeserializationSchema来获取完整的ConsumerRecord，再通过TimestampAssigner提取时间戳：

1. 自定义反序列化Schema，返回完整的ConsumerRecord：

public class FullKafkaRecordSchema implements KafkaDeserializationSchema<ConsumerRecord<String, String>> {
    @Override
    public boolean isEndOfStream(ConsumerRecord<String, String> nextElement) {
        return false;
    }

    @Override
    public ConsumerRecord<String, String> deserialize(ConsumerRecord<byte[], byte[]> record) throws IOException {
        // 将byte数组转成字符串类型的key和value
        String key = record.key() != null ? new String(record.key()) : null;
        String value = record.value() != null ? new String(record.value()) : null;
        // 返回完整的ConsumerRecord对象
        return new ConsumerRecord<>(
                record.topic(), record.partition(), record.offset(),
                record.timestamp(), record.timestampType(),
                record.checksum(), record.serializedKeySize(), record.serializedValueSize(),
                key, value
        );
    }
}

2. 消费消息并提取时间戳：

Properties kafkaProps = new Properties();
kafkaProps.setProperty("bootstrap.servers", "your-broker-address");
kafkaProps.setProperty("group.id", "consumer-group-id");

FlinkKafkaConsumer<ConsumerRecord<String, String>> consumer = 
        new FlinkKafkaConsumer<>("target-topic", new FullKafkaRecordSchema(), kafkaProps);

// 提取时间戳并生成水印
DataStream<KafkaMessage> messageStream = env.addSource(consumer)
        .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<ConsumerRecord<String, String>>() {
            @Override
            public long extractTimestamp(ConsumerRecord<String, String> record, long previousTimestamp) {
                // 直接从ConsumerRecord获取时间戳
                return record.timestamp();
            }

            @Override
            public Watermark getCurrentWatermark() {
                // 根据业务需求生成水印，比如延迟5秒
                return new Watermark(System.currentTimeMillis() - 5000);
            }
        })
        .map(record -> {
            // 在这里解析JSON value并封装成自定义对象
            YourParsedValue parsedValue = parseJsonToObject(record.value());
            return new KafkaMessage(record.key(), parsedValue, record.timestamp());
        });

总结一下，新版API更简洁，直接在projectionFn里就能拿到时间戳；旧版API需要先拿到完整的ConsumerRecord，再通过时间戳分配器提取。根据你的Flink版本选择对应的方式就好啦～

内容的提问来源于stack exchange，提问作者Kleyson Rios