Reactor-core: Passing Flux.error to concatWith results in infinite Flux

[Selenog]

Expected Behavior

Flux should emit an error.

Actual Behavior

Flux "hangs", it becomes an infinite flux.

Steps to Reproduce

@Test
fun repoCase() {
    val flux = Flux.just<Int>(13)
        .concatWith {
            Flux.error<Int>(IllegalStateException())
        }

    StepVerifier
        .create(flux)
        .expectNext(13)
        .expectError(IllegalStateException::class.java)
        .verify(Duration.ofSeconds(1))
}

Possible Solution

As a workaround this does work:

@Test
fun workaround() {
    val flux = Flux.just<Flux<Int>>(Flux.just<Int>(13), Flux.error<Int>(IllegalStateException()))
        .concatMap { it }

    StepVerifier
        .create(flux)
        .expectNext(13)
        .expectError(IllegalStateException::class.java)
        .verify(Duration.ofSeconds(1))
}

Your Environment

• Reactor version(s) used: 3.3.4.RELEASE
• JVM version (javar -version): Kotlin 1.3.50 & Java 11.0.2
• OS and version (eg uname -a): Windows 10

[bsideup]

@Selenog works for me:

    @Test
    public void gh2101() {
        Flux.just(13)
            .concatWith(Flux.error(new IllegalStateException("ooops")))
            .as(StepVerifier::create)
            .expectNext(13)
            .expectError(IllegalStateException.class)
            .verify(Duration.ofSeconds(1));
    }

[Selenog]

Thanks, you helped me figure out what I did wrong.
In Java you have:
concatWith(Publisher<? extends T> other)

Which when used with a lambda would look like:
concatWith(subscriber -> {... do something with subscriber ...})

Kotlin however allows you to do:
concatWith { ... do something with it ... } // it being the subscriber here

So basically I used { } instead of ( ) resulting in a Publisher which never uses it's subscriber.

Closing it as I made a stupid mistake because of Kotlin fanciness.