guyinyou opened a new pull request, #1377: URL: https://github.com/apache/rocketmq-clients/pull/1377
### Which Issue(s) This PR Fixes Aligns the Go SDK with the recovery behavior introduced by the **Java reference implementation: [PR #1310](https://github.com/apache/rocketmq-clients/pull/1310)** (commit `9fe1449d19449b41442aa3a97ab168ed6b5bd6b1`). The C# counterpart is #1371. The Java PR is treated as the behavioral reference, not as an API to copy. ### Brief Description A half-open TCP connection (peer gone without FIN/RST) still reports `Ready` to gRPC, so every RPC blocks until its deadline and the client never recovers. Recovery is now driven by connectivity evidence instead of ordinary request failures. - **Triggers.** Only heartbeat connectivity failures and the server-initiated `ReconnectEndpointsCommand` may replace a transport; ordinary unary failures no longer evict a connection, so `handleGrpcError` and its call sites are gone. - **Policy.** Recover after two consecutive `DeadlineExceeded` heartbeats, or on `Unavailable` only while the old connection still reports `Ready`; anything else clears the counter. A 30-second per-endpoint cooldown applies, which a server command bypasses but the in-progress guard never. Bookkeeping is keyed on the entry instance, so a late completion cannot account against its successor. - **Go adaptation.** gRPC-Go cannot evict a `Ready` connection: `Connect`/`ResetConnectBackoff` do not, `enterIdle` is unexported, and the idle timer never fires while a telemetry stream is active. Recovery therefore rebuilds the `ClientConn` from the options frozen at construction, publishes it under the cache lock only after re-checking that it is still current, wanted and not stopped, closes the replaced transport outside the lock, closes a rejected candidate, and then asks the affected telemetry sessions to reconnect. In-flight dials are tracked separately from cache entries so retiring an endpoint cannot release the guard on its address. - **Telemetry.** Each session gets one supervisor goroutine. It renews the stream after a 1s cancellable backoff instead of waiting for the next publish, resends `Settings` on every new stream, cancels the stream context *before* `CloseSend` so a `Send` stuck on a half-open connection cannot block shutdown, and reports the startup result once through a channel instead of an unsynchronized field. - **Route lifecycle.** Route updates count the endpoints they are still initializing, so a session that is starting is not retired; deprecated endpoints drop their recovery bookkeeping while transport retirement stays with the pre-existing 30-minute idle reaper (matching Java); route lookups clone endpoints before `EndpointsToString`, which sorts its input in place. - **Throttling.** Transport `ResourceExhausted` is normalized at the client manager boundary, for unary calls and the `ReceiveMessage` stream, into the SDK throttling code `42900` while keeping the original gRPC status reachable through `errors.Is`/`errors.As`. `AsErrRpcStatus` now uses `errors.As`, which also fixes it giving up on an error wrapped more than once. The producer treats protocol and transport throttling alike and backs off on a timer the caller's context can cancel. - **Concurrency fixes required by the above.** Because renewals re-apply settings while other paths read them, the producer retry policy is replaced as a whole under a lock and cloned into outbound settings instead of being mutated in place, and the meter provider's `clientMeter` pointer became atomic (`Reset` swaps it on the telemetry goroutine while `isEnabled`/`record` read it). Both are only visible under `-race`. **Compatibility.** Public API and wire protocol are unchanged: `ClientManager`/`RpcClient` gain no methods (the new capabilities are unexported optional interfaces resolved by type assertion), `ErrRpcStatus` keeps its fields, `IsEndpointUpdated`/`SetReceiveReconnect` keep their semantics (the flag is now behind accessors), and the default timers are untouched (heartbeat 10s, request timeout 3s, telemetry stream 365d, idle reaper 30min). Periodic route updates now resynchronize settings, which aligns with Java and is a prerequisite for correct recovery. **Deliberately out of scope**, both pre-existing on master and both to be sent separately: `pushConsumerSettings` fields (`isFifo`, `receiveBatchSize`, `longPollingTimeout`, `retryPolicy`) are still unsynchronized while `syncSettings` re-applies them every 5 minutes from the telemetry goroutine; and `process_queue.go` skips `doAfter` on `MESSAGE_NOT_FOUND`, so the inflight receive counter only grows and every push consumer shutdown waits the full `requestTimeout + longPollingTimeout` (~33s) before logging a timeout. **Commit guide** (feature and test changes are separate commits, and each commit builds, compiles its tests and passes the suite on its own): 1. `fix(golang): make the RPC transport rebuildable and its close idempotent` 2. `fix(golang): keep concurrently re-applied settings race-free` 3. `test(golang): cover the producer retry policy snapshot` 4. `fix(golang): recover half-open transports from heartbeat and server signals` 5. `test(golang): cover the recovery state machine and telemetry lifecycle` 6. `fix(golang): map gRPC RESOURCE_EXHAUSTED to the SDK throttling code` 7. `test(golang): cover throttling mapping and finish the local mock migration` 8. `test(golang): add a half-open TCP integration test` ### How Did You Test This Change? Verified locally on **macOS arm64**, Go 1.24.2 with gRPC-Go 1.72.0: - [x] `go build ./...` and `go vet ./...`: clean, **0** `lostcancel` warnings (master reports 13); `gofmt` clean; `codespell` 2.1.0 with the CI arguments. - [x] `go test -count=1 ./...`: ok (31.0s). `go test -race -count=1 ./...`: ok (32.5s), **0 data races**. Master's suite is not race-clean (`TestCMClearIdleRpcClients` read the cache map without the lock); this branch is, without skipping tests or weakening assertions. - [x] `GOMAXPROCS=2 go test -race -count=2` over the recovery, session, route, producer and metric groups: ok (58s). - [x] **Blackhole TCP proxy integration** (`TestProducerRecoversFromHalfOpenTCP`): a real producer against a loopback gRPC server behind a proxy that keeps reading both directions and discards the bytes, so the established connection goes black without a FIN or RST. Recovery in **~26s**, inside the **40s** bound, with the send succeeding on a **newly accepted** connection, settings resynced there, and the old transport untouched before the two heartbeat deadlines. Run three times consecutively under `-race`. - [x] Deterministic unit coverage asserted through channel barriers (no sleeps, no self-releasing timeouts): threshold and first-timeout-no-recovery, the five non-`Ready` states, counter resets, cooldown suppression and expiry, a server command bypassing the cooldown but not the in-progress guard (32 competing goroutines), retirement keeping the transport while clearing bookkeeping and canceling an in-flight recovery, a canceled recovery never publishing its candidate, re-entry recovering through a fresh job, late heartbeats, shutdown waiting for a late candidate, an initial dial that is cancellable and never holds the cache lock, the unary normalization matrix, EOF renewal resending settings, a blocked `Send` released by reconnect and by release, startup success and failure, route retirement, and pending references protecting an endpoint that is still initializing. - [x] Per-commit verification: all 8 commits individually pass `go build`, test compilation and the full suite (bisect-friendly). - [x] Against a real RocketMQ 5.0 instance: a 4-minute end-to-end run (452 normal and 450 FIFO messages sent and received, **0 ordering violations**, 0 failures, clean shutdown on SIGINT) and a **1000 x 50** producer create/destroy leak run — goroutines **124 -> 124 (leaked 0)**, **2004 transports created and 2004 closed**, **1002 clients started and 1002 terminated**, 0 errors, no panic and no transport error of any kind. - [ ] Cross-platform: Ubuntu/Windows results are left to this PR's CI runs; not verified locally. -- 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]
