Skip to content

Instantly share code, notes, and snippets.

@mchmielarz

mchmielarz/committable_kafka_akka.java

Created December 15, 2020 10:26
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save mchmielarz/9198dfaf3fd7bdcc0a23f1064aee344a to your computer and use it in GitHub Desktop.
Save mchmielarz/9198dfaf3fd7bdcc0a23f1064aee344a to your computer and use it in GitHub Desktop.
Download ZIP
Raw
committable_kafka_akka.java
Consumer
.committableSource(consumerSettings, Subscriptions.topics(topicName))
.map(committableMessage -> {
// processing logic comes here
return committableMessage;
})
.mapAsync(maxParallelism, msg -> CompletableFuture.completedFuture(msg.committableOffset()))
.runWith(Committer.sink(CommitterSettings.create(committerSettings)), materializer)
.toCompletableFuture()
.handle(AppSupport.doneHandler())
.join();
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment