NingG +

Apache Kafka 实践:重置 offset

1.概要

2.重置 Offset

几个方面:

  1. 情况收集:
    1. 哪个 Consumer Group
    2. 消费哪些 Topic
    3. 当前 Topic 和 Consumer Group 对应的 offset 都是多少
  2. Offset 处理:
    1. 存储位置:consumer group 对应 offset 的存储位置
    2. 设置 Offset:如何设置 Offset?

直接查看当前的 offset:

./kafka-consumer-groups.sh --bootstrap-server kafka1.hdp.ningg.cn:9092 --describe --group pull-for-abacus-prod
TOPIC                          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG        CONSUMER-ID                                       HOST                           CLIENT-ID
binlog.test_pay.test_refund      6          1472114         1472114         0          consumer-1-78074b08-5232-42d7-9a7a-68349f856095   /10.1.187.47                   consumer-1
binlog.test_pay.test_refund      7          1474380         1474380         0          consumer-1-78074b08-5232-42d7-9a7a-68349f856095   /10.1.187.47                   consumer-1
binlog.test_pay.test_refund      0          1470473         1470476         3          consumer-1-1b0855c5-404f-4eec-8c40-dfebc435faba   /10.30.173.144                 consumer-1
binlog.test_pay.test_refund      1          1474526         1474529         3          consumer-1-1b0855c5-404f-4eec-8c40-dfebc435faba   /10.30.173.144                 consumer-1
binlog.test_pay.test_refund      4          1472482         1472482         0          consumer-1-5d8f9bdc-af6f-4f6a-9f54-2a371c4cbc8b   /10.30.125.136                 consumer-1
binlog.test_pay.test_refund      5          1472702         1472707         5          consumer-1-5d8f9bdc-af6f-4f6a-9f54-2a371c4cbc8b   /10.30.125.136                 consumer-1
binlog.test_pay.test_refund      2          1473468         1473468         0          consumer-1-3875d283-6a7c-4691-8611-ca97b382c66a   /10.1.186.91                   consumer-1
binlog.test_pay.test_refund      3          1469992         1469995         3          consumer-1-3875d283-6a7c-4691-8611-ca97b382c66a   /10.1.186.91                   consumer-1
binlog.test_pay.test_refund      9          1473234         1473238         4          consumer-1-d5a49669-0c88-4b7b-b8a3-b68e379f10b6   /10.30.156.140                 consumer-1
binlog.test_pay.test_refund      8          1475842         1475842         0          consumer-1-92f73b05-b5c8-4ef3-b0c0-68243bdd4469   /10.30.149.137                 consumer-1

从上述查询可知,offset 没有托管在 ZK 上,而是托管在了 __consumer_offsets 中,因此,需要修改 __consumer_offsets 中的 offset

Note:

在 Kafka 0.11+ 之后的版本中,提供了重置 offset 的工具:(确认:此工具也可以操作 Kafka 0.10 集群)

// 查看 consumer group 的 offset 状态
./kafka-consumer-groups.sh --bootstrap-server kafka1.hdp.ningg.cn:9092 --describe --group pull-for-abacus-prod
 
// 尝试重置  consumer group
./kafka-consumer-groups.sh --bootstrap-server kafka1.hdp.ningg.cn:9092 --group pull-for-abacus-prod --reset-offsets --to-earliest --all-topics
 
// 确认重置  consumer group
./kafka-consumer-groups.sh --bootstrap-server kafka1.hdp.ningg.cn:9092 --group pull-for-abacus-prod --reset-offsets --to-earliest --all-topics --execute

Note:

完整的 kafka-consumer-groups 工具的操作选项:

$ ./kafka-consumer-groups.sh
List all consumer groups, describe a consumer group, delete consumer group info, or reset consumer group offsets.
Option                                  Description
------                                  -----------
--all-topics                            Consider all topics assigned to a
                                          group in the `reset-offsets` process.
