Minimizing failure without impacting recovery when building processes on top of Kafka

Question

I am working with a microservice that consumes messages from Kafka. It does some processing on the message and then inserts the result in a database. Only then am I acknowledging the message with Kafka.

It is required that I keep data loss to an absolute minimum but recovery rate is quick (avoid reprocessing message because it is expensive).

I realized that if there was to be some kind of failure, like my microservice would crash, my messages would be reprocessed. So I thought to add some kind of 'checkpoint' to my process by writing the state of the transformed message to the file and reading from it after a failure. I thought this would mean that I could move my Kafka commit to an earlier stage, only after writing to the file is successful.

But then, upon further thinking, I realized that if there was to be a failure on the file system, I might not find my files e.g. using a cloud file service might still have a chance of failure even if the marketed rate is that of >99% availability. I might end up in an inconsistent state where I have data in my Kafka topic (which is unaccessible because the Kafka offset has been committed) but I have lost my file on the file system. This made me realize that I should send the Kafka commit at a later stage.

So now, considering the above two design decisions, it feels like there is a tradeoff between not missing data and minimizing time to recover from failure. Am I being unrealistic in my concerns? Is there some design pattern that I can follow to minimize the tradeoffs? How do I reason about this situation? Here I thought that maybe the Saga pattern is appropriate, but am I overcomplicating things?

user_2148953 · Answer

If you are that concerned of data reprocess, you could always follow the paradigm of sending the offsets out of kafka.

For example, in your consumer-worker reading loop: (pseudocode)

while(...)
{
   MessageAndOffset = getMsg();
   //do your things
   saveOffsetInQueueToDB(offset);
}

saveOffsetInQueueToDB is responsible of adding the offset to a Queue/List, or whatever. This operation is only done one the message has been correctly processed.

Periodically, when a certain number of offsets are stored, or when shutdown is captured, you could implement another function that stores the offsets for each topic/partition in:

An external database.
An external SLA backed storing system, such as S3 or Azure Blobs.
Internal (disk) and remote loggers.

If you are concerned about failures, you could use a combination of two of those three options (or even use all three).

Storing these in a "memory buffer" allows the operation to be async, so there's no need for a new transfer/connection to the database/datalake/log for each processed message.

If there's a crash, you could read all messages from the beginning (easiest way is just changing the group.id and setting from beginning) but discarding those whose offset is included in the database, avoiding the reprocess. For example by adding a condition in your loop (yep pseudocode again):

while(...)
{
   MessageAndOffset = getMsg();
   if (offset.notIncluded(offsetListFromDB))
   {
      //do your things
      saveOffsetInQueueToDB(offset);
   }
}

You could implement better performant algorithms instead a "non-included" type one, just storing the last read offsets for each partition in a HashMap and then just checking if the partition that belongs to each consumer is bigger or not than the stored one. For example, partition 0's last offset was 558 and partitions 1's 600:

//offsetMap = {[0,558],[1,600]}

while(...)
{
   MessageAndOffset = getMsg();
   //get partition => 0
   if (offset > offsetMap.get(partition))
   {
      //do your things
      saveOffsetInQueueToDB(offset);
   }
}

This way, you guarantee that only the non-processed messages from each partition will be processed.

Regarding file system failures, that's why Kafka comes as a cluster: Fault tolerance in Kafka is done by copying the partition data to other brokers which are known as replicas.

So if you have 5 brokers, for example, you must experience a total of 5 different system failures at the same time (I guess brokers are in separate hosts) in order to lose any data. Even 4 different brokers could fail at the same time without losing any data.

All brokers save the same amount of data, same partitions. If a filesystem error occurs in one of the brokers, the others will still hold all the information:

Minimizing failure without impacting recovery when building processes on top of Kafka

1 Answers