danielkec

        <dependency>
            <groupId>io.helidon.microprofile.messaging</groupId>
            <artifactId>helidon-microprofile-messaging</artifactId>
        </dependency>
        <dependency>
            <groupId>io.helidon.messaging.kafka</groupId>
            <artifactId>helidon-messaging-kafka</artifactId>
        </dependency>

danielkec

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()

danielkec

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

danielkec

mp.messaging:

 incoming.from-kafka:
   connector: helidon-kafka
   topic: messaging-test-topic-1
   enable.auto.commit: false
   ack.timeout.millis: 10000
   group.id: example-group-1

danielkec

Channel<String> toKafka = Channel.create("to-kafka");
KafkaConnector kafkaConnector = KafkaConnector.create();

Emitter<String> emitter = Emitter.create(toKafka);

Messaging.builder()
       .emitter(emitter)
       .connector(kafkaConnector)
       .build()
       .start();

danielkec

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

Messaging.builder()
       .connector(kafkaConnector)
       .listener(fromKafka, payload -> {
           System.out.println("Kafka says: " + payload);
       })
       .build()
       .start();

danielkec

public class ExampleBean {

   private final SubmissionPublisher<String> publisher = new SubmissionPublisher<>();

   public void sendMessage(String message) {
       publisher.submit(message);
   }

   @Outgoing("to-kafka")
   public Publisher<String> preparePublisher() {

danielkec

@ApplicationScoped
public class ExampleBean {

   @Incoming("from-kafka")
   public void broadcast(String payload) {
       System.out.println("Kafka says: " + payload);
   }

danielkec

<dependency>
   <groupId>io.helidon.messaging</groupId>
   <artifactId>helidon-messaging</artifactId>
</dependency>
<dependency>
   <groupId>io.helidon.messaging.kafka</groupId>
   <artifactId>helidon-messaging-kafka</artifactId>
</dependency>

danielkec

mp.messaging:

 incoming.from-kafka:
   connector: helidon-kafka
   topic: messaging-test-topic-1
   auto.offset.reset: latest
   enable.auto.commit: true
   group.id: example-group-1

 outgoing.to-kafka: