Hello, Habr.
Recently, I on the parameters our team most often uses for Kafka Producer and Consumer to achieve guaranteed delivery. In this article, I want to talk about how we organized the reprocessing of an event received from Kafka due to the temporary unavailability of an external system.
Modern applications operate in a very complex environment. Business logic wrapped in a modern tech stack runs in a Docker image managed by an orchestrator like Kubernetes or OpenShift, communicating with other applications or enterprise solutions through a chain of physical and virtual routers. In such an environment, something can always go wrong, so reprocessing events in case one of the external systems is unavailable is an important part of our business processes.
How it was before Kafka
Previously, in the project, we used IBM MQ for asynchronous message delivery. When an error occurred during the service's operation, the received message could be placed in a dead-letter queue (DLQ) for further manual processing. The DLQ was created alongside the incoming queue, and the message was moved within IBM MQ.
If the error was temporary and we could determine this (for example, ResourceAccessException during an HTTP call or MongoTimeoutException during a MongoDb request), then a retry strategy was implemented. Regardless of the branching application logic, the original message was either moved to a system queue for delayed sending or to a separate application that had been created a long time ago for resending messages. The header of the message would include the retry number linked to the delay interval or to the end of the strategy at the application level. If we reached the end of the strategy but the external system was still unavailable, the message would be placed in the DLQ for manual processing.
Finding a solution
, you can find the following . In short, it is suggested to create a topic for each delay interval and implement Consumers on the application side that will read messages with the required delay.
Despite the many positive reviews, it seems somewhat unsuccessful to me. Primarily because, in addition to meeting business requirements, the developer will have to spend a lot of time implementing the described mechanism.
Furthermore, if access control is enabled on the Kafka cluster, it will take some time to create topics and ensure the necessary permissions for them. In addition, the correct retention.ms parameter will need to be selected for each of the retry topics so that messages can be resent without disappearing. Implementation and access requests will have to be repeated for every existing or new service.
Now let's look at the mechanisms for reprocessing messages provided by Spring as a whole and Spring-Kafka in particular. Spring-Kafka has a transitive dependency on Spring-Retry, which provides abstractions for managing various BackOffPolicies. This is quite a flexible tool, but a significant drawback is storing messages for resend in the application's memory. This means that restarting the application due to updates or errors during operation will result in the loss of all messages waiting for reprocessing. Since this point is critical for our system, we decided not to consider it further.
Spring-Kafka itself provides several implementations of ContainerAwareErrorHandler, for example, , which allows you to handle the message later without shifting the offset in the event of an error. Starting from Spring-Kafka version 2.3, it became possible to set a BackOffPolicy.
This approach allows reprocessed messages to survive the application restart, but the DLQ mechanism is still absent. This is the option we chose at the beginning of 2019, optimistically believing that a DLQ would not be needed (we were fortunate, and indeed, it wasn't needed during several months of operating the application with this reprocessing system). Temporary errors resulted in triggering SeekToCurrentErrorHandler. Other errors were logged, causing the offset to shift, and processing continued with the next message.
Final decision
The implementation based on SeekToCurrentErrorHandler led us to develop our own mechanism for resending messages.
First and foremost, we wanted to leverage existing experience and expand upon it based on the application's logic. For applications with linear logic, it would be optimal to pause the reading of new messages for a brief period defined within the retry strategy. For other applications, we wanted a unified point to ensure the execution of the retry strategy. Additionally, this unified point should have DLQ functionality for both approaches.
The retry strategy itself should be stored in the application responsible for determining the next interval when a temporary error occurs.
Stopping the Consumer for an application with linear logic
When working with spring-kafka, the code to stop the Consumer might look something like this:
public void pauseListenerContainer(MessageListenerContainer listenerContainer,
Instant retryAt) {
if (nonNull(retryAt) && listenerContainer.isRunning()) {
listenerContainer.stop();
taskScheduler.schedule(() -> listenerContainer.start(), retryAt);
return;
}
// to DLQ
}In this example, retryAt is the time when the MessageListenerContainer should be restarted if it is still running. The restart will occur in a separate thread launched in the TaskScheduler, which is also provided by spring.
We find the value of retryAt as follows:
- The value of the retry count is sought.
- According to the value of the retry count, the current delay interval in the retry strategy is identified. The strategy is declared within the application itself, and we chose JSON format for its storage.
- The found interval in the JSON array contains the number of seconds after which the processing should be retried. This number of seconds is added to the current time, forming the value for retryAt.
- If the interval is not found, the value of retryAt is null, and the message will be sent to DLQ for manual review.
With this approach, we only need to keep track of the number of retries for each message currently being processed, such as in the application's memory. Storing the attempt counter in memory is not critical for this approach, as the application with linear logic cannot process it as a whole. Unlike spring-retry, restarting the application will not result in the loss of all messages for reprocessing, but simply restart the strategy.
This approach helps reduce the load on an external system, which may be unavailable due to excessive demand. In other words, in addition to reprocessing, we have achieved the implementation of the pattern .
In our case, the error threshold is just 1, and to minimize system downtime due to temporary network failures, we use a very granular retry strategy with short delay intervals. This may not be suitable for all applications of the group, so the ratio between the error threshold and the interval size should be adjusted based on the system's characteristics.
A separate application for processing messages from applications with nondeterministic logic
Here is an example of code that sends a message to such an application (Retryer), which will retry sending to the DESTINATION topic when the RETRY_AT time is reached:
public void retry(ConsumerRecord record, String retryToTopic,
Instant retryAt, String counter, String groupId, Exception e) {
Headers headers = ofNullable(record.headers()).orElse(new RecordHeaders());
List arrayOfHeaders =
new ArrayList(Arrays.asList(headers.toArray()));
updateHeader(arrayOfHeaders, GROUP_ID, groupId::getBytes);
updateHeader(arrayOfHeaders, DESTINATION, retryToTopic::getBytes);
updateHeader(arrayOfHeaders, ORIGINAL_PARTITION,
() -> Integer.toString(record.partition()).getBytes());
if (nonNull(retryAt)) {
updateHeader(arrayOfHeaders, COUNTER, counter::getBytes);
updateHeader(arrayOfHeaders, SEND_TO, "retry"::getBytes);
updateHeader(arrayOfHeaders, RETRY_AT, retryAt.toString()::getBytes);
} else {
updateHeader(arrayOfHeaders, REASON,
ExceptionUtils.getStackTrace(e)::getBytes);
updateHeader(arrayOfHeaders, SEND_TO, "backout"::getBytes);
}
ProducerRecord messageToSend =
new ProducerRecord(retryTopic, null, null, record.key(), record.value(), arrayOfHeaders);
kafkaTemplate.send(messageToSend);
} The example shows that a lot of information is transmitted in the headers. The value of RETRY_AT is found just like in the retry mechanism via stopping the Consumer. In addition to DESTINATION and RETRY_AT, we pass:
- GROUP_ID, which we use to group messages for manual analysis and simplify searching.
- ORIGINAL_PARTITION, to attempt to retain the same Consumer for reprocessing. This parameter can be null, in which case a new partition will be obtained based on the record.key() of the original message.
- The updated value of COUNTER, to follow the strategy of retries.
- SEND_TO is a constant indicating whether to send the message for reprocessing upon reaching RETRY_AT or to place it in the DLQ.
- REASON is the reason why the message processing was interrupted.
The Retryer saves messages for resending and manual parsing in PostgreSQL. A timer triggers a job that finds messages with an elapsed RETRY_AT and sends them back to the ORIGINAL_PARTITION of the DESTINATION topic with the record.key().
After sending, messages are removed from PostgreSQL. Manual parsing of messages occurs in a simple UI that interacts with the Retryer via REST API. Its main features include resending or deleting messages from the DLQ, viewing error information, and searching messages, for example by error name.
Since access management is enabled on our clusters, it is necessary to additionally request access to the topic that the Retryer listens to, and allow the Retryer to write to the DESTINATION topic. This is inconvenient, but unlike the interval topic approach, we have a full-fledged DLQ and UI to manage it.
There are cases where multiple different consumer groups read from the incoming topic, with applications implementing different logic. Retrying the message through the Retryer for one of these applications will lead to a duplicate on another. To protect against this, we create a separate topic for reprocessing. The incoming and retry topics can be read by the same Consumer without any restrictions.
By default, this approach does not provide a circuit breaker capability, but it can be added to the application using or the new , wrapping the call sites to external services in the appropriate abstractions. Additionally, there is an option to choose a strategy for pattern, which can also be useful. For instance, in spring-cloud-netflix, this could be a thread pool or a semaphore.
Output
As a result, we have created a separate application that allows for the re-processing of messages when any external system is temporarily unavailable.
One of the main advantages of the application is that external systems operating on the same Kafka cluster can use it without significant modifications on their side! This application will only need access to the retry topic, fill in a few Kafka headers, and send a message to the Retryer. No additional infrastructure needs to be set up. To reduce the number of messages being transferred from the application to the Retryer and back, we focused applications with linear logic and implemented reprocessing through stopping the Consumer.
Source: habr.com
