There's a function in the product that after receive kafka msg, the system need to stream out the message content to third party system. For example, some merchant using this function to stream out their orders to warehouse after the customer paied the order.
However, not all the orders can be streamed out successfully, it might failed due to network issues or authentication issue or some runtime exceptions. The merchant is hoping to have retry function to stream out the failed orders.
DesignAfter received the kafka message, save it with "in-progress" status and stream it out. if it's successful, set the status to "succeed" and "failed" if not. When the user call the retry API, get the failed msg and do re-stream out function.
Concurrent situationThe existing implementation to consume Kafka message already has re-try mechanism :
@Transactional
@KafkaListener(topics = "${xxx}", containerFactory = "retryableKafkaListenerContainerFactory", concurrency = "#{@kafkaProperties.getConcurrencyValueForRetryableConsumers({'order-status-changed'}, ${service.app.instances:2})}")
public void processOrderStatusStreamingMessage(@Payload final Message message) {
...
}
And we also need to prevent the user call the "retry" API multi times, the data can only be streamed out once.
SolutionAfter receive the retry request, get the message with "failed" status with write lock, set the status to "in-progress", then run the stream out function.
If there's concurrent requests, it has to wait. After the first thread release the lock, the status already set to "in-progress", the stream out function won't be called.
Using db write lock is not the best way from performance view, but simple and easy to go. Redis lock should be considerred for better performance.
Using DB write lock, and the query is by the message status, when design the message persistance, should set index for the status so that only lock the row instead of the whole table.
