helidon-kafka-article-08-se.java

Channel<String> fromKafka = Channel.create("from-kafka");
KafkaConnector kafkaConnector = KafkaConnector.create();

Messaging messaging = Messaging.builder()
       .connector(kafkaConnector)
       .subscriber(fromKafka, ReactiveStreams.<Message<String>>builder()
               //Apply back-pressure, flatMapCompletionStage requests one by one
               .flatMapCompletionStage(message -> {
                   return CompletableFuture.runAsync(() -> {
                       //Do something lengthy
                       Multi.timer(300, TimeUnit.MILLISECONDS, executor)
                               .first()
                               .await();
                       //Acknowledge message has been consumed
                       message.ack();
                   });
               })
               .ignore())
       .build()
       .start();