apache / apache/rocketmq

[Bug] Proxy keeps a stale gRPC telemetry observer after stream closure

Open
#10,609 3 comments 0 reactions 0 assignees View on GitHub
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.