zhaohai666 opened a new pull request, #10707:
URL: https://github.com/apache/rocketmq/pull/10707
## [Studio] feat: Admin Service Integration with Messaging Pipeline and E2E
Tests
**Issue:** #10637
**Branch:** `feature/rip-2-pr6-messaging-integration`
---
### Overview
This PR completes the RIP-2 Proxy Admin Interface by integrating the admin
service with the messaging pipeline and providing end-to-end integration tests.
It wires the admin client service into the gRPC data plane for heartbeat
recording, enriches client channels with connection diagnostics metadata,
extends the receipt handle manager with diagnostic query capabilities, and
validates the entire system through comprehensive E2E tests.
### Key Changes
**Messaging Pipeline Integration**
- **GrpcMessagingActivity** (interface): Added `getGrpcChannelManager()` and
`getGrpcClientSettingsManager()` to expose internal managers for admin queries.
- **DefaultGrpcMessagingActivity**: Implements the new getters; added
`setProxyAdminClientService()` to wire the admin service for heartbeat
recording (RIP-2 section 5.2.2).
- **GrpcMessagingApplication**: Delegates the new getters and setter to the
activity layer.
- **GrpcChannelManager**: Added `getClientIdChannelMap()` returning an
unmodifiable view of the client-to-channel mapping for admin queries.
**Client Connection Diagnostics** (RIP-2 section 5.2)
- **GrpcClientChannel** (+121 lines): Enriched with per-connection metadata:
- `createTime` -- connection establishment timestamp.
- `lastRttMs` (volatile) -- estimated round-trip time from telemetry
command round-trip.
- `sslEnabled` (volatile) -- TLS state captured at connection time.
- `authUsername` (volatile) -- authenticated username from ACL.
- `updateRttMeasurement()` -- computes RTT from
`System.currentTimeMillis() - lastTelemetrySendTimeMs`.
- `forceClose(String reason)` -- sends `Status.UNAVAILABLE` error to
client via telemetry stream for server-side disconnection.
**ClientActivity Integration**
- Added `proxyAdminClientService` field (volatile, set post-construction to
break circular dependency).
- In `heartbeat()`: calls `recordHeartbeat(ctx.getClientID())` after
processing.
- In telemetry `onNext()`: calls `updateRttMeasurement(ctx)` and
`recordHeartbeat(ctx.getClientID())` on SETTINGS commands.
- Null-safe delegation with error handling to prevent data plane disruption.
**MessagingProcessor Extensions**
- Added `popLiteMessage()` method signature (same params as `popMessage`).
- Added `getReceiptHandleManager()` -- returns `ReceiptHandleManager` for
POP diagnostics.
- Added `getTopicRouteService()` -- returns `TopicRouteService` for route
change event subscriptions.
- `DefaultMessagingProcessor` implements all three, delegating to
`receiptHandleProcessor` and `serviceManager`.
**Receipt Handle Diagnostic Queries**
- **ReceiptHandleManager** (interface, +140 lines): Added
`describePopReceiptHandles()` and `describeBatchConsumeDiagnostics()` with
pagination. New inner classes: `PopReceiptHandleDiagnosticResult`,
`ChannelBatchConsumeData`, `BatchConsumeDiagnosticResult`.
- **DefaultReceiptHandleManager** (+216 lines): Implements both diagnostic
methods by scanning the concurrent receipt handle group map. Uses inner
`ChannelAggregator` class for per-channel stat aggregation during batch
diagnostics.
### End-to-End Integration Test
**ProxyAdminGrpcIT** (602 lines) -- Full gRPC client-to-admin-server
integration test (RIP-2 M1 milestone).
Test environment: Namesrv + 3 Brokers, data plane gRPC server
(`MessagingServiceGrpc`), admin plane gRPC server
(`ProxyAdminBindableService`), both with TLS (SelfSignedCertificate).
| Test Method | Scenario |
|-------------|----------|
| `test01_ListClients_NoClients_ReturnsOkWithEmptyList` | Empty state
returns OK with 0 total |
| `test02_DescribeClient_NonExistentClient_ReturnsNotFound` | Returns
NOT_FOUND for unknown client |
| `test03_ListClientsByGroup_NonExistentGroup_ReturnsOkWithEmptyList` |
Empty result for unknown group |
| `test04_ListClientsByTopic_NonExistentTopic_ReturnsOkWithEmptyList` |
Empty result for unknown topic |
| `test05_ListClients_Pagination` | Verifies pageSize/pageNum in response |
| `test06_ListClients_WithConnectedProducer` | Connects a real producer via
telemetry, verifies it appears in admin list |
| `test07_DescribeClient_WithConnectedProducer` | Describes a connected
producer, verifies clientId in response |
| `test08_ListClientsByGroup_WithConnectedConsumer` | Connects a push
consumer, queries by group |
| `test09_ListClientsByTopic_WithConnectedProducer` | Connects a producer,
queries by topic |
| `test10_ListClients_WithLanguageFilter` | Verifies JAVA filter includes,
GOLANG filter excludes the same Java client |
### Files Changed
| File | Description |
|------|-------------|
| `proxy/src/main/java/.../grpc/v2/GrpcMessagingActivity.java` | Added
channel/settings manager getters |
| `proxy/src/main/java/.../grpc/v2/DefaultGrpcMessagingActivity.java` |
Implemented getters + admin service setter |
| `proxy/src/main/java/.../grpc/v2/GrpcMessagingApplication.java` |
Delegated new methods to activity |
| `proxy/src/main/java/.../grpc/v2/channel/GrpcChannelManager.java` |
Exposed clientId-channel map |
| `proxy/src/main/java/.../grpc/v2/channel/GrpcClientChannel.java` |
Connection metadata + forceClose (+121 lines) |
| `proxy/src/main/java/.../grpc/v2/client/ClientActivity.java` | Heartbeat
recording + RTT measurement (+68 lines) |
| `proxy/src/main/java/.../processor/MessagingProcessor.java` | Added
popLiteMessage + getter methods |
| `proxy/src/main/java/.../processor/DefaultMessagingProcessor.java` |
Implemented new interface methods |
| `proxy/src/main/java/.../processor/ReceiptHandleProcessor.java` | Exposed
ReceiptHandleManager |
| `proxy/src/main/java/.../service/receipt/ReceiptHandleManager.java` |
Diagnostic query interface (+140 lines) |
| `proxy/src/main/java/.../service/receipt/DefaultReceiptHandleManager.java`
| Diagnostic implementation (+216 lines) |
| `proxy/src/main/java/.../service/channel/SimpleChannel.java` | Added
getLastAccessTime() |
| `WORKSPACE` | Added rules_proto dependency for E2E test |
| `test/src/test/java/.../grpc/admin/ProxyAdminGrpcIT.java` | E2E
integration test (new, 602 lines) |
| `proxy/src/test/java/.../receipt/ReceiptHandleDiagnosticResultTest.java` |
Diagnostic result tests (new, 172 lines) |
### Integration Architecture
```
Data Plane (port 8081) Admin Plane (port 8082)
GrpcMessagingApplication ProxyAdminGrpcService
-> DefaultGrpcMessagingActivity -> ProxyAdminBindableService
-> ClientActivity -> ProxyAdminClientService
-> heartbeat recording -> GrpcChannelManager
(read)
-> RTT measurement ->
GrpcClientSettingsManager (read)
-> GrpcChannelManager -> ReceiptHandleManager
(read)
-> GrpcClientSettingsManager -> RouteChangeNotifier
(subscribe)
```
--
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]
