[ISSUE #10609] Detach telemetry observer when gRPC stream closes

[123123213weqw]

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)

[RockteMQ-AI]

Review by github-manager-bot

Summary

Fixes #10609 — detaches the telemetry response observer from GrpcClientChannel when the gRPC stream closes (onCompleted / onError), so that channel liveness state (isActive / isOpen / isWritable) reflects stream closure immediately instead of lingering until a later write failure.

Findings

• [Info] ClientActivity.java — The new grpcClientChannel field on the anonymous ContextStreamObserver is not volatile. gRPC serialises callbacks per stream, so this is safe in practice, but adding volatile would make the intent explicit and guard against future refactors that might introduce cross-thread access.
• [Info] ClientActivity.java:processAndWriteClientSettings — Return type changed from void to GrpcClientChannel. This is a breaking change for any subclass that overrides this method (compile error). If there are known external subclasses, consider deprecating the old signature first or adding an overload.
• [Info] GrpcClientChannel.java:clearClientObserver — Visibility widened from protected to public. Backward-compatible, but consider whether package-private + a dedicated accessor on a shared interface would be a tighter encapsulation.

Suggestions

1. Consider volatile on grpcClientChannel — Low cost, documents the cross-callback visibility contract:

   private volatile GrpcClientChannel grpcClientChannel = null;

2. The compareAndSet in clearClientObserver correctly prevents a stale stream's delayed callback from clearing a newer observer — good use of CAS for reconnection safety.

Tests

Test coverage is thorough:

• ✅ Basic set/clear cycle (testSetAndClearClientObserver)
• ✅ CAS prevents old observer from clearing newer one (testClearClientObserverDoesNotClearNewerObserver)
• ✅ onCompleted clears channel state (testTelemetryOnCompletedClearsClientObserver)
• ✅ onError clears channel state (testTelemetryOnErrorClearsClientObserver)
• ✅ Reconnect safety — old stream completion doesn't detach newer observer (testOldStreamCompletionDoesNotClearNewerObserver)

Cross-repo Note

This change is Proxy-side only and does not affect the gRPC wire protocol or client SDKs. No coordinated change needed in apache/rocketmq-clients.

---

Automated review by github-manager-bot