Definition
- Stream-level errors – delivered as Reactive Streams error signals on
Mono/Flux, handled withonErrorResume(substitute a fallback) andretryWhen(retry with backoff). - Keep-alive – periodic KEEPALIVE frames between peers; if no acknowledgment arrives within the max-lifetime window, the connection is declared dead.
- Session resumption – a resume token is negotiated on connect; on transport drop, the client reconnects and the protocol replays unacknowledged frames.
- Server-side mapping –
@MessageExceptionHandlerconverts domain exceptions into clean error payloads so clients see structured errors. - Idempotency – retries on money-moving operations must carry a client-generated idempotency key to prevent double-debit.
Imagine Rajesh is on a live call with National Bank of India's trading desk. If the desk hits a problem answering one question, it tells Rajesh "that failed" and his app can either show a cached fallback (onErrorResume) or politely ask again a moment later (retryWhen with backoff). Meanwhile the phone line has a heartbeat: both sides say "still there?" every few seconds (keep-alive). If the line drops, the system dials back and picks up the exact same conversation where it left off using a claim ticket (resume token), so Rajesh does not lose his place in the live price feed.
Understanding the Concepts
How do you handle stream-level errors?
Stream-level errors arrive as Reactor error signals. You compose recovery operators directly on the returned publisher.
package com.bank.rsocket;
import io.rsocket.RSocket;
import io.rsocket.util.DefaultPayload;
import reactor.core.publisher.Mono;
import java.math.BigDecimal;
public class StreamErrorHandling {
public Mono<BigDecimal> getBalance(RSocket rSocket, String accountId) {
return rSocket.requestResponse(DefaultPayload.create(accountId))
.map(payload -> new BigDecimal(payload.getDataUtf8()))
.onErrorResume(ex -> {
System.out.println("Failed: " + ex.getMessage()
+ ", returning cached balance");
return Mono.just(cachedBalance(accountId));
});
}
private BigDecimal cachedBalance(String accountId) {
return new BigDecimal("150000.00"); // Rajesh's last known balance
}
}
How does retryWhen work with backoff?
retryWhen with exponential backoff and jitter recovers transient faults. A predicate filter ensures you only retry retryable errors.
package com.bank.rsocket;
import io.rsocket.RSocket;
import io.rsocket.util.DefaultPayload;
import reactor.core.publisher.Mono;
import reactor.util.retry.Retry;
import java.math.BigDecimal;
import java.time.Duration;
public class RetryDemo {
public Mono<BigDecimal> getBalanceWithRetry(RSocket rSocket, String accountId) {
return rSocket.requestResponse(DefaultPayload.create(accountId))
.map(payload -> new BigDecimal(payload.getDataUtf8()))
.retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
.jitter(0.5)
.filter(this::isTransient)
.doBeforeRetry(signal -> System.out.println(
"Retry #" + signal.totalRetries()
+ " for " + accountId)))
.onErrorResume(ex -> {
System.out.println("All retries exhausted, using fallback");
return Mono.just(new BigDecimal("150000.00"));
});
}
private boolean isTransient(Throwable ex) {
// Retry timeouts and connection resets, not validation errors
return ex instanceof java.net.ConnectException
|| ex instanceof java.util.concurrent.TimeoutException;
}
}
Retry #0 for SAV2001
Retry #1 for SAV2001
Retry #2 for SAV2001
All retries exhausted, using fallback
How do you handle errors on the server side?
On the server, centralize error mapping with @MessageExceptionHandler so the client receives structured errors instead of raw stack traces.
package com.bank.rsocket;
import org.springframework.messaging.handler.annotation.MessageExceptionHandler;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.stereotype.Controller;
import reactor.core.publisher.Mono;
import java.math.BigDecimal;
@Controller
public class AccountController {
@MessageMapping("account.balance")
public Mono<BigDecimal> getBalance(String accountId) {
if (accountId == null || accountId.isBlank()) {
return Mono.error(new IllegalArgumentException(
"Account ID is required"));
}
return Mono.just(new BigDecimal("150000.00"));
}
@MessageExceptionHandler
public Mono<String> handleError(Exception ex) {
// Client sees a clean error message, not a stack trace
return Mono.just("{\"error\":\"" + ex.getMessage() + "\"}");
}
}
How does keep-alive detect dead peers?
Both peers exchange KEEPALIVE frames at a configured interval. If no acknowledgment arrives within the max-lifetime window, the connection is torn down and active streams receive a connection error.
package com.bank.rsocket;
import io.rsocket.core.RSocketConnector;
import io.rsocket.transport.netty.client.TcpClientTransport;
import java.time.Duration;
public class KeepAliveConfig {
public static void main(String[] args) {
RSocketConnector.create()
.keepAlive(
Duration.ofSeconds(20), // ping every 20 seconds
Duration.ofSeconds(90)) // declare dead after 90 seconds
.connect(TcpClientTransport.create("localhost", 7000))
.subscribe(rSocket ->
System.out.println("Connected with keep-alive: "
+ "interval=20s, maxLifetime=90s"));
}
}
Keep-alive timeline: 0s → Client sends KEEPALIVE 20s → Client sends KEEPALIVE 40s → Server goes unresponsive 60s → Client sends KEEPALIVE (no ack) 80s → Client sends KEEPALIVE (no ack) 90s → Max lifetime exceeded → connection declared dead
How does session resumption work?
When resume is enabled, the connection negotiates a resume token. If the transport drops, the client reconnects and presents the token, and the protocol replays unacknowledged frames so logical streams continue.
package com.bank.rsocket;
import io.rsocket.core.RSocketConnector;
import io.rsocket.core.Resume;
import io.rsocket.transport.netty.client.TcpClientTransport;
import reactor.util.retry.Retry;
import java.time.Duration;
public class ResumableSessionDemo {
public static void main(String[] args) {
RSocketConnector.create()
.resume(new Resume()
.sessionDuration(Duration.ofMinutes(5))
.retry(Retry.backoff(10, Duration.ofSeconds(1))))
.connect(TcpClientTransport.create("localhost", 7000))
.flatMapMany(rSocket ->
rSocket.requestStream(
io.rsocket.util.DefaultPayload.create("RELIANCE")))
.doOnNext(payload -> System.out.println(
"Price: " + payload.getDataUtf8()))
.subscribe();
// If the TCP connection drops, the client reconnects,
// presents the resume token, and the stream continues
// from where it left off.
}
}
What happens on disconnect without resumption?
Without resumption, a disconnect terminates active streams with an error signal, and the client typically reconnects and re-subscribes.
package com.bank.rsocket;
import io.rsocket.RSocket;
import io.rsocket.util.DefaultPayload;
import reactor.util.retry.Retry;
import java.time.Duration;
public class ReconnectWithoutResume {
public void streamWithReconnect(RSocket rSocket) {
rSocket.requestStream(DefaultPayload.create("RELIANCE"))
.doOnError(err -> System.out.println(
"Stream error: " + err.getMessage()))
.retryWhen(Retry.backoff(5, Duration.ofSeconds(2))
.doBeforeRetry(signal -> System.out.println(
"Reconnecting, attempt #"
+ signal.totalRetries())))
.subscribe(payload -> System.out.println(
"Price: " + payload.getDataUtf8()));
}
}
What about idempotency for money-moving operations?
Retries on transfers must carry a client-generated idempotency key so the server can detect and ignore a duplicate caused by a retry.
package com.bank.rsocket;
import java.math.BigDecimal;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
public class IdempotentTransfer {
private final ConcurrentMap<String, BigDecimal> processedTransfers
= new ConcurrentHashMap<>();
public BigDecimal processTransfer(String idempotencyKey,
String fromAccount,
String toAccount,
BigDecimal amount) {
// If this key was already processed, return the cached result
return processedTransfers.computeIfAbsent(idempotencyKey, key -> {
System.out.println("Processing transfer: " + key);
// Debit fromAccount, credit toAccount
return amount;
});
}
public static void main(String[] args) {
IdempotentTransfer service = new IdempotentTransfer();
String key = UUID.randomUUID().toString();
// First call processes the transfer
service.processTransfer(key, "SAV2001", "SAV2002",
new BigDecimal("50000.00"));
// Retry with same key: server detects duplicate, no double-debit
service.processTransfer(key, "SAV2001", "SAV2002",
new BigDecimal("50000.00"));
}
}
Processing transfer: a1b2c3d4-...
// Second call: no output, duplicate detected
What should a senior engineer know?
- Never use fire-and-forget for money movement because its delivery is not guaranteed.
- Order matters: the circuit breaker usually wraps the retry, so retries happen inside a closed breaker.
- Distinguish retryable errors (timeouts, connection resets) from non-retryable ones (validation errors).
- For request-channel sessions, decide whether to resume the channel or restart from a known checkpoint.
- Set retry limits and backoff with jitter to avoid retry storms across thousands of clients.
Visual Overview
Error Handling Layers: ┌─────────────────────────────────────────────┐ │ Stream Level (Reactor) │ │ retryWhen(backoff) → onErrorResume │ │ ┌─────────┐ error ┌──────────────┐ │ │ │ Request │────────▶│ retryWhen │ │ │ │ Mono │ │ (3x backoff) │ │ │ └─────────┘ └──────┬───────┘ │ │ │ exhausted │ │ ▼ │ │ ┌──────────────┐ │ │ │onErrorResume │ │ │ │(cached value)│ │ │ └──────────────┘ │ └─────────────────────────────────────────────┘ ┌─────────────────────────────────────────────┐ │ Connection Level (Keep-Alive) │ │ KEEPALIVE ──▶ ack ──▶ alive │ │ KEEPALIVE ──▶ no ack ──▶ dead (after max) │ └─────────────────────────────────────────────┘ ┌─────────────────────────────────────────────┐ │ Session Level (Resumption) │ │ Connect ──▶ negotiate token │ │ Drop ──▶ reconnect + present token │ │ ──▶ replay unacked frames │ │ ──▶ streams continue │ └─────────────────────────────────────────────┘
Key Takeaway
RSocket error handling works at two layers. At the stream layer, errors arrive as Reactive Streams error signals, so you handle them with Reactor operators. Use retryWhen with exponential backoff and jitter to recover transient faults, filtered by a predicate so you only retry timeouts and connection resets, never validation errors. When all retries are exhausted, use onErrorResume to degrade gracefully, for example returning a cached balance instead of showing Rajesh an error.
At the connection layer, RSocket uses keep-alive frames to detect dead peers. Both sides ping at a configured interval, and if acknowledgments stop arriving within the max-lifetime window, the connection is declared dead and active streams receive a connection error. The most important part on disconnect is resumption: when resume is enabled, the connection negotiates a token, and if the transport drops, the client reconnects, presents the token, and the protocol replays unacknowledged frames so logical streams continue.
At National Bank of India Rajesh's live price feed survives a brief mobile-network blip because resumption replays unacknowledged frames without the subscriber seeing an error. For transfers, every retry carries a client-generated idempotency key, so if a retry duplicates a request that actually reached the server, the server dedupes it and Rajesh is never double-debited.
Remember one sentence: handle per-stream errors with retryWhen and onErrorResume, detect dead peers with keep-alive, and survive drops with session resumption tokens so streams continue transparently.
---