FeatureRequest for @RetryableTopics mechanism: Clean up retry-topics with tombstones after message recovery

[fabiobauer]

Hello there 👋

If I got this right, the @RetryableTopic mechanism is built in such a way, that messages are written into retry-topics and are processed again after a predefined backoff. If the retry-process then is successful, the message in the retry-topic is left behind. In order to keep the topic small, we would have to configure a delete-retention.

What do you guys think about a retry-mechanism with automatically created tombstones (in the retry topic) for "recovered" messages? We would then be able to set a log compaction and could clean up the topics more dynamically.

p.s.: Thank you for your work! Working with spring-kafka is awesome!

[garyrussell]

That sounds like a useful feature. Of course, it would only apply to records published with a key.

cc/ @tomazfernandes

[tomazfernandes]

Hi, thanks @fabiobauer for looking into this and bringing this suggestion!

I'm not that familiar with the tombstone records, but it seems a pretty straightforward process.

On a first look, I think there are two complexities we'd need to address to make this work.

The first one is I think we'd need to keep track of all the topics the message has been sent to. For example, in case of a timeout, or a non-retriable exception in the middle of the retry flow, the message could go straight to the DLT, so we shouldn't send tombstone records to all the topics.

The second one is we'd need to define what exactly a successful processing is - if it's the processing not throwing an error, it should be simple enough. But if we take into account whether the message has actually been acknowledged, it might get a bit trickier.

Another concern we might have is about immutability - should something weird happen, it might be more difficult to figure it out if we can't be sure whether there used to be a record there or not, or what the content was. But this could be a decision left to the user, who might opt in for this or not.

Does that sound right?

So I'm wondering if the benefit of this feature would be worth the complexity it would bring to the retry mechanism both code and operation-wise, or if it would introduce new points of failure to something that's currently straightforward, both the retry topics and Kafka's built-in deletion mechanism.

Again, I'm not that familiar with tombstone records, so I may have gotten the wrong angle on this.

WDYT?

[tomazfernandes]

Duh, it just occurred to me that if the message is in the DLT it can't have been a successful processing, and thus the tombstones would not apply - or would we want to consider the message arriving at the DLT a trigger for clearing the previous topics?

Either way, I think we'd need to decide whether to keep track of the topics the message has been sent to, probably via headers, or just assume based on the current topic.

[fabiobauer]

Hey @tomazfernandes, thanks for your feedback!

I'm not that familiar with the Spring-Kafka internals, but wouldn't it be easier here to pause the consumption and wait for a valid connection? I think users would expect a workflow where messages are passed through every retry-topic (on failures only, of course) and end up in the DLT only as a last resort.

IMHO, on a user's perspective, the business logic would be successful when the user code executes completely without an exception or when the user manually acknowledges the messages by passing through the Acknowledgment-object and calling acknowledge() by himself.

To simplify the process, I would suggest that we create tombstones whether the retry-process was successful or not:

• The retry-message is no longer needed when the business logic (user code) applies successfully on the record
• The retry-message is no longer needed, if the business logic fails. The message is then being “transferred” to the next retry topic and the original retry-message can be discarded

I created a simple overview how the message handling with tombstones could work in the future (pretty abstract):

Consumers would be created based on the naming strategies (retryTopicSuffix, topicSuffixingStrategy, dltTopicSuffix).
Can you elaborate more on your Header-Solution?

Just FYI: The DLT Handler implements an DLQ-Merge/Purge pattern of Uber https://eng.uber.com/reliable-reprocessing/.
I don't think this is required in this feature request, this was just put together for a better explanation/presentation.

Debugging this construct might not be that easy. Settings some custom headers in order to give the users the information where message was coming from could make the life easier. 👍

I get were you're coming from and if this feature-request brings us problems it might be wise to skip this one.
What do you think about my suggestions?

[artembilan]

I concur with @fabiobauer .

When we done with the record in the particular step of the retry, we really don't need it in that retry topic any more.
It is not a business information to keep forever like in the original topic we consume from.

[tomazfernandes]

