KSQL流Retention Policy未生效？查询返回超10分钟前数据

你遇到的核心问题是对RETENTION_MS参数的作用理解有误，同时忽略了KSQL持续查询的消费逻辑：

1. RETENTION_MS不是过滤写入的数据 #

RETENTION_MS是Kafka主题的属性，它只控制消息在主题中被保留的时长——超过这个时间的消息会被Kafka自动删除，但不会阻止旧数据被写入主题。你创建的持续查询（CREATE STREAM AS SELECT）会消费原流points中的数据（包括历史旧数据），并全部写入points-recent主题，这些旧数据会留在主题里直到达到保留期才被删除，所以你查询到的是10分钟前的历史数据。

2. 正确实现“只保留最近N秒数据”的两种方式 #

方式一：直接在查询时过滤时间（推荐） #

不需要创建新流，直接查询原流时添加时间过滤条件，只获取最近N秒的数据：

SELECT TIMESTAMPTOSTRING(rowtime, 'yyyy-MM-dd HH:mm:ss.SSS'), TIMESTAMPTOSTRING(unix_timestamp(), 'yyyy-MM-dd HH:mm:ss.SSS') 
FROM points 
WHERE rowtime >= CURRENT_TIMESTAMP() - INTERVAL 5 MINUTES  -- 替换成你需要的时长，比如INTERVAL 30 SECONDS
LIMIT 1;

方式二：创建仅包含新数据的流 #

如果一定要创建专门的短保留期流，需要同时做两件事：

• 在持续查询中添加时间过滤，只写入符合时间条件的新数据
• 指定查询从原流的最新位置开始消费，避免把历史数据写入新主题

示例代码：

-- 先删除已有的points-recent主题（如果里面已经有旧数据）
DROP STREAM points_recent IF EXISTS;
-- 重新创建流，添加时间过滤并指定从最新位置消费
CREATE STREAM points_recent WITH (RETENTION_MS=300000, KAFKA_TOPIC='points-recent') 
AS SELECT * 
FROM points 
WHERE rowtime >= CURRENT_TIMESTAMP() - INTERVAL 5 MINUTES
EMIT CHANGES
WITH (STARTING OFFSET='LATEST');

这样，新流只会接收原流中查询启动后产生的、且在最近5分钟内的数据，旧历史数据不会被写入，结合RETENTION_MS的设置，主题里只会保留最近5分钟的有效数据。

内容的提问来源于stack exchange，提问作者Dennis Jansky