我知道使用Kafka使用者的api我们可以获得与特定时间戳(getOffsetsByTimes())相对应的偏移量。
我们如何获得偏移并从基于KSQL或KSTream中的时间戳的点开始重放流?
KSQL尚不支持此功能。您有一个选择是使用谓词来过滤掉您不想要的记录:
SELECT * FROM SOURCE_STREAM WHERE ROWTIME > xxxx
但这可能效率低,具体取决于偏移量相对于日志中数据量的位置。