Message loss during idempotent Produce when caller Context is canceled

[FZambia]

Dear Travis,

Hi! First of all, we are active users of your library — it works great and is superior compared to other Go Kafka libraries we have experience with. Thanks a lot for your awesome work.

Over the last several months, we have been investigating occasional message loss that occurred in our system during message production.

The top-level symptoms were: we see a message successfully produced with an offset assigned and no error returned, but then this message cannot be found in the Kafka topic. This usually happened when there were issues with Kafka — such as broker restarts, partition leadership changes due to network problems, etc.

It seems we eventually found the issue and have a reproducer for it. The full reproducer, with detailed steps to reproduce, can be found in this repo:

https://github.com/FZambia/message_loss_reproducer

It uses several replicas for partition, puts the system under CPU pressure to get REQUEST_TIMED_OUT with a higher probability. The reproducer automatically verifies message loss by consuming all messages and comparing against what was ACKed. In case loss is detected it reports sth like this:

╔══════════════════════════════════════════════════════════════╗
║  REPRODUCED: MESSAGE LOSS DETECTED                           ║
╠══════════════════════════════════════════════════════════════╣
║  Messages ACKed: 847                                         ║
║  Messages found: 846                                         ║
║  Messages LOST:  1                                           ║
╠══════════════════════════════════════════════════════════════╣
║  Lost messages:                                              ║
║    - a1b2c3d4-message-42 (ACKed at offset 156)               ║
╚══════════════════════════════════════════════════════════════╝

Versions

• Kafka 3.6
• franz-go v1.20.6

The problem

The problem reproduces when all of the following happen:

• Producing with idempotency enabled (default in franz-go)
• Producer call context gets canceled
• Kafka already wrote the message to log, but returned an error (like REQUEST_TIMED_OUT)

What happens step by step (our assumption)

1. Producer sends Message A with (PID=1, epoch=0, seq=0)
2. Kafka writes Message A at offset 0
3. Kafka responds with REQUEST_TIMED_OUT (or similar error)
4. Context is canceled, franz-go treats produce as failed
5. BUT: franz-go does NOT invalidate current idempotency invariant (bug!?)

