123123213weqw opened a new pull request, #10610:
URL: https://github.com/apache/rocketmq/pull/10610
## What is the purpose of the change
Fix #10609.
Proxy stored the response observer of a gRPC telemetry stream in
`GrpcClientChannel` when the client `SETTINGS` command arrived, but the
stream's
`onCompleted()` and `onError()` callbacks never detached it. As a result a
closed telemetry stream remained referenced by the channel, and
`isActive()` / `isOpen()` / `isWritable()` kept returning `true` until a
later
`writeTelemetryCommand` failed (or another cleanup path removed the channel).
This makes the channel liveness state reflect stream closure immediately,
while
ensuring a delayed callback from an old stream cannot affect a reconnected
client's newer observer.
## Brief changelog
- `proxy/.../grpc/v2/client/ClientActivity.java`
- The `telemetry` stream observer now captures the `GrpcClientChannel`
bound
to this stream while processing `SETTINGS`.
- `onCompleted()` and `onError()` detach the stream observer from that
channel via `clearClientObserver(responseObserver)` (null-safe).
- `processAndWriteClientSettings` now returns the registered
`GrpcClientChannel` so the stream observer can clear the correct channel.
- `proxy/.../grpc/v2/channel/GrpcClientChannel.java`
- `clearClientObserver` is widened from `protected` to `public` so the
cross-package `ClientActivity` can invoke it. It still uses
`compareAndSet(future, null)`, so only the observer that actually belongs
to the finishing stream is removed.
Producer/Consumer unregistration is intentionally untouched; it continues to
use
the existing termination and heartbeat timeout mechanisms, as noted in the
issue.
## How was this patch verified
Added/extended unit tests:
- `ClientActivityTest`
- `testTelemetryOnCompletedClearsClientObserver`: channel becomes inactive
after `onCompleted()`.
- `testTelemetryOnErrorClearsClientObserver`: channel becomes inactive
after
`onError()`.
- `testOldStreamCompletionDoesNotClearNewerObserver`: after a reconnect, a
delayed `onCompleted()` of the old stream does not detach the newer
observer; only the active stream's completion detaches it.
- `GrpcClientChannelTest`
- `testSetAndClearClientObserver`: set/clear drives
`isActive`/`isOpen`/`isWritable`.
- `testClearClientObserverDoesNotClearNewerObserver`: `compareAndSet`
prevents
an old observer from clearing a newer one.
Ran the focused tests and checkstyle under JDK 17:
```
mvn -pl proxy -am test -Dtest=ClientActivityTest,GrpcClientChannelTest
-DfailIfNoTests=false \
-Dcheckstyle.skip=true -Drat.skip=true -Dspotbugs.skip=true
-Denforcer.skip=true
# Tests run: 21, Failures: 0, Errors: 0, Skipped: 0 (BUILD SUCCESS)
mvn -pl proxy validate -Drat.skip=true
# You have 0 Checkstyle violations. (BUILD SUCCESS)
```
--
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]
