如何更改主题的起始偏移量？答案

【问题标题】：How to change start offset for topic?如何更改主题的起始偏移量？
【发布时间】：2015-06-29 17:44:35
【问题描述】：

是否可以更改新主题的起始偏移量？我想创建一个新主题并从偏移量10000 开始阅读。怎么样？

【问题讨论】：

【解决方案1】：

如果您需要更改偏移量。

kafka-consumer-groups --bootstrap-server {url} \
--topic {topic} \
--group {consumer} \
--reset-offsets --to-datetime 2020-11-11T00:00:00.000+0900 \
--execute

Unparseable Date Error when parsing UTC string through SimpleDateFormat to Date

【讨论】：

【解决方案2】：

由于kafka 0.11.0.0你可以使用脚本kafka-consumer-groups.sh 来自this answer的示例

kafka-consumer-groups.sh --bootstrap-server kafka-host:9092 --group my-group --reset-offsets --to-earliest --all-topics --execute

KIP-122: Add Reset Consumer Group Offsets tooling中列出的其他选项

.----------------------.-----------------------------------------------.----------------------------------------------------------------------------------------------------------------------------------------------.
|      Scenario        |                   Arguments                   |                                                                    Example                                                                   |
:----------------------+-----------------------------------------------+----------------------------------------------------------------------------------------------------------------------------------------------:
| Reset to Datetime    |  --to-datetime YYYY-MM-DDTHH:mm:SS.sss±hh:mm  |  Reset to first offset since 01 January 2017, 00:00:00 hrs: --reset-offsets –group test.group --topic foo --to-datetime 2017-01-01T00:00:00Z |
:----------------------+-----------------------------------------------+----------------------------------------------------------------------------------------------------------------------------------------------:
| Reset by Duration    |  --by-duration  PnDTnHnMnS                    |  Reset to first offset since one week ago (from current timestamp): --reset-offsets --group test.group --topic foo --by-duration P7D         |
:----------------------+-----------------------------------------------+----------------------------------------------------------------------------------------------------------------------------------------------:
| Reset to Earliest    |  --to-earliest                                |  Reset to earliest offset available: --reset-offsets --group test.group --topic foo --to-earliest                                            |
:----------------------+-----------------------------------------------+----------------------------------------------------------------------------------------------------------------------------------------------:
| Reset to Latest      |  --to-latest                                  |  Reset to latest offset available: --reset-offsets --group test.group --topic foo --to-latest                                                |
:----------------------+-----------------------------------------------+----------------------------------------------------------------------------------------------------------------------------------------------:
| Reset to Offset      |  --to-offset                                  |  Reset to offset 1 in all partitions: --reset-offsets --group test.group --topic foo --to-offset 1                                           |
:----------------------+-----------------------------------------------+----------------------------------------------------------------------------------------------------------------------------------------------:
| Shift Offset by 'n'  |  --shift-by n                                 |  Reset to current offset plus 5 positions: --reset-offsets --group test.group –topic foo --shift-by 5                                        |
:----------------------+-----------------------------------------------+----------------------------------------------------------------------------------------------------------------------------------------------:
| Reset from File      |  --from-file PATH_TO_FILE                     |  Reset using a file with reset plan: --reset-offsets --group test.group --from-file reset-plan.csv                                           |
'----------------------'-----------------------------------------------'----------------------------------------------------------------------------------------------------------------------------------------------'

你也可以定义你要重置的分区，例如：

将主题 foo 分区 0 的偏移量重置为 1

--reset-offsets --group test.group --topic foo:0 --to-offset 1
将主题 foo 分区 0,1,2 的偏移量重置为最早

--reset-offsets --group test.group --topic foo:0,1,2 --to-earliest

提醒：不要忘记--execute 标志（请参阅 KIP 中的执行选项）。如果没有这个标志，脚本只会按范围打印出场景的结果，例如：

TOPIC                 PARTITION NEW-OFFSET NEW-LAG LOG-END-OFFSET CONSUMER-ID HOST CLIENT-ID
foo                   0         90         10      100            -           -    -

感谢this answer。使用ascii tables创建的表

【讨论】：

没有多少人熟悉这个伟大的工具。非常感谢你拯救了我的一天。
topic:partition 节省了我的时间。谢谢。
需要注意的是，您的消费者组必须被视为非活动状态才能正常工作，否则不会应用任何更改。

【解决方案3】：

由于 kafka 0.9 偏移量存储在主题中。要更改偏移量，请使用seek() method：

public void seek(TopicPartition partition, long offset)

覆盖消费者将在下一个poll(timeout) 上使用的获取偏移量。如果多次为同一个分区调用此 API，则在下一次 poll() 中将使用最新的偏移量。请注意，如果在消费过程中任意使用此 API 来重置获取偏移量，您可能会丢失数据

【讨论】：

【解决方案4】：

您可以在 zookeeper shell 的帮助下做到这一点。 Kafka 使用 zookeeper 来跟踪消费者的偏移量。

进入kafka bin目录并调用zookeeper shell。（我的kafka版本是0.8.0）

./zookeeper-shell.sh localhost:2181

现在使用 zookeeper get 命令

get /consumers/consumer_group_id/offsets/topic/0

它显示类似的东西

2043
cZxid = 0x4d
ctime = Wed Mar 18 03:56:32 EDT 2015
...

这里 2043 是消耗的最大偏移量。使用 zookeeper set 命令将其设置为所需的值

set /consumers/consumer_group_id/offsets/topic/0 10000

路径的框架如下 /consumers/[consumer_group_id]/offsets/[topic]/[partition_id]。
您必须用适当的消费者组、主题和分区 ID 替换。

*另外，既然你提到它是一个新的 kafka 实例，我不确定消费者是否会连接并创建消费者组。

【讨论】：