zhaohai666 opened a new pull request, #1404:
URL: https://github.com/apache/rocketmq-clients/pull/1404
# fix(rust): keep the telemetry session alive across server rolling
deployments
## Summary
This PR fixes a cluster of interrelated session-lifecycle defects in the
Rust client
(branch `fix/rust-telemetry-session-defects`, **4 commits** touching
`rust/src/session.rs`,
`rust/src/client.rs`, `rust/src/simple_consumer.rs`,
`rust/src/push_consumer.rs`,
`rust/tests/rolling_restart_it.rs`), all aligned with the Java reference
client.
Before the fix, every rolling deployment of the server permanently broke the
client's
long-lived session: the `Settings` object never grew (consistently 88 bytes)
because the client
degraded into one-off telemetry streams lasting 3–43 ms, one per 30-second
sync cycle. The
accumulation of dead endpoints made every settings report cycle fail and
interrupted consumption
for 8–50 seconds. Two secondary defects made things worse: the gRPC deadline
of `receive` did not
match the documented contract (100% failure when `long_polling_timeout` was
set below 5s), and
settings pushed by the server were silently discarded.
**Root causes, all fixed here**
1. `sync_settings` rebuilt the telemetry stream on a *shadow* session — a
throw-away clone — so
the rebuilt stream was lost immediately.
2. The session map only ever grew: dead endpoints were never evicted, so
every report cycle kept
failing on them.
3. No HTTP/2 keep-alive, so a server-side rolling deployment silently broke
the long-lived
connection.
4. After a failover the new server instance answers unary RPCs with `40015
cannot find client` /
`50001 settings is null`; the client only logged those errors and never
re-registered.
5. Server-pushed settings were discarded by `SimpleConsumer` and had no
effect on the receive
deadline even for `PushConsumer`.
6. Five code paths panicked (`unwrap`) on session lookup when the server was
unreachable.
---
## Fix 1: reuse the registered session and keep the session map bounded
(`74ccc3a6`)
- `SessionManager::sync_settings` now syncs on the **registered** session
and evicts it when the
sync is unrecoverable, instead of rebuilding the stream on a shadow clone
that is dropped right
away.
- Sessions are tracked with a shared last-activity timestamp
(`shadow_session()` shares the
`Arc<AtomicU64>`), plus `is_healthy()` / `idle_duration()`. A periodic
task (every 30s) drops
sessions whose telemetry stream is gone or that nobody has used for 30
minutes, mirroring
`ClientManagerImpl#clearIdleRpcClients` in Java.
- HTTP/2 keep-alive is configured on every endpoint (300s interval / 30s
timeout /
`keep_alive_while_idle`), aligned with `RpcClientImpl.KEEP_ALIVE_*` of the
Java client. The
interval must stay at 300s because grpc-java servers reject faster client
pings with
`GOAWAY(ENHANCE_YOUR_CALM)`.
- The receive deadline is now `long_polling_timeout + timeout`, as
documented and as the Java
client computes it (`awaitDuration.plus(tolerance)`). With the old
deadline the server could
hold the response for exactly as long as the client was willing to wait,
so every poll failed
once `long_polling_timeout` was short. Verified against a live cluster:
the old behaviour
produced `code 40000 "The deadline time remaining is not enough for
polling"` from the server.
- `PushConsumer` applies server-pushed settings (`receive_batch_size`,
`long_polling_timeout`, `fifo`), mirroring `PushSubscriptionSettings#sync`.
## Fix 2: propagate session errors instead of panicking (`07b4c5cb`)
`send_message`, `recall_message`, `receive_message`, `ack_message` and
`change_invisible_duration` called `.unwrap()` on the session lookup. A
server-side rolling
deployment made the lookup fail, and the panic took down the whole client
process — an outage
amplifier on top of an outage. All five paths now propagate the error with
`?`.
## Fix 3: proactive, error-driven session recovery (`9c5c0ff2`)
Mirrors the Java client's `ClientSessionImpl#renewRequestObserver` and
`ClientImpl#recoverTransport`:
- The telemetry stream task notifies the client layer through a
stream-closed channel the moment
the server closes the stream (EOF or error) — no more waiting for the next
periodic sync.
- `Client::start` listens on that channel and schedules a reconnect that
retries on the
**registered** session **every second**. The delay is deliberately fixed,
not exponential: with
a growing back-off the client keeps missing the moment the new instance
becomes ready, which
measurements confirmed. `SessionManager::reconnect_session` takes the
session out of the map
for the duration of the RPC (so other callers are never blocked on the
network round trip) and
puts it back on failure, so an idle client always keeps a driver for
recovery.
Re-establishing the stream also re-registers the client with the new
server instance.
- A heartbeat *business* error (e.g. `40015 cannot find client`, `50001
settings is null` after a
failover) now triggers the same reconnect instead of being only logged.
The heartbeat interval
drops from 30s to 10s (`ClientImpl#HEART_BEAT_PERIOD`), and a session is
evicted only after 2
consecutive transport failures (`HEART_BEAT_FAILURE_THRESHOLD`), so a
transient blip no longer
tears down a working connection.
- The long-polling timeout lives behind a shared handle
(`Arc<parking_lot::RwLock<Duration>>`): when the server pushes new
settings, the new value
takes effect on the receive deadline **and** on the request body of every
consumer without
recreating anything.
- `SimpleConsumer` no longer discards server-pushed settings: it applies the
pushed
`long_polling_timeout` (`SimpleSubscriptionSettings#sync`).
## Fix 4: integration-test robustness (`a6f99a46`)
The live integration test tolerates the client recovering faster than the
external readiness
probe.
---
## Testing
**Offline suite** (`cargo fmt`, `cargo clippy --all-features -- -D warnings`
— the CI gate,
`cargo test --all-features`): all green, 98 lib tests including new coverage
for
`reconnect_session` (stream rebuilt on the registered session / session kept
on failure /
no-op for unknown endpoints), settings-sync eviction, shared long-polling
timeout, and the
receive-deadline contract.
**Live-cluster verification** (RocketMQ 5.5.1, remoting broker 10911 + gRPC
proxy 8081 in
CLUSTER mode, `kill -9` of the proxy to simulate a rolling deployment):
| Metric | Before | After |
|---|---|---|
| Telemetry stream pattern after deployment | one-off 3–43 ms streams, one
per 30s cycle, forever | single long-lived stream, re-established within one
retry cycle |
| Dead-endpoint accumulation | every settings report fails, 8–50s
consumption gaps per cycle | none (eviction + 2-failure threshold) |
| Client-side recovery after the new instance is ready | 30s sync polling,
no re-registration | ≤ 2s (next 1s retry tick) |
| Measured end-to-end recovery window (`kill -9` → first successful receive)
| 42–70s | **5.9s** when broker routes are warm (~20s when the proxy must
rebuild routes from broker heartbeats, which no client can skip) |
| Server-pushed settings | discarded by SimpleConsumer; no deadline effect |
applied by both consumers; deadline + request body updated live |
| Unreachable server during `receive`/`send`/... | process panic | error
propagated |
The remaining window is server-side: the restarted proxy itself needs time
to start and to
rebuild topic routes; the client reconnects on the first retry tick once the
gRPC port accepts
connections. `SimpleConsumer` applying a server-pushed
`long_polling_timeout: 20s` was observed
in the live test logs.
The rolling-restart scenario is captured in
`rust/tests/rolling_restart_it.rs`. It is opt-in
(`ROCKETMQ_ROLLING_IT=1`) because it needs a live cluster plus an external
driver that kills and
restarts the proxy while the test measures the interruption window; in a
plain `cargo test` run
it returns immediately.
---
## Commits
- `74ccc3a6` — stop rebuilding telemetry streams on shadow sessions and keep
the session map bounded
- `07b4c5cb` — propagate session errors instead of panicking when the server
is unreachable
- `9c5c0ff2` — recover the telemetry stream proactively instead of waiting
for periodic syncs
- `a6f99a46` — tolerate the client recovering faster than the readiness
probe (test)
--
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]
