You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by "zhaohai666 (via GitHub)" <gi...@apache.org> on 2026/07/31 02:42:41 UTC
[PR] Feature/rip 2 pr4 route change event [rocketmq]
zhaohai666 opened a new pull request, #10705:
URL: https://github.com/apache/rocketmq/pull/10705
## [Studio] feat: Route Change Event Detection and Notification System
**Issue:** #10635
**Branch:** `feature/rip-2-pr4-route-change-event`
---
### Overview
This PR implements the route change event detection and real-time notification system for the RIP-2 Proxy Admin Interface. It monitors topic route data changes (broker online/offline, queue scaling, topic creation/deletion) and pushes events to subscribed gRPC clients via server-streaming RPCs.
### Key Changes
**RouteChangeEventDetector** -- Core detection engine (357 lines) that compares old vs. new `TopicRouteData` to produce a list of `RouteChangeEvent` objects.
- `detectChanges(topic, oldRoute, newRoute)`: Main entry point. Delegates to specialized detection methods and returns a consolidated event list.
- `detectBrokerChanges()`: Compares `brokerDatas` maps (brokerName -> brokerId -> address). Emits `BROKER_ONLINE` / `BROKER_OFFLINE` for new/removed broker names, new/removed broker instances (brokerId), and address changes (treated as offline + online pair).
- `detectQueueChanges()`: Compares `queueDatas` per brokerName. Emits `QUEUE_SCALE` when read or write queue counts differ.
- `detectBrokerOnlineEvents()` / `detectBrokerOfflineEvents()`: Bulk-generate broker events for topic create/delete scenarios.
- `buildSnapshot()`: Builds a `TopicRouteSnapshot` for initial state synchronization.
**RouteChangeNotifier** -- Manages gRPC subscriber streams and broadcasts detected events (245 lines). Implements `TopicRouteService.RouteRefreshListener`.
- `subscribe(topics, eventTypes, StreamObserver)`: Registers a subscriber with optional topic and event-type filters. Wires up `setOnCancelHandler` for client disconnect cleanup. Sends an initial `ROUTE_SNAPSHOT` for all cached topics upon subscription.
- `onRouteRefreshed(topic, oldView, newView)`: Callback from `TopicRouteService` cache reload. Delegates to `RouteChangeEventDetector` then broadcasts matching events to all subscribers.
- Uses `CopyOnWriteArrayList` for thread-safe subscriber management.
- Inner class `RouteEventSubscription` holds topic filter (`Set<String>`), event-type filter (`Set<RouteChangeEventType>`), and the `StreamObserver` with `matches()` filter logic.
**TopicRouteService Extensions**
- Added `RouteRefreshListener` interface with `onRouteRefreshed(topic, oldView, newView)` method.
- Modified the Caffeine cache `reload()` callback to notify all registered listeners after a successful route reload, passing both old and new `MessageQueueView`.
- Added `addRouteRefreshListener()` for listener registration.
- Added `getAllTopicNames()` and `getCachedTopicRouteData()` for accessing cached route state.
**Data Models**
| Model | Description |
|-------|-------------|
| `RouteChangeEvent` | Event payload: eventType, timestamp, topic, cluster, brokerName, brokerAddress, brokerId, previous/current read/write queue nums, routeSnapshot |
| `RouteChangeEventType` | Enum with 6 values: `BROKER_ONLINE`, `BROKER_OFFLINE`, `QUEUE_SCALE`, `TOPIC_CREATE`, `TOPIC_DELETE`, `ROUTE_SNAPSHOT` |
| `TopicRouteSnapshot` | Point-in-time route state with `BrokerInfo` (cluster, brokerName, brokerAddrs map) and `QueueInfo` (brokerName, readQueueNums, writeQueueNums, perm) inner classes |
### Event Flow
```
TopicRouteService Caffeine cache reload
-> onRouteRefreshed(topic, oldView, newView)
-> RouteChangeEventDetector.detectChanges(topic, oldRoute, newRoute)
-> List<RouteChangeEvent>
-> RouteChangeNotifier broadcasts to matching subscribers
-> gRPC server-streaming response to client
```
### Files Changed
| File | Description |
|------|-------------|
| `proxy/src/main/java/.../grpc/admin/RouteChangeEventDetector.java` | Detection engine (new, 357 lines) |
| `proxy/src/main/java/.../grpc/admin/RouteChangeNotifier.java` | Subscriber management and broadcasting (new, 245 lines) |
| `proxy/src/main/java/.../grpc/admin/model/RouteChangeEvent.java` | Event POJO (new, 136 lines) |
| `proxy/src/main/java/.../grpc/admin/model/RouteChangeEventType.java` | 6-value enum (new, 31 lines) |
| `proxy/src/main/java/.../grpc/admin/model/TopicRouteSnapshot.java` | Route snapshot POJO (new, 138 lines) |
| `proxy/src/main/java/.../service/route/TopicRouteService.java` | RouteRefreshListener interface + cache reload notification (+74 lines) |
| `proxy/src/test/java/.../grpc/admin/RouteChangeEventDetectorTest.java` | 20 test methods (new, 610 lines) |
| `proxy/src/test/java/.../grpc/admin/RouteChangeNotifierTest.java` | Subscriber/broadcast tests (new, 531 lines) |
| `proxy/src/test/java/.../grpc/admin/model/RouteChangeEventTest.java` | Event model tests (new, 138 lines) |
| `proxy/src/test/java/.../grpc/admin/model/RouteChangeEventTypeTest.java` | Enum tests (new, 100 lines) |
| `proxy/src/test/java/.../grpc/admin/model/TopicRouteSnapshotTest.java` | Snapshot model tests (new, 155 lines) |
### Design Decisions
- **Passive detection**: Events are detected by comparing route snapshots during the existing Caffeine cache reload cycle (~20s default), avoiding additional network probes or broker polling.
- **Initial snapshot on subscribe**: New subscribers receive a `ROUTE_SNAPSHOT` event containing the current route state for all cached topics, enabling clients to build a complete initial view before processing incremental changes.
- **Filter support**: Subscribers can filter by topic list and/or event type, reducing unnecessary network traffic for clients interested in specific changes.
- **Thread safety**: `CopyOnWriteArrayList` for subscribers ensures safe concurrent access during route refresh callbacks, which may fire from multiple cache reload threads.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Studio] feat: Feature/rip 2 pr4 route change event [rocketmq]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #10705:
URL: https://github.com/apache/rocketmq/pull/10705#issuecomment-5140069085
## [Codecov](https://app.codecov.io/gh/apache/rocketmq/pull/10705?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `57.95711%` with `745 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 48.38%. Comparing base ([`d8c3f9e`](https://app.codecov.io/gh/apache/rocketmq/commit/d8c3f9e735c482f5e2be77dc40681a2e2ad9471e?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`37024f9`](https://app.codecov.io/gh/apache/rocketmq/commit/37024f9d75719f7cbeb690a28d19a165cc017505?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 2 commits behind head on develop.
| [Files with missing lines](https://app.codecov.io/gh/apache/rocketmq/pull/10705?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...tmq/proxy/grpc/admin/ProxyAdminProtoConverter.java](https://app.codecov.io/gh/apache/rocketmq/pull/10705?src=pr&el=tree&filepath=proxy%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Frocketmq%2Fproxy%2Fgrpc%2Fadmin%2FProxyAdminProtoConverter.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHJveHkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3Byb3h5L2dycGMvYWRtaW4vUHJveHlBZG1pblByb3RvQ29udmVydGVyLmphdmE=) | 47.44% | [244 Missing and 13 partials :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/10705?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../service/admin/DefaultProxyAdminClientService.java](https://app.codecov.io/gh/apache/rocketmq/pull/10705?src=pr&el=tree&filepath=proxy%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Frocketmq%2Fproxy%2Fservice%2Fadmin%2FDefaultProxyAdminClientService.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHJveHkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3Byb3h5L3NlcnZpY2UvYWRtaW4vRGVmYXVsdFByb3h5QWRtaW5DbGllbnRTZXJ2aWNlLmphdmE=) | 59.12% | [116 Missing and 34 partials :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/10705?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...y/service/receipt/DefaultReceiptHandleManager.java](https://app.codecov.io/gh/apache/rocketmq/pull/10705?src=pr&el=tree&filepath=proxy%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Frocketmq%2Fproxy%2Fservice%2Freceipt%2FDefaultReceiptHandleManager.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHJveHkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3Byb3h5L3NlcnZpY2UvcmVjZWlwdC9EZWZhdWx0UmVjZWlwdEhhbmRsZU1hbmFnZXIuamF2YQ==) | 0.00% | [107 Missing :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/10705?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...mq/proxy/common/BatchConsumeClientDiagnostics.java](https://app.codecov.io/gh/apache/rocketmq/pull/10705?src=pr&el=tree&filepath=proxy%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Frocketmq%2Fproxy%2Fcommon%2FBatchConsumeClientDiagnostics.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHJveHkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3Byb3h5L2NvbW1vbi9CYXRjaENvbnN1bWVDbGllbnREaWFnbm9zdGljcy5qYXZh) | 0.00% | [30 Missing :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/10705?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...he/rocketmq/proxy/common/PopReceiptHandleInfo.java](https://app.codecov.io/gh/apache/rocketmq/pull/10705?src=pr&el=tree&filepath=proxy%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Frocketmq%2Fproxy%2Fcommon%2FPopReceiptHandleInfo.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHJveHkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3Byb3h5L2NvbW1vbi9Qb3BSZWNlaXB0SGFuZGxlSW5mby5qYXZh) | 0.00% | [30 Missing :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/10705?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...mq/proxy/service/receipt/ReceiptHandleManager.java](https://app.codecov.io/gh/apache/rocketmq/pull/10705?src=pr&el=tree&filepath=proxy%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Frocketmq%2Fproxy%2Fservice%2Freceipt%2FReceiptHandleManager.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHJveHkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3Byb3h5L3NlcnZpY2UvcmVjZWlwdC9SZWNlaXB0SGFuZGxlTWFuYWdlci5qYXZh) | 27.50% | [29 Missing :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/10705?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ocketmq/proxy/service/route/TopicRouteService.java](https://app.codecov.io/gh/apache/rocketmq/pull/10705?src=pr&el=tree&filepath=proxy%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Frocketmq%2Fproxy%2Fservice%2Froute%2FTopicRouteService.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHJveHkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3Byb3h5L3NlcnZpY2Uvcm91dGUvVG9waWNSb3V0ZVNlcnZpY2UuamF2YQ==) | 4.34% | [22 Missing :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/10705?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ketmq/proxy/grpc/v2/channel/GrpcClientChannel.java](https://app.codecov.io/gh/apache/rocketmq/pull/10705?src=pr&el=tree&filepath=proxy%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Frocketmq%2Fproxy%2Fgrpc%2Fv2%2Fchannel%2FGrpcClientChannel.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHJveHkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3Byb3h5L2dycGMvdjIvY2hhbm5lbC9HcnBjQ2xpZW50Q2hhbm5lbC5qYXZh) | 43.24% | [18 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/10705?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...tmq/proxy/grpc/admin/RouteChangeEventDetector.java](https://app.codecov.io/gh/apache/rocketmq/pull/10705?src=pr&el=tree&filepath=proxy%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Frocketmq%2Fproxy%2Fgrpc%2Fadmin%2FRouteChangeEventDetector.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHJveHkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3Byb3h5L2dycGMvYWRtaW4vUm91dGVDaGFuZ2VFdmVudERldGVjdG9yLmphdmE=) | 89.47% | [5 Missing and 15 partials :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/10705?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../org/apache/rocketmq/proxy/config/ProxyConfig.java](https://app.codecov.io/gh/apache/rocketmq/pull/10705?src=pr&el=tree&filepath=proxy%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Frocketmq%2Fproxy%2Fconfig%2FProxyConfig.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHJveHkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3Byb3h5L2NvbmZpZy9Qcm94eUNvbmZpZy5qYXZh) | 50.00% | [16 Missing :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/10705?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| ... and [13 more](https://app.codecov.io/gh/apache/rocketmq/pull/10705?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## develop #10705 +/- ##
=============================================
+ Coverage 48.32% 48.38% +0.05%
- Complexity 13511 13820 +309
=============================================
Files 1380 1398 +18
Lines 101091 102863 +1772
Branches 13101 13327 +226
=============================================
+ Hits 48854 49766 +912
- Misses 46276 47037 +761
- Partials 5961 6060 +99
```
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/rocketmq/pull/10705?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Studio] feat: Feature/rip 2 pr4 route change event [rocketmq]
Posted by "zhaohai666 (via GitHub)" <gi...@apache.org>.
zhaohai666 closed pull request #10705: [Studio] feat: Feature/rip 2 pr4 route change event
URL: https://github.com/apache/rocketmq/pull/10705
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org