如何从起始位置监听Kafka流？（已通过mgconsole创建流）

你已通过mgconsole创建了如下Kafka流：

CREATE KAFKA STREAM music_stream
TOPICS music-stars
TRANSFORM music_books.music_stars
CONSUMER_GROUP randomseed141234
BOOTSTRAP_SERVERS 'localhost:9093'
CREDENTIALS {'sasl.username':'public',
                'sasl.password':'public',
                'security.protocol':'SASL_PLAINTEXT',
                'sasl.mechanism':'PLAIN'};

要监听该流的所有历史消息，核心是让消费者从Kafka主题的最起始位置开始消费，以下是两种可行方案：

方案一：重新创建流时指定起始偏移量 #

在创建流的SQL语句中添加START_OFFSET 'earliest'参数，强制消费者从主题的第一条消息开始读取。修改后的完整语句如下：

CREATE KAFKA STREAM music_stream
TOPICS music-stars
TRANSFORM music_books.music_stars
CONSUMER_GROUP randomseed141234
BOOTSTRAP_SERVERS 'localhost:9093'
START_OFFSET 'earliest'
CREDENTIALS {'sasl.username':'public',
                'sasl.password':'public',
                'security.protocol':'SASL_PLAINTEXT',
                'sasl.mechanism':'PLAIN'};

注意：如果music_stream流已存在，需先执行DROP KAFKA STREAM music_stream;删除原有流，再执行上述创建语句。

方案二：重置现有消费者组的偏移量 #

若不想重新创建流，可通过Kafka官方工具重置消费者组的偏移量到最早位置，步骤如下：

1. 先查看目标消费者组的当前偏移量状态（可选）：

kafka-consumer-groups.sh --bootstrap-server localhost:9093 --group randomseed141234 --describe

2. 执行偏移量重置命令：

kafka-consumer-groups.sh --bootstrap-server localhost:9093 --group randomseed141234 --topic music-stars --reset-offsets --to-earliest --execute

执行完成后，重启流的消费进程（或重新通过mgconsole激活该流），即可开始接收所有历史消息。

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