I’ve run into the following use case several times and would like to know what is the idiomatic way of solving it.
I would like to read a message from Kafka, do some processing which results in an an optional result which needs to be written to a database. After it is successfully written to the database (or the message processing did not deliver a result) the offset needs to be committed to Kafka.
The problem I run into is that all Alpakka connectors provide flows to write records in a database and not optional records so when the processing does not deliver a result the flow to write to the database should be skipped while not changing the order of the messages.
Up until now I solved this by not using an Alpakka connector but instead use an asynchronous database driver to optionally write records to a database which results in a Future, but I would like to use an Alpakka connector for this. Any idea on how to do this? I’ve attempted to use Partition and MergeSorted but was not successful (yet ;)). Is there a better way of doing this?
I hope my question is clear; if it’s not, be sure to ask for additional details ;) Thanks in advance!