yin-bo-Final opened a new issue, #10609: URL: https://github.com/apache/rocketmq/issues/10609
### Before Creating the Bug Report - [x] I found a bug, not just asking a question, which should be created in [GitHub Discussions](https://github.com/apache/rocketmq/discussions). - [x] I have searched the [GitHub Issues](https://github.com/apache/rocketmq/issues) and [GitHub Discussions](https://github.com/apache/rocketmq/discussions) of this repository and believe that this is not a duplicate. - [x] I have confirmed that this bug belongs to the current repository, not other repositories of RocketMQ. ### Runtime platform environment Ubuntu 24.04 ### RocketMQ version Latest `develop` branch. ### JDK Version JDK 17 ### Describe the Bug Proxy stores the response observer of a gRPC telemetry stream in `GrpcClientChannel` when it receives the client settings: ```java grpcClientChannel.setClientObserver(responseObserver); ``` `GrpcClientChannel` uses this observer to determine whether the channel is active: ```java @Override public boolean isActive() { return this.telemetryCommandRef.get() != null; } ``` However, when the telemetry stream is completed or closed with an error, `ClientActivity#telemetry` does not clear the observer: ```java @Override public void onError(Throwable t) { log.error("telemetry on error", t); handleGrpcCancel(proxyCtx, t); } @Override public void onCompleted() { responseObserver.onCompleted(); } ``` Therefore, a closed telemetry stream may remain referenced by the channel, and `isOpen()`, `isActive()`, and `isWritable()` may continue to return `true`. The observer is currently cleared only when a later call to `writeTelemetryCommand` fails. ### Steps to Reproduce 1. Create a telemetry stream through `ClientActivity#telemetry`. 2. Send a valid settings command so that the response observer is stored in `GrpcClientChannel`. 3. Verify that the channel is active. 4. Complete the telemetry stream by calling `onCompleted()`. 5. Check the state of the same channel. The channel still reports itself as active because the closed stream observer remains in `telemetryCommandRef`. The same behavior can be observed when the stream terminates through `onError()`. ### What Did You Expect to See? The observer associated with the closed telemetry stream should be detached from `GrpcClientChannel`, so the closed stream is no longer considered active or writable. Cleanup should only affect the observer belonging to that stream. If the client has already reconnected, a delayed callback from the old stream must not clear the new observer. ### What Did You See Instead? The observer remains in `telemetryCommandRef` after the stream is closed. As a result, Proxy may continue to treat the closed telemetry stream as active until a later write fails or the channel is removed by other cleanup logic. ### Additional Context The relevant methods are: ```text ClientActivity#telemetry GrpcClientChannel#setClientObserver GrpcClientChannel#clearClientObserver GrpcClientChannel#writeTelemetryCommand ``` `GrpcClientChannel#clearClientObserver` already uses `compareAndSet`, which may help avoid clearing a newer observer when an old stream finishes later: ```java this.telemetryCommandRef.compareAndSet(future, null); ``` Closing the telemetry stream does not necessarily mean that the whole client has terminated. This report only concerns detaching the observer of the closed stream; Producer and Consumer unregistration can continue to use the existing termination and heartbeat timeout mechanisms. -- 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]
