Hi,
I would like to pick up this issue if no one is working on it yet.

Sure.

I want to share my theory:
"ConcatWith" function needs to get @nonnull Publisher<@nonnull ? extends T> as a parameter but in the example above it gets if-else expression instead. In my opinion, the reason for the StackOverflow Error is that "ConcatWith" take somehow only the following expression :
incr(n) .observeOn(Schedulers.single()).toFlowable().concatMap { next -> numbers(next, max)}
without the base case of the recursion.

one of the comments In the original post on stackOverFlow suggest alternative working implementation:

fun <T> unfold(seed: T, next: (T) -> T?): Flowable<T> =
        UnicastProcessor.create<T>().toSerialized().let { proc ->
            proc
                .startWithItem(seed)
                .doOnNext { prev ->
                    when (val curr = next(prev)) {
                        null ->
                            proc.onComplete()
                        else ->
                            proc.onNext(curr)
                    }
                }
        }

    fun numbers(first: Int, max: Int): Flowable<Int> =
        unfold(first) { prev -> if (prev < max) prev + 1 else null }

    numbers(1, 1_000_000_000)
        .sample(1, SECONDS)
        .blockingForEach(::println)

As we can see the unfold function takes as an argument a function that contains the if-else expression, and this is the right way to do so because the unfold return a flowable object that contains the base case of the recursion.

@akarnokd , What do you think about my theory?

It creates a long chain because of concatMap is kept alive even though it ever emits one item due to incr. I still have to think about what we can do about it or even if we should.