Yup, I was just writing about this, and @fabiobauer’s suggestions address the points I was worried about.

Just for clarification, the timeout I’m referring to is a global timeout the user can set for the retrial process as a whole, not a connection timeout. If the message is unable to be processed after a given time, it will go straight to the DLT upon the next failure, no matter where it is in the retry topics sequence.

The other possibility is if an exception occurs that should not be retried for some reason, then the message would go straight to the DLT as well - that’s covered in the Uber paper, which was part of the basis for this solution.

Back to the issue at hand, if we publish the message to the next topic and then send the tombstone record right away we wouldn’t need to keep track of the topics, so no need to add that to the headers.

Also, considering we have already successfully propagated the message to the next topic, or ended the flow with success, I don’t think we need to worry about the acknowledgment, this should be handled normally by either the user or the framework after a successful execution.

If @garyrussell is also ok with that, is this something you’d like to contribute a PR to, @fabiobauer?

And thanks again for the suggestion!

[garyrussell]

We need to be careful to not overcomplicate this; since compacted topics with tombstones for "deleted" entries is a rather obscure use case; I am surprised that the retryable topic mechanism is being used at all with compacted topics.

Typically, such topics are used as as key/value store (e.g. exchange rates) loaded into memory during initialization and updated at runtime.

It is not clear to me how non-blocking retry makes any sense here because, by definition, strict ordering is important for such use cases.

That said, if there is a valid use case, I concur that the simplest solution would be to immediately write a tombstone whenever a record is read from the topics in the retry hierarchy.

Before making such a change, I'd like to understand the use case.

[fabiobauer]

When using many topics, the storage of the brokers can get quite full after some time. To counteract this, we clean up our retry-topics regularly with the help of delete-retention. This solution is alright.

Technically speaking, we cannot be 100 % sure that the set delete-retention is correct.

• What if the retry-delay of the consumer is pretty high?
• What if the consumer was not running at all because of a bug?

Setting the retention higher will tremendously reduce the probability that we delete an unprocessed record, but we are not for certain if retry-consumer reprocessed the message after all.

This is where the tombstones come in, which “confirm” the reprocessed messages. Cleaning up these kinds of messages (by log compaction) could result in a more clean queue, without deleting unprocessed messages by accident.

We are limited in our resources, so one could argue that this is an edge case.
What do you think about this? @garyrussell @tomazfernandes

Hey @tomazfernandes, I only have sporadically time, so this could take a while! But I could look into it :)

[garyrussell]

That doesn't really make any sense; consider the following records.

key:foo value:val1
...
key:foo value:val2

If you compact that topic, you will lose val1, regardless of whether it has been processed.

With

key:foo value:val1
...
key:foo value:val2
...
key:foo value:null

After compaction, there will be no records for key:foo at all.

[fabiobauer]

Just to not get confused here:
We are using Kafka not for event sourcing. In our use-case, strict ordering is not required because our messages do not depend on each other.

I think we are not talking about the same thing here. In your example, my intention would actually be to delete key:foo when compacting. But other messages before would/could still be there.

[garyrussell]

I understand the second case is ok for what you want, but what about the first case (losing val1)?

If your keys are always unique it will work, but if they are not, this is not the correct use of a compacted topic because you can lose records.

[fabiobauer]

We create records with unique keys only that we then want to delete via tombstones.
In our use case, losing val1 would be no problem since it would contain the same data as val2.

Practically, this would never happen because we write a record with a given key only once.

[tomazfernandes]

Just my 2c, I think for use cases where the non-blocking retries are acceptable, we're already not talking about event sourcing and other architectures where ordering matters.

So for use cases where the key is already unique, this strategy should work fine, and when there is no key, the framework could add the messageId or some other UUID as key to guarantee uniqueness.

The framework's job would be to add the tombstone record marking it for deletion. The documentation should make it clear that keys should not be repeated for use with this strategy due to the risk of losing messages upon compaction.

It does seem like a bit of an edge case, where the user needs to have a more aggressive deletion policy due to resource limitation, or where they can't risk losing an unprocessed record. But I think it makes sense overall.