helidon-kafka-article-09-se.java

Channel<String> fromKafka = Channel.<String>builder()
       .publisherConfig(KafkaConnector.configBuilder()
               .bootstrapServers("localhost:9092")
               .groupId("example-group-1")
               .topic("messaging-test-topic-1")
               .autoOffsetReset(KafkaConfigBuilder.AutoOffsetReset.LATEST)
               .enableAutoCommit(false)
               .keyDeserializer(StringDeserializer.class)
               .valueDeserializer(StringDeserializer.class)
               .build()
       )
       .build();

// Prepare Kafka connector, can be used by any channel
KafkaConnector kafkaConnector = KafkaConnector.create();

Messaging.builder()
       .connector(kafkaConnector)
       .subscriber(fromKafka, ReactiveStreams.<Message<String>>builder()
               //Apply back-pressure, flatMapCompletionStage request 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();