[RIP-79] Route Change Notification - #9549
Conversation
1. 创建路由事件专用系统Topic - 新增TopicValidator.RMQ_ROUTE_EVENT_TOPIC定义 - 加入系统主题集合(SYSTEM_TOPIC_SET) 2. 实现主备切换事件发送 - 在changeToMaster()中触发发送操作
1. broker端通知机制
- 当broker切换为slave时，构造ROUTE_EVENT(299)事件请求
- 通过remotingServer获取所有活跃连接(activeChannels)
- 向所有活跃客户端连接发送一次性通知
2. Client路由更新处理
- ClientRemotingProcessor处理ROUTE_EVENT请求
- 触发客户端立即更新路由信息(updateTopicRouteInfoFromNameServer)
3. RemotingServer新增
- 新增getActiveChannels()
- NettyRemotingServer实现活跃连接维护:
* 新增activeChannels集合
* channel激活时自动添加
* channel关闭时自动移除
1. 新增RouteEventSubscriber订阅路由事件 -创建轻量级消费者监听系统Topic 2.实现缓存失效标记机制 -标记相关Topic缓存为dirtyTopics -dirtyTopics加入pendingTopics队列 3.路由异步更新机制： -200ms窗口处理pendingTopics -使用现有cacheRefreshExecutor线程池并行更新 -失败重试 -保留原有定时刷新作为兜底策略
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## feature-reverse-route-update #9549 +/- ##
==================================================================
- Coverage 48.41% 48.28% -0.13%
+ Complexity 12255 12242 -13
==================================================================
Files 1314 1318 +4
Lines 93668 93875 +207
Branches 12011 12040 +29
==================================================================
- Hits 45347 45331 -16
- Misses 42733 42917 +184
- Partials 5588 5627 +39 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
|
Hi @saz97 , is this all submitted? I understand that when the master switches, the client will be notified to re-acquire the route. Will the client's method of obtaining the route regularly be retained? I think the server should generally rely on the client as little as possible. |
The PR hasn't been fully submitted yet—the current commits are just a demo. |
If the scheduled pull is not retained, this method strongly relies on the client, and the client and server must be upgraded together. In fact, in actual production applications, many client and server versions are inconsistent, because the cost of upgrading the client is very high. Is compatibility considered in this way? |
Thank you for your valuable insights and the practical perspective from production environments! |
1. Broker事件发布机制:
- 在BrokerController添加`publishRouteEvent`方法
- Broker启动/关闭/时发送事件到`RMQ_ROUTE_EVENT_TOPIC`
2. Proxy事件订阅机制:
- 创建RouteEventSubscriber消费路由事件
- 实现事件处理流水线：
* 订阅系统Topic `RMQ_ROUTE_EVENT_TOPIC`
* 提取受影响Topic列表
* 调用markCacheDirty标记脏数据
- 向namesrv获取新路由
测试了单副本Broker的IP变更场景，Proxy端能及时感知，并更新路由。
Broker 层事件发布重构 1. 移除原有 publishRouteEvent()方法及其硬编码逻辑。 -引入新的 RouteEventService类，专门负责将 Broker 的生命周期事件（暂时是启动和关闭）发布到系统Topic (RMQ_ROUTE_EVENT_TOPIC)。 -用RouteEventType枚举替代原有的。 2.修改 BrokerController： -在启动 (start()) 和关闭 (shutdown()) 流程中调用 RouteEventService发布相应事件。 Proxy 层事件消费与缓存刷新重构 1.移除 ClusterTopicRouteService中直接启动和关闭 RouteEventSubscriber的逻辑（该职责移至基类）。 2.重构 TopicRouteService： -引入 RouteCacheRefresher组件，专门负责管理代理层的路由缓存失效与刷新策略。 -维护broker-topic的映射，用于确定affectedTopic 3.将路由功能收敛到工具类 -RouteEventSubscriber仅负责订阅 Broker 发布的事件，提取 affectedTopics并通知 RouteCacheRefresher将这些 Topic 标记为“脏数据”。 -RouteCacheRefresher负责维护脏数据队列、执行异步批量刷新任务（使用专用线程池 cacheRefreshExecutor）、处理刷新超时（超时后重回队列重试）以及更新缓存状态。 -删除原有的 batchRefreshRoutes, refreshSingleRoute以及基于 dirtyTopics/pendingTopics/lazyUpdateExecutor的刷新逻辑，全部由 RouteCacheRefresher接管。 4. 简化 ClusterTopicRouteService实现，仅保留与集群路由获取相关的基类逻辑。 5.移除冗余功能 -删除 ReplicasManager中试图直接通知客户端连接关于主从切换的逻辑 (buildNotify, activeChannels通知等)，其功能已被事件机制替代。 -删除client修改，将修改集中于Proxy修改
1.在Broker端，当Topic配置发生变化： -通过RouteEventService发布TOPIC_CHANGE事件 -事件携带受影响的主题名称（AFFECTED_TOPIC） 2.Proxy端重构逻辑： -增强维护brokerToTopics映射关系 3.增强时效性检查： -消息携带时间戳字段 -暂定延迟超过1分钟的事件将被丢弃，避免处理陈旧事件造成的无效刷新 4.完善路由刷新机制： -刷新前检查topicCache是否存在对应主题 -无缓存主题时跳过刷新操作
bbd9807 to
7efc56e
Compare
1. 移除错误的checkRefreshStatus - 原 checkRefreshStatus 方法通过 topicCache.getIfPresent 检测刷新完成不可靠 - 该方法错误地将缓存存在视为刷新完成，导致状态误判 2. 重构状态确认 - 新增 markCompleted/markRetry 方法在 reload 回调中精确更新状态 - 在 processDirtyTopics 中增加超时检测，避免无限重试 3. 优化错误处理流程 - 刷新失败时通过 markRetry 重新加入队列 - 超时topic直接标记完成并报出错误日志，避免阻塞
7efc56e to
8139467
Compare
本次提交在Broker和Proxy模块为新增的路由事件通知功能引入了可配置的开关。 1.Broker端修改： - 在 BrokerConfig 中新增 routeEventServiceEnable，用于控制RouteEventService 的初始化 - 修改 BrokerController，使其根据配置值有条件地初始化RouteEventService - 在流程中增加空指针检查 2.Proxy端修改： - 在ProxyConfig中新增 brokerRouteEventServiceEnable - 修改 TopicRouteService，使其根据配置初始化 RouteEventSubscriber 和 RouteCacheRefresher 测试新增： - 新增 RouteEventServiceTest 用于测试Broker端的事件发布 - 新增 RouteEventSubscriberTest 用于测试Proxy端的事件处理 - 新增 RouteCacheRefresherTest 用于测试Proxy端的缓存刷新机制
1. 修改Broker端触发逻辑 - 移除putTopicConfig和removeTopicConfig方法的事件触发 - 在Topic创建/删除的方法中触发 2. 修改开关命名 - 将routeEventServiceEnable更名为enableRouteChangeNotification 3. 调整开关逻辑 - 事件发布时检查enableRouteChangeNotification开关 4. 删除Proxy端重试逻辑 - 删除RouteCacheRefresher中的markRetry方法 - 失败时不再重试，依赖基础定时更新
fc9a83f to
2d2c1fb
Compare
1.broker侧 - 在BrokerController的start中调用publishevent，发送START事件 2.proxy侧 - 增加对START事件的处理，逻辑和TOPIC_CHANGE一致，即更新受影响的topic路由
758c181 to
13bc8ff
Compare
1. 收敛重复代码： - 提取公共事件发布逻辑到统一方法 - 消除TopicConfigManager中各操作点的重复事件发布代码 2. 实现事件分块发送： - 当topic数量超过阈值时自动分块发送 - 新增partitionTopics方法处理主题分批 - 支持MAX_TOPICS_PER_EVENT配置控制批量大小
- 将BrokerController中的START事件发布时机移至Broker registerBrokerAll后执行 - 在发送START和SHUTDOWN事件前，增加enableRouteChangeNotification配置检查 - 将RouteEventService中MAX_TOPICS_PER_EVENT批次大小从100调整为5000 - 在TopicConfigManager中新增带publishEvent参数的deleteTopicConfig重载方法 - 修改SlaveSynchronize在同步操作时调用publishEvent=false的deleteTopicConfig方法 - 将TOPIC_CHANGE逻辑统一收敛到registerBrokerData方法中 - 在RouteEventSubscriber构造函数中设置消费位点为CONSUME_FROM_LAST_OFFSET
1. 将RouteEventSubscriber和RouteCacheRefresher合并为RouteChangeNotifier 2. 简化BrokerController中事件发布代码，改为直接调用routeEventService封装方法 3. 在RouteEventService中添加publishEvent重载方法 4. 更新相关测试代码适应新的类结构
- 移除向本地store写入的逻辑 - 基于RMQ_ROUTE_EVENT_TOPIC的路由信息，向其他相关的远端Broker发送实现消息 - 添加Broker名称去重处理，避免重复发送 - 使用putMessageToRemoteBroker进行跨Broker消息传输
b4cff03
into
apache:feature-reverse-route-update
4 participants
The route update mechanism in RocketMQ currently relies on a periodic polling approach by Clients and Proxy. This passive update mechanism causes some issues during route events (e.g., master/slave switch, Broker shutdown):
Fundamentally, operations during polling intervals rely on incorrect routing decisions, compromising service high availability.
To address this, I would like to start an email thread to discuss the RIP-79 Route Change Notification, reconstructing the route synchronization mechanism to achieve event-driven route state synchronization.
The redesigned architecture can resolves untimely route updates: Clients no longer connect to invalid nodes, eliminating service interruptions during master/slave switches, and new resources can be immediately discovered and leveraged. Additionally, redundant query pressure on NameServer is reduced through merged update requests and a lazy update policy.
Relevant work is already in progress. Proposal documentation is available here:
https://docs.google.com/document/d/1iMQQ0wDO4dULGpjsKyq6kCbcsUL0JjNcii-KbM90a18/edit?usp=sharing