Multi.concatMap Javadoc claims concatenate(true) equivalence, but implementation uses concatenate(false)

[julianhowarth]

Context

Multi.concatMap(mapper)'s Javadoc (Mutiny versions 2.4.0, 2.9.0, 3.3.0) says it is equivalent to multi.onItem().transformToMulti(mapper).concatenate(true), but the implementation actually calls concatenate(false). The two are not equivalent - concatenate(true) requests eagerly from the upstream on subscription (MultiFlatMapOp, prefetch), while concatenate(false) requests lazily, one at a time, only as its own downstream asks for more (MultiConcatMapOp, no prefetch). This flag changes whether the upstream ever receives request(n >= 1) in cases where the operator's own downstream demand is (at least momentarily) zero..

Description

io.smallrye.mutiny.Multi.concatMap:

/**
 * ...
 * This method is equivalent to {@code multi.onItem().transformToMulti(mapper).concatenate(true)}.
 * ...
 */
@CheckReturnValue
default <O> Multi<O> concatMap(Function<? super T, ? extends Publisher<? extends O>> mapper) {
    return onItem().transformToMultiAndConcatenate(mapper);
}

MultiOnItem.transformToMultiAndConcatenate:

public <O> Multi<O> transformToMultiAndConcatenate(Function<? super T, ? extends Publisher<? extends O>> mapper) {
    return transformToMulti(mapper).concatenate();
}

MultiFlatten.concatenate():

public Multi<O> concatenate() {
    return concatenate(false);
}

i.e. concatMap(mapper) actually resolves to transformToMulti(mapper).concatenate(false) - the exact opposite of what its own Javadoc says (concatenate(true)).

Additional details

We hit this for real with a Multi.createFrom().publisher(...)-wrapped foreign Publisher (bridging an AWS SDK SdkPublisher via mutiny-zero-flow-adapters) that is lazy/pull-based: it does no work, including emitting completion, until it receives demand. We concatMap over it as part of a larger chain. concatenating() only forwards a downstream's remaining demand to the next stream in the chain. When that remaining demand is exactly exhausted by an earlier stream (e.g. a subscriber that requested exactly enough to cover an initial snapshot, then moves on to check for new deltas), the wrapped publisher never receives request(n >= 1) and never completes - silently stalling the whole chain forever, with no error, no timeout, nothing in logs.

Had concatMap actually behaved as documented (concatenate(true), i.e. MultiFlatMapOp with its upstream prefetch), the operator would have requested from the upstream eagerly on subscription regardless of downstream demand, and we would never have hit this.

Possible resolutions

We don't know which side is "correct" - just that they currently disagree:

1. The implementation is the bug: concatMap should call concatenate(true), matching its own Javadoc and MultiFlatMapOp's eager-prefetch semantics.
2. The Javadoc is the bug (e.g. copy-pasted or stale): concatMap is meant to be the lazy, no-prefetch, one-at-a-time operator (which matches MultiConcatMapOp's actual behavior, and reads consistently with concatenate(boolean)'s own Javadoc distinguishing "prefetch" from "no prefetch, lazy, one at a time").

In the latter case our workaround was to subscribe to the bridged publisher internally with unconditional demand from inside a Multi.createFrom().emitter(...)).

I can provide a reproducer for the behaviour we saw if the implementation is correct but first would like to establish which is intended.

[jponge]

Thanks for the report, I'll get back to you next week