title: Is there @KafkaListener in Reactor/Reactive Kafka?
body: <p>I wish to write Reactive Kafa to listen on new coming message. But I don't know how to do that. Like @KafkaListener in blocking Kafka - it's waiting for new message</p>
  
  <p>The code demo on Spring Boot Webflux and Reactor Kafka:</p>
  
  <pre><code>public class KafkaConsumer {
      private static final String BOOTSTRAP_SERVERS = "localhost:9092";
      private static final String TOPIC = "Kafka_Example";
  
      private final ReceiverOptions&lt;String, String&gt; receiverOptions;
  
      public KafkaConsumer(String bootstrapServers) {
  
          Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
          props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
          props.put(ConsumerConfig.CLIENT_ID_CONFIG, "sample-consumer");
          props.put(ConsumerConfig.GROUP_ID_CONFIG, "sample-group");
          props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
          props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
          props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
          receiverOptions = ReceiverOptions.create(props);
  
      }
  
      public Disposable consumeMessages(String topic) {
  
          ReceiverOptions&lt;String, String&gt; options = receiverOptions.subscription(Collections.singleton(topic))
                  .addAssignListener(partitions -&gt; System.out.println("onPartitionsAssigned " + partitions))
                  .addRevokeListener(partitions -&gt; System.out.println("onPartitionsRevoked " + partitions));
  
          Flux&lt;ReceiverRecord&lt;String, String&gt;&gt; kafkaFlux = KafkaReceiver.create(options).receive();
          return kafkaFlux.subscribe(record -&gt; {
              ReceiverOffset offset = record.receiverOffset();
              System.out.printf("Received message: offset=%d key=%d value=%s\n", offset.offset(), record.key(), record.value());
              offset.acknowledge();
          });
      }
  
      public static void main(String[] args) throws Exception {
          KafkaConsumer consumer = new KafkaConsumer(BOOTSTRAP_SERVERS);
          consumer.consumeMessages(TOPIC);
      }
  }
  </code></pre>
  
  <p>It's running and stop. I wish it's always waiting for new message.</p>