flatMap's Second Argument Saved 1.7 Million Requests

· 6 min read · java, reactor, micronaut, r2dbc, backpressure, reactive
The Memory Benchmark · Part 10
Contents

Midway through a code review of the reactive pipeline, staring at this line in NeoService.java:

.flatMap(neo -> saveAsteroidWithApproaches(sessionId, neo))

That’s Flux.flatMap() with one argument. The function. No concurrency limit. Reactor’s default: 256 concurrent inner subscriptions.

The R2DBC connection pool has 20 connections. Netty has 4 worker threads. flatMap will cheerfully create 256 concurrent database save operations on 20 connections and 4 threads, queuing 236 of them in memory. Every one holds a NeoAsteroidRecord entity, a NeoCloseApproach list, and whatever intermediate state Reactor needs for the inner publisher chain.

The Code Review

Eleven memory issues in one reactive codebase. Four categories.

1. Unbounded flatMap

Every flatMap in the codebase uses the single-argument form:

// NeoService.java — fetch NEO feed, save each asteroid
return neoClient.fetchFeed(apiKey, start, end)
    .flatMapMany(feed -> Flux.fromIterable(feed.near_earth_objects().values())
        .flatMap(Flux::fromIterable))
    .flatMap(neo -> saveAsteroidWithApproaches(sessionId, neo))  // concurrency: 256

Three nested flatMap calls. The outer one: 256 concurrent per-day lists. The middle one: 256 concurrent per-asteroid expansions. The inner one: 256 concurrent database saves. Theoretical maximum: 256 x 256 x 256 = 16.8 million concurrent operations. Practical maximum: whatever fills memory first.

The fix is flatMap’s second argument:

.flatMap(neo -> saveAsteroidWithApproaches(sessionId, neo), 4)

Four concurrent saves. Matches the Netty worker thread count. Matches the realistic database throughput. The 252 pending items that were queued in memory are now processed sequentially in batches of 4.

ApodService has the same pattern:

return apodClient.fetchRange(apiKey, start, end)
    .map(response -> toEntity(sessionId, response))
    .flatMap(repository::save)  // concurrency: 256

Same fix: .flatMap(repository::save, 4).

2. Unbounded Queries

NeoController.all():

@Get
public Flux<NeoAsteroidRecord> all() {
    return asteroidRepository.findAll();
}

Full table scan. Every asteroid record in the database, streamed to the client with no limit. After a week of daily NEO collection, that’s ~140 asteroids per day x 7 days = ~1,000 records. After a month: ~4,200. After a year: ~51,000. Each record fetched, serialized to JSON, and pushed through the Netty pipeline.

The findAll() in R2DBC doesn’t truly stream — Micronaut Data’s @R2dbcRepository implementation materializes the full result set before emitting it as a Flux. The “reactive” part is the delivery, not the query.

Fix: paginate with a hard cap.

@Get
public Flux<NeoAsteroidRecord> all(@QueryValue(defaultValue = "0") int page,
                                    @QueryValue(defaultValue = "50") int size) {
    int cappedSize = Math.min(size, 500);
    return asteroidRepository.findAll(Pageable.from(page, cappedSize))
            .map(Page::getContent)
            .flatMapMany(Flux::fromIterable);
}

Or better: use Slice instead of Page. Page runs a SELECT COUNT(*) alongside the query — an extra full table scan that you’re paying for to populate a totalPages field. Slice returns only the rows plus a boolean hasNext.

3. Three collectList() in Parallel

ObservationController.buildSessionDetail():

Mono<List<ApodEntry>> apods = apodRepository.findBySessionId(id).collectList();
Mono<List<NeoAsteroidRecord>> neos = neoRepository.findBySessionId(id).collectList();
Mono<List<IssPositionWithWeather>> issPositions = issRepository.findBySessionId(id)
    .flatMap(pos -> weatherRepository.findByIssPositionId(pos.getId())
        .map(weather -> new IssPositionWithWeather(pos, weather))
        .defaultIfEmpty(new IssPositionWithWeather(pos, null)))
    .collectList();

return Mono.zip(apods, neos, issPositions)
    .map(tuple -> new ObservationSessionDetail(...));

Three collectList() calls. Each materializes an entire query result into a java.util.ArrayList. All three run concurrently via Mono.zip(). If a session has 50 APOD entries, 100 asteroids (each with close approaches), and 10 ISS positions with weather — that’s three lists in memory simultaneously, plus the intermediate Flux buffers.

