Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
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();
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment