apache/rocketmq
Watch
1
0
Fork
You've already forked rocketmq
 0
mirror of https://github.com/apache/rocketmq.git synced 2026-11-06 09:59:44 +00:00
Code Issues Projects Releases Packages Wiki Activity
Page: RIP 48 Enhance server side offset management ability
Pages
FAQ Home How to Configure TLS RIP 1 MQTT Bridge RIP 10 RocketMQ Unit Test RIP 11 Evolution of The Next Decade Architecture for RocketMQ RIP 12 Message Connector RIP 13 RocketMQ Console Project RIP 14 RocketMQ Community Operation Conventions RIP 15 RocketMQ IPv6 Support Project RIP 16 RocketMQ RPC(Request Response model) Support RIP 17 RocketMQ HTTP Proxy Support RIP 18 Metadata Management Architecture Upgrade RIP 2 RocketMQ Spring RIP 20 Streaming Tiered Storage Optimize RIP 21 logical queue abstraction for static topic and fast scale out RIP 26 Improve Batch Message Processing Throughput RIP 27 Auto batching in producer RIP 28 Light message queue (LMQ) RIP 29 Optimize RocketMQ NameServer RIP 3 RocketMQ Python Client RIP 30 Support Compaction topic RIP 31 Support RocketMQ BrokerContainer RIP 32 Slave Acting Master Mode RIP 33 RocketMQ MQTT RIP 34 Support quorum write and adaptive degradation in master slave architecture RIP 4 RocketMQ Go Client RIP 41 RocketMQ E2E Test RIP 42 Support Schema Registry RIP 44 Support DLedger Controller RIP 45 RocketMQ Replicator 2.0 RIP 45 Support Graceful Shutdown RIP 46 Observability Improvement for RocketMQ RIP 47 Data Layout V2 RIP 48 Enhance server side offset management ability RIP 49 RocketMQ remoting module improvement RIP 5 RocketMQ ACL RIP 50 RocketMQ Transaction Message Improvement RIP 51 Pop orderly improvement RIP 52 Optimize Building ConsumeQueue RIP 54 Backporting of Remoting based Client RIP 55 Support remoting protocol in rocketmq proxy module RIP 56 Replace Logging Module with Shaded Logback RIP 57 Tiered storage for RocketMQ RIP 6 Message Trace RIP 63 Queue selection strategy optimization RIP 65 Tiered Storage Optimization RIP 66 Support KV(Rocksdb) Storage RIP 7 Multiple Directories Storage Support RIP 8 Consumer RACK Support RIP 9 RocketMQ Developer Guide RIP‐67 jRaft‐Controller Implemention RIP‐80 Priority Message RIP‐82 Implement Timer message, transaction message, and index based on RocksDB RIP‐82 Implement Timer transaction message and index based on RocksDB RIP‐83 Lite Topic: A New Message Model Release and Lifetime Support Policy RocketMQ Improvement Proposal SoC [RIP 19] Server side rebalance, lightweight consumer client support [RIP 58] RocketMQ Streams improvement [RIP‐70] Optimizing Lock Mechanisms [RIP‐73] Pop Consumption Improvement Based on RocksDB [RIP‐74] Flink‐Connector‐RocketMQ Dynamic Load Balancing [RIP‐77] Deprecate and Remove ACL 1.0
1 RIP 48 Enhance server side offset management ability
lizhimins edited this page 2022-10-11 10:37:49 +08:00
Unescape Escape
Table of contents
This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

Status

Background & Motivation

What do we need to do

  • Will we add a new module? No.

  • Will we add new APIs? No.

  • Will we add a new feature? Yes.

Why should we do that

  • Are there any problems with our current project?

There are some problems in the current implementation of rocketmq offset management.

  1. Restarting the client in broadcast mode will consume message from max offset by default
  2. The accumulation amount is virtual-high when we uses "ConsumeProgress" command to observe the consumption progress.
  3. Reset offset by rpc is not reliable.
  • What can we benefit from proposed changes?

The implementation of enhance server-side offset management ability improved the above problems.

Goals

  • What problem is this proposal designed to solve?

RIP X Enhance server-side offset management ability

Non-Goals

  • What problem is this proposal NOT designed to solve?

Changes

Architecture

  1. Support server-side offset management in broadcast consumption mode

RocketMQ's broadcast mode means that messages are subscribed to all client which belong to a single consumer group, and is generally used to clear the cache or build the cache on the client for business logic. In the original design, if the local offset does not exist, the client cannot consume messages within the restart interval because the client starts to consume from the max(latest) offset after application restarting. We introduce a feature to avoid this situation by using the minimum consumption offset of all clients as the offset of this consumer group in broadcast mode.

  1. Introduces the concept of ready and inflight messages

In the old version of RocketMQ, if the user's consumption logic takes a long time, there will be a large number of messages being processed. The broker only saves the offset that have been successfully processed (clients returning success or failure), the accumulation amount is virtual-high when the user uses "ConsumeProgress" command to observe the consumption progress. Therefore, we define two new concepts, called ready and Inflight messages, to observe the consumption progress more accurately.

The ready state means that the message is ready on the server side, that is the ConsumeQueue is successfully constructed and can be pulled and consumed by the client. The inflight state refers to the state in which the message is acquired by the consumer client, and the consumption result has not been returned in the process of consumption. So how do we estimate the amount of messages being processed? In fact, it is the difference between the position of the latest news obtained by the pull request and the submitted position.

image-20220928172259995

  1. Support reset offset in server side to improve the success rate.

There are multiple links related to update offset in RocketMQ

  1. Pull Message（non-descending sequence）, org.apache.rocketmq.broker.processor.PullMessageProcessor#processRequest
  2. Consumer persist offset to remote（non-descending sequence） org.apache.rocketmq.broker.processor.ConsumerManageProcessor#updateConsumerOffset
  3. Use admin API to reset offset, org.apache.rocketmq.client.impl.ClientRemotingProcessor#resetOffset
  4. HA offset synchronize, org.apache.rocketmq.broker.slave.SlaveSynchronize#syncConsumerOffset

Due to complex concurrency issues in a distribution systen, these links may overlap each other. For example, link 1 2 and 3 may conflict, causing the reset offset API fail. We add a map on the server-side to temporarily store the request to reset the offset, and when the client pulls the message next time, the requested offset is overwritten. And This operation does not require the client to implement the reset offset interface, which reduces the complexity of the client.

image-20221008152546251

Interface Design/Change

The specific pull request is still in progress, but it will be seen by everyone soon.

Compatibility, Deprecation, and Migration Plan

  • re backward and forward compatibility taken into consideration? No
  • Are there deprecated APIs? No
  • How do we do migration? No

Implementation Outline

We will implement the proposed changes in three pull requests.

Rejected Alternatives

How do alternatives solve the issue you proposed?

Pros and Cons of alternatives

Why should we reject above alternatives

Copyright © 2016~2022 The Apache Software Foundation.