6. Producer sends Message B with (PID=1, epoch=0, seq=0)  ← same invariant!
7. Kafka sees duplicate (PID, epoch, seq) → treats as retry of Message A
8. Kafka returns success with offset 0 (Message A's offset)
9. franz-go reports Message B successfully produced at offset 0

Result:
┌─────────┬─────────────┬──────────────────┐
│ Offset  │ Actual Data │ What client sees │
├─────────┼─────────────┼──────────────────┤
│ 0       │ Message A   │ Message B (!)    │
└─────────┴─────────────┴──────────────────┘

Message B is LOST - it was never written to Kafka,
but the client received a successful ACK with an offset belonging to another message.

Other observations

We believe there are several places in franz-go where canceled context is processed. For example, injecting RETRIABLE error in Kafka response and non-RETRIABLE - both result into message loss, but the code path in the franz-go seems to be different.

Our current fix

For now we started to use context.WithoutCancel and set kgo.ProduceRequestTimeout to be less than customer's HTTP request timeout (source of context cancellations in our case). Will appreciate if you could validate this workaround 🙏

[twmb]

franz-go/pkg/kgo/sink.go

Lines 846 to 855 in a09f0e7

	// Since we have received a response and we are the first batch, we can
	// at this point re-enable failing from load errors.
	//
	// We do not need the mutex lock on the batch. We already have the
	// recBuf mu (guarding most concurrency). The only place batch fields
	// are accessed & modified without the recBuf mu is when writing a
	// batch, and we only ever use a batch in inflight request at a time
	// (regardless of the partition being canceled or moving to a
	// different sink).
	batch.canFailFromLoadErrs = true

I suspect this needs a bit of extra logic to only only set canFailFromLoadErrs to true if the partition error is nil, or is a definitive hard error. I didn't consider REQUEST_TIMED_OUT - that message itself is ambiguous in that the client cannot know the state of the broker (which is exactly what canFailFromLoadErrs is all about - do not allow failing unless we definitively know the state).

[twmb]

The implementation as-is assumes that if we get a response from the broker, we know the state - it was definitely written, or it was definitely not written, and the switch statement below handles the sequence numbers as needed. canFailFromLoadErrs needs to be set back to false on errors that indicate the state of the write is unknown.

I think #1218 should solve this.

[FZambia]

Hi! Thanks for the response!

What I am worrying about – while the reproducer I provided demonstrates REQUEST_TIMED_OUT, I was assuming that Kafka can theoretically return other errors, not only REQUEST_TIMED_OUT.

For example, NOT_ENOUGH_REPLICAS and NOT_ENOUGH_REPLICAS_AFTER_APPEND seem good candidates? When message was written to log and we then retry with the same invariant for another message - we may hit the same problem? wdyt?

[twmb]

I went through the errors in kerr.go, and (from what I know) no other error is ambiguous as to the state of what actually happened on the broker. More specifically, from what I know of the guts of actually writing in Kafka, no other error has a chance of actually writing the message. All other errors indicate the write operation did not occur.

[FZambia]

It seems incorrect to me from protocol guarantees.. I did not find any good description yet though. But from general understanding – probably only REQUEST_TIMED_OUT happens in practice now, but Kafka should provide an answer which errors must be retried safely. It seems at least for all retriable errors the case with canceled context should be handled in a way analogous to REQUEST_TIMED_OUT. If client will get any other retriable error, then we will have the same situation as we have now with a canceled context?

Also, one more note – we experimented with injecting a non-retriable error into Kafka response (just faking, the reproducer code has this also) – and with a canceled context it leads to message loss also. I guess Kafka may return non-retriable error at any point in this flow – and it should be also handled properly by producer. Or am I missing sth here? UPD: It seems this should be safe – because in case of non-retriable most probably previous message should never appear in log, but just assuming.

[twmb]

should provide an answer which errors must be retried safely

There is a gap here unfortunately. They do say which errors are retryable, but it's a bit nuanced and really you have to kinda just read all the KIPs when implementing (and sometimes the source...) to know what should be done.

It seems at least for all retriable errors the case with canceled context should be handled in a way analogous to REQUEST_TIMED_OUT. If client will get any other retriable error, then we will have the same situation as we have now with a canceled context?

Not so - retryable errors indicate a temporary state on the broker, and retrying will allow the client to succeed. Most retryable errors mean the operation didn't happen.

The one other error I see actually that may be relevant is NOT_ENOUGH_REPLICAS_AFTER_APPEND (specifically after). It's not entirely clear what happens in that case. The conditional in the broker to return this was added originally in 2014 and, besides a rename, has not been touched since.

https://github.com/apache/kafka/blob/718202dbf4b5dde65365051566a2d209da83f248/core/src/main/scala/kafka/cluster/Partition.scala#L976 <- current code
apache/kafka@043190c#diff-16590fdbe6661105c6faa87d68e39a553e66b34ee0a6d0b14a48d02e273bab4fR289 <- where it was introduced
apache/kafka@043190c#diff-2cafdc864bbb8058a9f62021ada7f730f89425b617cf4f25d59ee7a9bed4beadR22 <- doc line indicating retries will cause duplicates
https://issues.apache.org/jira/browse/KAFKA-1555 <- introducing Kafka issue

This is the only error that mentions retries will cause duplicates: https://github.com/search?q=repo%3Aapache%2Fkafka+path%3A%2F%5Eclients%5C%2Fsrc%5C%2Fmain%5C%2Fjava%5C%2Forg%5C%2Fapache%5C%2Fkafka%5C%2Fcommon%5C%2Ferrors%5C%2F%2F+duplicate&type=code
Which, for the purposes of this issue, means that the record was actually written and should be added to the conditional in my PR. I don't think any other error should be added besides request_timed_out and this one.

[twmb]

PR updated.

[FZambia]

How about these scenarios?

Case 1

1. client will get NOT_ENOUGH_REPLICAS_AFTER_APPEND (but message is in leader log). Client will continue retries according to new logic
2. client retries, and gets NOT_ENOUGH_REPLICAS most probably upon this retry (I am assuming it's the same condition as NOT_ENOUGH_REPLICAS_AFTER_APPEND, just before append)

Upon canceled context client will stop here assuming NOT_ENOUGH_REPLICAS is a safe error code to not increment a sequence right? Then on next logical message we will hit the same problem - Kafka will accept as it's the previous one.

Case 2

1. client will get REQUEST_TIMED_OUT (but message is in log). Client will continue retries according to new logic
2. client retries, and gets NOT_ENOUGH_REPLICAS.

It will stop here assuming NOT_ENOUGH_REPLICAS upon canceled context is a safe error code to not increment a sequence? Then on next logical message we will hit the same problem - Kafka will accept as it's the previous one.

[twmb]

This is starting to get a bit contrived but I think I see the point. Once either of the two errors (timeout, after_append) are encountered, the batch needs to enter some unknown state where only a definitive success or a definitive non-retryable error indicates status. If we experience a retryable error, we still don't know the state since the first operation.

Both of your cases are retryable, so then the client would not allow cancellation and the client would loop retrying.

If instead this were non-retryable: for starters I'm assuming that, due to idempotency, if the record were truly actually written, then a non-retryable error would not be encountered: the broker would deduplicate the retry and reply with success. If the record were truly not written, then the broker has a chance to return a non-retryable error.

[FZambia]

Not sure that even getting non-retriable error after going into this path is safe enough - it will be based on the assumption that this is impossible scenario, but looking at non-retriable errors I can imagine that in rare situations, especially during incidents, something can be returned.

Looking at other clients, there is a fatal state in confluent-kafka-go:

https://github.com/confluentinc/confluent-kafka-go/blob/14809f440d628fa815abd8a26e40a9d86dba46cb/examples/idempotent_producer_example/idempotent_producer_example.go#L98

In Java client it's written:

https://kafka.apache.org/11/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html

To take advantage of the idempotent producer, it is imperative to avoid application level re-sends since these cannot be de-duplicated. As such, if an application enables idempotence, it is recommended to leave the retries config unset, as it will be defaulted to Integer.MAX_VALUE. Additionally, if a send(ProducerRecord) returns an error even with infinite retries (for instance if the message expires in the buffer before being sent), then it is recommended to shut down the producer and check the contents of the last produced message to ensure that it is not duplicated. Finally, the producer can only guarantee idempotence for messages sent within a single session.

Maybe it's due to this issue? It seems that once idempotent message produce hit this path with an unknown write during produce only success may be safe to handle. And session should be re-created if not? 🤔

[twmb]

confluent-kafka-go uses librdkafka, that fatal error is only set in a few cases which are not retryable in this client: https://github.com/confluentinc/librdkafka/blob/f383af93b3b2ca975a85119da8258c44a7dbc0bf/src/rdkafka_idempotence.c#L159-L173

I think that whole paragraph from the Java client is about avoiding duplicates, not avoiding accidental de-duplication, because the whole point of idempotence is to avoid the extremely common scenario of accidental duplicates.

I think it's worth trying the poisoned-batch approach, where once an error is encountered from any attempt that makes the state of producing unknown, the batch cannot be canceled. It must be either definitively produced, or definitively failed. For what it's worth, this basically switches it to a state that the client behaves like the Java client: once an ambiguous error is encountered, the batch cannot be canceled, it will retry until we know the final state.