Question about zero tolerance for message loss about akka-kafka HOT 3 CLOSED

This is more of a general akka/akka-cluster question, around how to achieve at-least-once delivery rather than the default of akka, which is at-most-once.

There is no free lunch here, so you will have to do some bookkeeping to achieve your goals.

Assuming you are using the batch consumer here, but if not the same principles apply, the implementation may be slightly different.

The first thing you might try is co-locating a 'bookkeeping' actor with the consumer. The bookkeeper would then be the receiver of messages from the consumer. When the bookkeeper receives a message, it would send the batch into your cluster and wait for a response. If the response was not delivered within some timeout, the bookkeeper would re-send, and continue to do so after each timeout encountered.

This is relatively straightforward using the scheduler, something like

val clusterActorRef: ActorRef = ... ///entrypoint to your cluster processing
val consumer:ActorRef = ...          ///the akka-kafka consumer
var cancelTimer:Option[Cancellable] = None

//Bookkeeper Recieve
case batch: Batch =>
  cancelTimer =  Some(context.system.scheduler.schedule(Duration.zero, redeliveryTimeout, clusterActorRef, batch))
case BatchAck =>
  cancelTimer.foreach(_.cancel)
  consumer !  BatchConnectorFSM.BatchProcessed

If this is not 'reliable enough' for you, you should probably investigate using akka-persistence AtLeastOnceDelivery http://doc.akka.io/docs/akka/2.3.9/scala/persistence.html#At-Least-Once_Delivery

you could use that with a local leveldb journal. But in this case I would think that this is 'good enough' since if the consumer/bookkeeper node crashes, on restart you will get any in-flight messages replayed since they were not committed.

Make sense?

from akka-kafka.

Thanks a lot for your detailed explanation.
Appreciate your help

from akka-kafka.

Cheers!

from akka-kafka.