You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by "yin-bo-Final (via GitHub)" <gi...@apache.org> on 2026/07/11 17:23:38 UTC
[I] [Bug] Proxy keeps a stale gRPC telemetry observer after stream closure [rocketmq]
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: commits-unsubscribe@rocketmq.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Bug] Proxy keeps a stale gRPC telemetry observer after stream closure [rocketmq]
Posted by "RockteMQ-AI (via GitHub)" <gi...@apache.org>.
RockteMQ-AI commented on issue #10609:
URL: https://github.com/apache/rocketmq/issues/10609#issuecomment-4956532885
Thanks for picking this up! No one else is working on it — PR #10610 looks like the right approach. I've left a review there with some suggestions. Looking forward to seeing the final version. 👍
---
*Automated reply by github-manager-bot*
--
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: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Bug] Proxy keeps a stale gRPC telemetry observer after stream closure [rocketmq]
Posted by "RockteMQ-AI (via GitHub)" <gi...@apache.org>.
RockteMQ-AI commented on issue #10609:
URL: https://github.com/apache/rocketmq/issues/10609#issuecomment-4948214252
**Issue Evaluation**
Category: `type/bug` | Status: **Confirmed**
The reported issue has been verified against the current `develop` branch (c4ce550a574e).
**Root Cause:** `ClientActivity.telemetry()` stores the gRPC `responseObserver` in `GrpcClientChannel` via `setClientObserver()`, but neither `onCompleted()` nor `onError()` calls `clearClientObserver()`. Since `isActive()`/`isOpen()`/`isWritable()` all check `telemetryCommandRef.get() != null`, a closed stream continues to appear active.
**Impact:** Proxy may route operations to or consider alive channels whose telemetry stream has already terminated. The stale reference is only eventually cleared when the next `writeTelemetryCommand` throws — creating a window of incorrect liveness state.
**Severity:** Medium — affects channel lifecycle accuracy in the Proxy gRPC layer.
---
**Proposed Fix Summary:**
1. **`ClientActivity.java`** — In the anonymous `ContextStreamObserver` inside `telemetry()`:
- Capture the `GrpcClientChannel` reference when settings are processed
- In `onCompleted()`: after `responseObserver.onCompleted()`, call `grpcClientChannel.clearClientObserver(responseObserver)`
- In `onError()`: after `handleGrpcCancel()`, call `grpcClientChannel.clearClientObserver(responseObserver)`
2. **`GrpcClientChannel.java`** — No changes needed. The existing `clearClientObserver()` already uses `compareAndSet(future, null)`, which safely handles the race condition where a delayed old-stream callback must not clear a newer stream's observer.
3. **Edge cases covered:**
- Stream closes before any SETTINGS → `grpcClientChannel` is null → skip clear (safe)
- Multiple topics for producer → final channel assignment is sufficient (same channel via `clientId`)
- Concurrent new stream → `compareAndSet` protects the new observer
An automated fix proposal will be generated. Reply `/approve` to proceed with PR generation, `/revise <feedback>` to request changes, or `/reject` to decline.
---
*Automated evaluation by github-manager-bot*
--
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: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Bug] Proxy keeps a stale gRPC telemetry observer after stream closure [rocketmq]
Posted by "123123213weqw (via GitHub)" <gi...@apache.org>.
123123213weqw commented on issue #10609:
URL: https://github.com/apache/rocketmq/issues/10609#issuecomment-4949451600
I've read the issue and will reproduce it, then open a focused fix with verification results attached.
--
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: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org