Before Creating the Bug Report
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:
grpcClientChannel.setClientObserver(responseObserver);
GrpcClientChannel uses this observer to determine whether the channel is active:
@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:
@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
- Create a telemetry stream through
ClientActivity#telemetry.
- Send a valid settings command so that the response observer is stored in
GrpcClientChannel.
- Verify that the channel is active.
- Complete the telemetry stream by calling
onCompleted().
- 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:
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:
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.
Before Creating the Bug Report
I found a bug, not just asking a question, which should be created in GitHub Discussions.
I have searched the GitHub Issues and GitHub Discussions of this repository and believe that this is not a duplicate.
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
developbranch.JDK Version
JDK 17
Describe the Bug
Proxy stores the response observer of a gRPC telemetry stream in
GrpcClientChannelwhen it receives the client settings:GrpcClientChanneluses this observer to determine whether the channel is active:However, when the telemetry stream is completed or closed with an error,
ClientActivity#telemetrydoes not clear the observer:Therefore, a closed telemetry stream may remain referenced by the channel, and
isOpen(),isActive(), andisWritable()may continue to returntrue.The observer is currently cleared only when a later call to
writeTelemetryCommandfails.Steps to Reproduce
ClientActivity#telemetry.GrpcClientChannel.onCompleted().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
telemetryCommandRefafter 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:
GrpcClientChannel#clearClientObserveralready usescompareAndSet, which may help avoid clearing a newer observer when an old stream finishes later: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.