调试Faust流处理-从主题开头重启应用

Question

我正在调试一个简单的应用程序：

import faust

app = faust.App('app08')

# want to start from the beginning of the 
# topic every time the application restarts
@app.agent(topic) 
async def process(stream):
    async for event in stream:
        print(event)

并且希望在该应用程序重启时代理从最早的偏移量读取。现在，它是智能的，知道最后读取的消息的位置，并在重启时从该位置开始。尽管我翻阅了一段时间的文档，但我没有找到一个如何做到这一点的例子。我知道如何执行此操作的唯一方法是更改应用程序名称，例如:将app08更改为app09。

Answer 1

请记住，偏移量是由Kafka服务器使用与faust应用程序同名的消费者组控制的，我一直在使用kafaka-consumer-groups命令行界面( kafka安装的一部分)来执行此操作。

kafka-consumer-groups --bootstrap-server kafka_bootstrap --reset-offsets --to-earliest --group faust_appname --execute --all-topics

如果您运行的是相对较新的Kafka版本，也可以用--to-datetime替换--to-earliest，并以2020-09-20T00:00:00.00格式提供时间戳。

如果你想自动化这一点，我建议你也可以使用Python API来自动化对消费者组的控制。