Introduction

The race condition appears when you send a request and only afterward start waiting for the matching response. If the response arrives quickly, the subject can emit before your subscription is ready, and the result is missed. The reliable fix is to register the response listener before the request is sent, usually with a correlation ID and a per-request reactive handle.

Why the Naive Pattern Races

This style is fragile:

1transport.send(request);
2
3return responses
4    .filter(r -> r.getCorrelationId().equals(request.getCorrelationId()))
5    .firstOrError();

It looks reasonable, but there is a timing window between send and subscribe. If the transport or peer responds during that gap, the item is emitted before the subscription exists.

The faster the system gets, the easier it is to hit that race.

Use a Correlation ID and Register First

A common robust design is:

1. create a request ID
2. register a waiting subject or single for that ID
3. send the request
4. complete the waiting subject when the response arrives

That removes the race because the waiting entry exists before the request leaves your process.

1import io.reactivex.rxjava3.core.Single;
2import io.reactivex.rxjava3.subjects.SingleSubject;
3
4import java.util.Map;
5import java.util.UUID;
6import java.util.concurrent.ConcurrentHashMap;
7import java.util.concurrent.TimeUnit;
8
9public class RequestClient {
10    private final Map<String, SingleSubject<Response>> pending = new ConcurrentHashMap<>();
11
12    public Single<Response> sendAndAwait(Transport transport, Request request) {
13        String correlationId = UUID.randomUUID().toString();
14        request.setCorrelationId(correlationId);
15
16        SingleSubject<Response> reply = SingleSubject.create();
17        pending.put(correlationId, reply);
18
19        transport.send(request);
20
21        return reply
22            .timeout(5, TimeUnit.SECONDS)
23            .doFinally(() -> pending.remove(correlationId));
24    }
25
26    public void onIncomingResponse(Response response) {
27        SingleSubject<Response> reply = pending.get(response.getCorrelationId());
28        if (reply != null) {
29            reply.onSuccess(response);
30        }
31    }
32}

This pattern is often better than trying to “await” a shared subject directly.

Why a Shared Subject Alone Is Not Enough

A shared Subject<Response> can still be useful as an event bus, but by itself it does not solve request-response matching. The race is not just about reactivity. It is about lifecycle and correlation.

If you insist on filtering a shared subject directly, the subscription must be active before the send side effect occurs. That is easy to get wrong and harder to reason about than the pending-map pattern above.

The registry approach makes the order explicit:

• register
• send
• complete

That is the real safety property.

Make the Pending Registry Thread-Safe

The pending response registry must handle concurrent access because one thread may send requests while another receives responses.

That is why the example uses ConcurrentHashMap. The same logic also needs cleanup for:

• timeouts
• transport failures
• late duplicate responses
• application shutdown

Without cleanup, the map becomes a memory leak.

Handle Timeout and Cancellation

A waiting response that never arrives should not live forever.

1return reply
2    .timeout(5, TimeUnit.SECONDS)
3    .doOnError(error -> System.err.println("Request failed: " + error.getMessage()))
4    .doFinally(() -> pending.remove(correlationId));

This guarantees the pending entry is removed whether the request succeeds, times out, or fails for another reason.

If cancellation matters in your application, propagate it through the same cleanup path so abandoned requests do not leave stale entries behind.

When a Shared Subject Still Makes Sense

A shared subject is still useful for distributing incoming responses to the part of the system that performs matching. For example, a receiver loop may deserialize network frames and hand them into an internal pipeline.

But the awaited response should usually be represented by a per-request Single, SingleSubject, CompletableFuture, or similar object keyed by correlation ID. That is much clearer than trying to turn one shared hot stream directly into a safe request-response primitive.

Common Pitfalls

The first pitfall is sending the request before creating the subscription or pending entry. That is the race condition in one sentence.

Another issue is using a shared subject with no correlation ID. Even if you avoid the race, you still cannot reliably match overlapping requests.

Developers also forget to remove timed-out or failed requests from the pending registry, which creates leaks and false matches later.

Finally, do not assume “reactive” automatically means “race-free.” Ordering and ownership still matter.

Summary

• The race happens when the response can arrive before the waiting subscription is registered.
• Register the awaiting object before sending the request.
• Use correlation IDs so concurrent requests can be matched safely.
• Keep pending requests in a thread-safe registry and clean them up on success, timeout, or failure.
• A per-request Single or SingleSubject is usually safer than awaiting a shared subject directly.