Use async/await where possible

[gala-dan]

In the producer, if we want to wait for a response we currently call Future.get. My understanding is this is a blocking request, thus meaning a thread may be tied up for the duration of the wait.

It would be better to await the future (unless there is an alternative method for the async/await model). This will setup a continuation for when the future is completed, meaning no threads will be tied up. This should improve throughput (how much will depend on how much time is spent waiting for messages to be sent).

Would obviously need to consider how to handle timeouts.

[paulo-reichert]

If you're using KafkaProducer you can specify a callback on send(). However, the producer API will throw an exception if the send to the broker fails for whatever reason, so it is not really necessary to provide the callback or to do Future.get. You should be able to fire and forget and just ensure you have exception handling in case you get a KafkaException.

Edit -> after some further research, the behavior above seems to be guaranteed only for transactional writes or when idempotent writes are enabled. In this case, you should use send(ProducerRecord, Callback) if your code does not use one of those two features.

[Gala-Sortino]

Waiting on #22.

This can, therefore, use a callback method such that it increments the received value on successful callback I could also use some error handling approach if needed on timeouts. But for now, the focus will be on async sending and incrementing of acknowledged messages.

An example for this can be seen on the producer here: https://kafka-python.readthedocs.io/en/master/usage.html