title: Reactive Kafka Receiver and paraller records processing flow
body: <p>I'm trying to process batches of records from Kafka, using reactor-kafka like this:</p>
  <pre><code> KafkaReceiver.create(
          kafkaProperties.createReceiverOptions&lt;String, String&gt;(String::class.java).subscription(listOf(&quot;test_topic1&quot;))
      )
          .receiveExactlyOnce(kafkaProducer.transactionManager())
          .concatMap {
              it.flatMap { record -&gt;
                  log.info(&quot;start handling $record&quot;)
                  Thread.sleep(200)
                  log.info(&quot;finish handling $record&quot;)
                  Mono.just(&quot;OK&quot;)
              }
                  .then(kafkaProducer.transactionManager().commit&lt;String&gt;())
          }
          .subscribe()
  </code></pre>
  <p>With max.poll.records = 5. And it works, I can see in logs:</p>
  <pre><code>13:46:50.303 [reactive-kafka-defaultConsumerGroup-1] DEBUG r.k.r.internals.ConsumerEventLoop - Emitting 5 records, requested now 1
  13:46:50.310 [reactive-kafka-defaultConsumerGroup-1] DEBUG r.k.r.internals.ConsumerEventLoop - Paused - back pressure
  ...
  13:46:51.517 [b851635e-caec-46bf-a195-0b05e18158b6-1] INFO  r.n.r.k.r.KafkaReceiveOnceFlowTest - start handling ConsumerRecord
  13:46:51.727 [b851635e-caec-46bf-a195-0b05e18158b6-1] INFO  r.n.r.k.r.KafkaReceiveOnceFlowTest - finish handling ConsumerRecord
  ... ( start - finish 4 more times )
  13:46:52.817 [reactive-kafka-defaultConsumerGroup-1] DEBUG r.k.r.internals.ConsumerEventLoop - Resumed
  13:46:52.818 [reactive-kafka-defaultConsumerGroup-1] DEBUG r.k.r.internals.ConsumerEventLoop - Emitting 5 records, requested now 1
  13:46:52.818 [reactive-kafka-defaultConsumerGroup-1] DEBUG r.k.r.internals.ConsumerEventLoop - Paused - back pressure
  13:46:52.818 [b851635e-caec-46bf-a195-0b05e18158b6-1] DEBUG r.k.r.internals.ConsumerEventLoop - onRequest.toAdd 1
  13:46:52.818 [b851635e-caec-46bf-a195-0b05e18158b6-1] INFO  r.n.r.k.r.KafkaReceiveOnceFlowTest - start handling ConsumerRecord
  ... ( etc. )
  </code></pre>
  <p>That is, poll is executed after each batch is processed</p>
  <p>The problem occures, when I try to process records in parallel, adding <strong>publishOn</strong> to code above:</p>
  <pre><code>...
  .receiveExactlyOnce(kafkaProducer.transactionManager())
  .concatMap {
      it.publishOn(Schedulers.boundedElastic())
        .flatMap { record -&gt;
              log.info(&quot;start handling $record&quot;)
              Thread.sleep(200)
              log.info(&quot;finish handling $record&quot;)
              Mono.just(&quot;OK&quot;)
          }
              .then(kafkaProducer.transactionManager().commit&lt;String&gt;())
      }
  ...
  </code></pre>
  <p>It looks like poll is executed independently in this case:</p>
  <pre><code>....
  14:03:44.809 [reactive-kafka-defaultConsumerGroup-1] DEBUG r.k.r.internals.ConsumerEventLoop - Emitting 5 records, requested now 1
  14:03:44.816 [reactive-kafka-defaultConsumerGroup-1] DEBUG r.k.r.internals.ConsumerEventLoop - Paused - back pressure
  14:03:46.072 [boundedElastic-1] INFO  r.n.r.k.r.KafkaReceiveOnceFlowTest - start handling ConsumerRecord(topic = test_topic1, partition = 0, leaderEpoch = 0, offset = 0, CreateTime = 1643627020072, serialized key size = 17, serialized value size = 19, headers = RecordHeaders(headers = [], isReadOnly = false), key = key-1643627020072, value = [B@37211011)
  14:03:46.099 [kafka-producer-network-thread | producer-755eb08a-4bd3-4914-b22d-cd39f15c5e5e] INFO  o.a.k.c.p.i.TransactionManager - [Producer clientId=producer-755eb08a-4bd3-4914-b22d-cd39f15c5e5e, transactionalId=755eb08a-4bd3-4914-b22d-cd39f15c5e5e] Discovered group coordinator localhost:9092 (id: 1001 rack: null)
  14:03:46.350 [reactive-kafka-defaultConsumerGroup-1] DEBUG r.k.r.internals.ConsumerEventLoop - Resumed
  14:03:46.351 [reactive-kafka-defaultConsumerGroup-1] DEBUG r.k.r.internals.ConsumerEventLoop - Emitting 5 records, requested now 1
  14:03:46.351 [reactive-kafka-defaultConsumerGroup-1] DEBUG r.k.r.internals.ConsumerEventLoop - Paused - back pressure
  14:03:46.352 [755eb08a-4bd3-4914-b22d-cd39f15c5e5e-1] DEBUG r.k.r.internals.ConsumerEventLoop - onRequest.toAdd 1
  14:03:46.462 [reactive-kafka-defaultConsumerGroup-1] DEBUG r.k.r.internals.ConsumerEventLoop - Resumed
  14:03:46.463 [reactive-kafka-defaultConsumerGroup-1] DEBUG r.k.r.internals.ConsumerEventLoop - Emitting 5 records, requested now 1
  14:03:46.463 [755eb08a-4bd3-4914-b22d-cd39f15c5e5e-1] DEBUG r.k.r.internals.ConsumerEventLoop - onRequest.toAdd 1
  14:03:46.463 [reactive-kafka-defaultConsumerGroup-1] DEBUG r.k.r.internals.ConsumerEventLoop - Paused - back pressure
  14:03:46.570 [reactive-kafka-defaultConsumerGroup-1] DEBUG r.k.r.internals.ConsumerEventLoop - Resumed
  14:03:46.585 [boundedElastic-1] INFO  r.n.r.k.r.KafkaReceiveOnceFlowTest - finish handling ConsumerRecord
  14:03:46.585 [boundedElastic-1] INFO  r.n.r.k.r.KafkaReceiveOnceFlowTest - start handling ConsumerRecord
  ....
  </code></pre>
  <p>As you can see, records are polled before batch is handled.
  Desirable behaviour is to poll Kafka for next batch only after current batch is fully processed, but with batch records handled in parallel. How can I achieve this?</p>
  <p>Gratefull for any help or tip</p>
  <p><strong>UPD.</strong>
  After a severe struggle, I think I found a solution:</p>
  <pre><code>....
      .receiveExactlyOnce(kafkaProducer.transactionManager())
      .concatMap ({
              it.collectList()
                  .flatMap  { records -&gt;
                      Flux.fromIterable(records)
                          .parallel()
                          .runOn(Schedulers.boundedElastic())
                          .flatMap { record -&gt;
                                  log.info(&quot;start handling $record&quot;)
                                  Thread.sleep(1000)
                                  log.info(&quot;finish handling $record&quot;)
                                  Mono.just(&quot;OK&quot;)
                          }
                          .sequential()
                          .then()
                  }
                  .then(kafkaProducer.transactionManager().commit&lt;String&gt;())
          }, 0)
  ....
  </code></pre>
  <p>This works perfectly
  But now there is next problem - if I transform <strong>receiveExactlyOnce</strong> to hot publisher (i.e. add <strong>.publish(1)</strong>), similar error occurs, onNext and batch processing start working independently</p>
  <p><strong>UPD 2</strong>.
  There is no problem with <strong>publish</strong>, if you use prefetch = 1. But you really get some &quot;poll forward&quot; behaviour(which is supposed by term &quot;prefetch&quot;, I think), but polling rate is effectively held by batch processing speed.</p>
  <p>Problem solved.</p>