Reactor Kafka - Parallel consumption from multiple topics?

TL;DR; Do I need a per-topic KafkaReceiver instance if I want to process the messages from separate topics in parallel and the message processing time from different topics is different?

I am experimenting with 1 KafkaReceiver subscribed to 5 topics, each having 1 partition.

It can happen that processing of the messages coming from a particular topic slows down. To avoid pausing the consumer, since processing of the messages from other topics works fine, I group messages per topic and do processing on separate threads. It looks something like this:

...
scheduler = Schedulers.newParallel("P", 5);
...
flux.log()
    .groupBy(m -> m.receiverOffset().topicPartition())
    .flatMap(partitionFlux -> partitionFlux.publishOn(scheduler)
        .map(this::transform)
        .map(this::process)
    .doOnError(throwable -> {
        log.error("Ups!", throwable);
    })
    .subscribe();

What I would like to prevent is filling up the memory by constantly consuming messages and I am not sure when the KafkaConsumer#poll is triggered.

Since I am new to reactor-kafka I am guessing that poll happens when request(NNN)... is logged and it seems that the thread from the scheduler pool is invoking it:

[            P-4] reactor.Flux.UsingWhen.2                 : request(256)
[            P-4] reactor.Flux.UsingWhen.2                 : onNext(ConsumerRecord...
[            P-4] reactor.Flux.UsingWhen.2                 : onNext(ConsumerRecord...
[            P-4] reactor.Flux.UsingWhen.2                 : onNext(ConsumerRecord...

I assume here that the processing thread(P-4) is saying "I can process more!!!".

Sometimes I get:

...
[r-coordinator-3] reactor.Flux.UsingWhen.2                 : onNext(ConsumerRecord(...
[r-coordinator-3] reactor.Flux.UsingWhen.2                 : onNext(ConsumerRecord(...
[r-coordinator-3] reactor.Flux.UsingWhen.2                 : onNext(ConsumerRecord(...
...

where r-coordinator-3 is the thread that use to do processing before .publishOn was added to the pipeline.

1. Does poll happen when request(NNN)... is logged?
2. Why is it the case that sometimes P-4 and sometimes r-coordinator-3 is logging the onNext event?
3. If poll is invoked when the processing thread, e.g. P-4, is ready to further process messages, will poll fetch messages from all the topics and eventually cause out of memory error? In "1 KafkaReceiver - 5 topics" example, thread P-4 might be processing quickly and it might be frequently calling poll while other threads would fill up their queues if poll fetches from all the topics and they are slow.
4. Is my only option in this use case to use KafkaReceiver per topic?