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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
