sclasen
commented
10 years ago @jbweeks that looks a bit unexpected, are you sending the StreamFSM.Processed to the connector? Thats what it looks like in your code sample. That would explain the unexpected log messages too.
Each message needs to be acked to the sender (which is one of the StreamFSM actors underneath the connector).
The individual messages dont need to be tracked in your case, but you would need to track the number of messages sent by each stream and then send a corresponding number of processed back.
Something along the lines of this could be the receiver actorRef provided to your consumer props?
class BatcherReceiver extends Actor{
val outstanding = new mutable.HashMap[ActorRef, Int]()
var backend:ActorRef = _
case object AckBatch
override def receive = {
case AckBatch =>
outstanding.foreach{
case (sender, count) =>
(1 to count).foreach(_ => sender ! StreamFSM.Processed)
}
outstanding.clear()
case x:Any =>
backend ! x
val out = outstanding.getOrElseUpdate(sender(), 0)
outstanding += sender() -> (out +1)
}
}
jbweeks
rsarawgi
jfoy
Most of what we do with Akka-Kafka involves many thousands of messages a second, and these messages are sorted into categories, and processed in bulk into various backend sinks.
We started to work with the back-pressure support in akka-kafka and rather than just return with a StreamFSM.Processed acknowledgement immediately, we started to tie these to the success of the backend batch operations.
However, with batch sizes of over around 10 messages (although we typically deal with batches on the backend of size 20-50k), we start to see the behavior get "racy", e.g. occasionally seeing:
The higher the batch size, the more likely we see this error, which eventually leads to consumer failure due to unhandled acknowledgements.
(We have an Akka method that, on success, takes the batch size as an argument and then turns around and does this:)
I tried playing with different settings for maxInFlightPerStream, but that didn't seem to help...
Perhaps this could be solved by adding an Int or Long argument to the StreamFSM.Processed message like I did for the relay above?