collectList() is the reactive equivalent of SELECT * INTO MEMORY. Sometimes necessary. But these queries have no size bounds — a session spanning a year would pull thousands of records into three lists.

Fix: add findBySessionId methods with a limit, or validate that session date ranges are bounded (which they are — the collector runs daily). The real defense is .take(500) on each Flux before collectList():

Mono<List<ApodEntry>> apods = apodRepository.findBySessionId(id)
    .take(500)
    .collectList();

4. Fire-and-Forget Subscribe

ScheduledCollectorJob.dailyCollection():

collectorService.collectObservations(yesterday, today, null)
    .subscribe(
        session -> log.info("Scheduled collection completed: session {}", session.getId()),
        error -> log.error("Scheduled collection failed: {}", error.getMessage())
    );

.subscribe() with no timeout. No backpressure. No cancellation token. If the NASA API hangs (the ISS position API is “frequently down” per the project docs), this subscription lives forever. The Mono never completes. The scheduled executor thread holds it. The next scheduled run creates another subscription. After a few hung runs, you have multiple zombie subscriptions holding HTTP connections, R2DBC connections, and memory.

Fix:

collectorService.collectObservations(yesterday, today, null)
    .timeout(Duration.ofMinutes(5))
    .subscribe(
        session -> log.info("Scheduled collection completed: session {}", session.getId()),
        error -> log.error("Scheduled collection failed: {}", error.getMessage())
    );

Also, Mono.when() in ObservationCollectorService.executeCollection() discards values:

return Mono.when(apodCount, neoCount, issMono)
    .then(Mono.defer(() -> { ... }));

Mono.when() waits for all publishers to complete but discards their values. The apodCount and neoCount results are only captured via doOnNext side effects into an AtomicReference. This works, but it’s fragile — if anyone removes the doOnNext, the counts silently become zero. Mono.zip() would preserve the values.

The Counterintuitive Heap

One more finding, from the benchmarks: -Xmx48m performs better than -Xmx64m at 256 MB.

Wait.

Serial GC does stop-the-world collection. The time per collection is proportional to the live object set, not the heap size. But the frequency is inversely proportional to the free space. Smaller heap = more frequent GC = live objects stay younger = each individual collection finishes faster.

At -Xmx64m, Serial GC collects less frequently, objects age into the tenured generation, and mark-sweep-compact of the tenured space is expensive. At -Xmx48m, the young generation fills faster, collections happen before objects tenure, and the copy collector (cheap) handles most of the work.

Less heap, more GC pauses, each pause shorter, lower peak memory. The RSS drops because the committed heap pages are smaller. At 256 MB, every megabyte matters — and 16 MB of heap you’re not using is 16 MB you can’t use for anything else either.

The Punchline

These are real bugs. flatMap with 256 concurrency on a 20-connection pool is wrong. findAll() with no limit is a time bomb. Fire-and-forget .subscribe() with no timeout is a leak. Every one of these should be fixed.

They were fixed. On a separate branch that also used -Xmx48m instead of -Xmx64m. The branch ran the full benchmark: 1.74 million requests, zero errors, JVM at 256 MB.

Then Part 6 happened. AOT was enabled. OpenAPI was removed. Non-heap dropped from 119 MB to 101 MB. The full benchmark ran without any code changes: 1.66 million requests, zero errors.

Two independent paths to the same result. The code fixes solve memory pressure under sustained load — when 200 concurrent users create hundreds of in-flight reactive chains. The AOT fix solves memory pressure at idle — when 18 MB of metaspace classes that could have been resolved at build time sit in the non-heap consuming the margin that Serial GC needs to survive.

Different bugs. Different layers. Both push the app below the 256 MB threshold. Neither knows about the other.

The flatMap concurrency fix is still correct — you don’t want 256 concurrent database operations on 20 connections regardless of container memory. The bounded queries are still correct — findAll() on a growing table will eventually matter. The timeout on scheduled jobs is still correct — zombie subscriptions will accumulate.

But the 18 MB of free metaspace from AOT? That’s the fix that shipped. Zero code changes. One properties file. The reactive pipeline is still technically wrong, still technically unbounded. In a 256 MB container, “technically wrong” and “production failure” are separated by exactly 18 megabytes.

Sources