--bootstrap-server <String: server to   REQUIRED (for consumer groups based on
  connect to>                             the new consumer): The server to
                                          connect to.
--by-duration <String: duration>        Reset offsets to offset by duration
                                          from current timestamp. Format:
                                          'PnDTnHnMnS'
--command-config <String: command       Property file containing configs to be
  config property file>                   passed to Admin Client and Consumer.
--delete                                Pass in groups to delete topic
                                          partition offsets and ownership
                                          information over the entire consumer
                                          group. For instance --group g1 --
                                          group g2
                                        Pass in groups with a single topic to
                                          just delete the given topic's
                                          partition offsets and ownership
                                          information for the given consumer
                                          groups. For instance --group g1 --
                                          group g2 --topic t1
                                        Pass in just a topic to delete the
                                          given topic's partition offsets and
                                          ownership information for every
                                          consumer group. For instance --topic
                                          t1
                                        WARNING: Group deletion only works for
                                          old ZK-based consumer groups, and
                                          one has to use it carefully to only
                                          delete groups that are not active.
--describe                              Describe consumer group and list
                                          offset lag (number of messages not
                                          yet processed) related to given
                                          group.
--execute                               Execute operation. Supported
                                          operations: reset-offsets.
--export                                Export operation execution to a CSV
                                          file. Supported operations: reset-
                                          offsets.
--from-file <String: path to CSV file>  Reset offsets to values defined in CSV
                                          file.
--group <String: consumer group>        The consumer group we wish to act on.
--list                                  List all consumer groups.
--new-consumer                          Use the new consumer implementation.
                                          This is the default, so this option
                                          is deprecated and will be removed in
                                          a future release.
--reset-offsets                         Reset offsets of consumer group.
                                          Supports one consumer group at the
                                          time, and instances should be
                                          inactive
                                        Has 3 execution options: (default) to
                                          plan which offsets to reset, --
                                          execute to execute the reset-offsets
                                          process, and --export to export the
                                          results to a CSV format.
                                        Has the following scenarios to choose:
                                          --to-datetime, --by-period, --to-
                                          earliest, --to-latest, --shift-by, --
                                          from-file, --to-current. One
                                          scenario must be choose
                                        To define the scope use: --all-topics
                                          or --topic. . One scope must be
                                          choose, unless you use '--from-file'
                                          scenario
--shift-by <Long: number-of-offsets>    Reset offsets shifting current offset
                                          by 'n', where 'n' can be positive or
                                          negative
--timeout <Long: timeout (ms)>          The timeout that can be set for some
                                          use cases. For example, it can be
                                          used when describing the group to
                                          specify the maximum amount of time
                                          in milliseconds to wait before the
                                          group stabilizes (when the group is
                                          just created, or is going through
                                          some changes). (default: 5000)
--to-current                            Reset offsets to current offset.
--to-datetime <String: datetime>        Reset offsets to offset from datetime.
                                          Format: 'YYYY-MM-DDTHH:mm:SS.sss'
--to-earliest                           Reset offsets to earliest offset.
--to-latest                             Reset offsets to latest offset.
--to-offset <Long: offset>              Reset offsets to a specific offset.
--topic <String: topic>                 The topic whose consumer group
                                          information should be deleted or
                                          topic whose should be included in
                                          the reset offset process. In `reset-
                                          offsets` case, partitions can be
                                          specified using this format: `topic1:
                                          0,1,2`, where 0,1,2 are the
                                          partition to be included in the
                                          process. Reset-offsets also supports
                                          multiple topic inputs.
--zookeeper <String: urls>              REQUIRED (for consumer groups based on
                                          the old consumer): The connection
                                          string for the zookeeper connection
                                          in the form host:port. Multiple URLS
                                          can be given to allow fail-over.

使用 Kafka 0.11+ 的工具,也可以操作 Kafka 0.10:

3.参考资料

同类文章:

微信搜索: 公众号 ningg ,即可联系我

Top