[Bug] Proxy keeps a stale gRPC telemetry observer after stream closure
- Dominant language
- Java
- Stars
- 22.6k
- Forks
- 12k
- Avg merge
- 3d 1h
- Merged PRs (30d)
- 27
Description
### 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.
Contributor guide
Research direction
Start by tracing ClientActivity#telemetry and the observer lifecycle through GrpcClientChannel#setClientObserver, clearClientObserver, and writeTelemetryCommand. Reproduce completion and error callbacks, then verify that only the closed stream's observer is detached and that a newer observer remains active after reconnection.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- grpc, java
- Domain
- backend, distributed-systems
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Quiet
- Clarity
- Clearly specified
- Newbie friendliness
- 68/100