Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
Channel<String> fromKafka = Channel.create("from-kafka");
KafkaConnector kafkaConnector = KafkaConnector.create();
Messaging messaging = Messaging.builder()
.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)
//Acknowledge message has been consumed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment