`ProduceSync` may block indefinitely despite context cancellation

[iwinux]

Description

The ProduceSync method in the kafka client can block indefinitely even when the context is cancelled. This occurs because ProduceSync uses a sync.WaitGroup to wait for all records to be processed, but there's no mechanism to break out of wg.Wait() when the context is cancelled.

Steps to Reproduce

1. start a broker locally with default settings
2. env KAFKA_HOST=127.0.0.1:9092 KAFKA_TOPIC=test go run produce.go
3. the client starts producing records every 5s
4. shut down the broker
5. the client hangs at ProduceSync(), even if deadline (1min) already exceeded.

produce.go - client code that produces records repeatedly within deadline

package main

import (
	"context"
	"fmt"
	"os"

	"log"
	"time"

	"github.com/twmb/franz-go/pkg/kgo"
)

func produce(ctx context.Context) error {
	client, _ := kgo.NewClient(
		kgo.SeedBrokers(os.Getenv("KAFKA_HOST")),
		kgo.RecordDeliveryTimeout(10*time.Second),
		kgo.AllowAutoTopicCreation(),
	)
	defer client.Close()

	var sequence int64
	topic := os.Getenv("KAFKA_TOPIC")
	ticker := time.NewTicker(5 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-ticker.C:
			sequence++

			record := &kgo.Record{
				Topic: topic,
				Value: []byte(fmt.Sprintf("%d", sequence)),
			}

			deadline, _ := ctx.Deadline()
			timeLeft := time.Until(deadline).Round(time.Second)
			log.Printf("Producing message with sequence %d to topic %s (%s remaining)", sequence, topic, timeLeft)

			if err := client.ProduceSync(ctx, record).FirstErr(); err != nil {
				return fmt.Errorf("failed to produce message: %w", err)
			}

			log.Printf("Message produced to partition %d at offset %d", record.Partition, record.Offset)
		}
	}
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
	defer cancel()

	if err := produce(ctx); err != nil {
		log.Fatalf("Failed to produce: %v", err)
	}
}

Possible Root Cause(s)

#520 (comment)

So, the client retries sending the same batch over and over again (which will be deduplicated due to the sequence numbers) until the client receives a response -- and the client ignores any end-user context cancellation or retry limit.

Proposed Fix

If the goal is to unblock ProduceSync, either one might work:

1. replace sync.Group with golang.org/x/sync/errgroup.Group
2. wg.Wait() in another goroutine and signal by close(waitCh), meanwhile waitCh can be checked with a select {} block.

However, the comment mentioned data loss and idempotent write, which I don't quite understand.

[tr4pezohedron]

+1. We ran into the same problem using ProduceSync. At some point, Kafka went down, and our requests hung indefinitely until Kafka came back up or we restarted our application pods, even though we expected them to fail by the context timeout.

[twmb]

By default, the producer uses idempotent production. This opts into a bunch of things that allow avoiding duplicates if requests are repeated. However, if a client produces but does not receive a response, problems arise.

If a client produces but does not get a response, the client does not know whether the broker received and processed the request or not. Internally, there is a sequence number that is incremented for every message produced.

If the client assumes the request was received when the broker did not process the request:

• The client advances sequence numbers internally
• Your next produce will use the higher sequence numbers
• The broker will send an error code back OUT_OF_ORDER_SEQUENCE_NUMBER

This error signals data loss occurred. And, in this case, it did: the broker did not actually receive the request.

If the client assumes the broker did not receive the request but the broker did,

• The client does not advance sequence numbers, instead returning to you that producing some set of records failed
• The client keeps the prior sequence number
• The next time the client produces, the client will use sequence numbers that the broker already saw and persisted
• The broker will return OUT_OF_ORDER_SEQUENCE_NUMBER

As above, the error signals that data loss occurred. In this case, the client doesn't really know what happened here. The client doesn't know the state of the partition on the broker. There is no information in a produce response that says the high watermark (end) offset of the partition. It doesn't know if the broker lost some data, then a different client came in a produced to replace the offsets. It doesn't know if only half of the batch that it previously produced remains. All the client knows is that it tried producing at an offset and the broker is saying "that's the wrong offset to produce at".

If the OOOSN error occurs, the fix within the client is to complain loudly -- an ERROR log occurs, and you can use StopOnDataLossDetected to forcefully stop the client -- and then reset sequence numbers to 0 and continue producing as if the client is initialized anew. Not great, not terrible.

Currently, the client will repeat the outstanding request until a response is eventually received. By repeating the request, eventually the broker should recover and reply, and the reply will be for records that were received in order. If the broker did handle the request before dying, the broker will be replying to a duplicate request -- and, since the client is producing idempotently, the broker will just say "everything is good, we handled this!" without duplicating the data server side. If the broker did not handle the request before dying, the broker will then handle the new request and reply "everything is good!".

This issue is pointing out something that is documented on Produce:
"""
Once a record is buffered into a batch, it can be canceled in three ways: canceling the context, the record timing out, or hitting the maximum retries. If any of these conditions are hit and it is currently safe to fail records, all buffered records for the relevant partition are failed. Only the first record's context in a batch is considered when determining whether the batch should be canceled. A record is not safe to fail if the client is idempotently producing and a request has been sent; in this case, the client cannot know if the broker actually processed the request (if so, then removing the records from the client will create errors the next time you produce).
"""

ProduceSync refers you to Produce for the exact semantics of what happens when producing.

If the client allowed you to cancel what you produced while a request has an outstanding response, you will inadvertently trigger one of the two cases above. You'll cancel the request either before the broker receives it or before the broker replies, but regardless, you are trying to cancel records that were sent to the broker.

It's easy to reason about wanting to cancel records, after all, they aren't produced, right? In the latter case above, they actually were produced. Maybe it's ok to say -- who cares, I'll reproduce them or maybe I don't care about them anymore even if they were produced and it doesn't matter. I don't really want to make that assumption on behalf of the user, though, so I take the paranoid stance that whatever I tell you will be accurate -- even if it means I have to disobey a context cancellation.

Perhaps a compromise can be to introduce a new client option,

// AllowOutstandingProduceCancellations switches the client to obey context cancellations
// even on records that currently have an unresolved produce request to a broker.
//
// (more docs)
func AllowOutstandingProduceCancellations()

Is this something you would use?

Separately -- @tr4pezohedron the client should have recovered once the broker came up, and the records should have eventually been produced. If that's not the case, it's a bug. Please let me know a reproducer.

[iwinux]

Thanks for the detailed explanation!

I'm unfamiliar with the idempotent production mechanism. The doc of DisableIdempotentWrite() says:

1. opts out of Kafka server-side deduplication in the face of reissued requests due to transient network problems.
2. upper-bounds the number of in-flight produce requests per broker to 1, vs. the default of 5 when using idempotency.

For my use cases, opting out of server-side deduplication is acceptable. But I'm not sure about the 1 vs 5 upper-bound. Does it mainly affect the produce throughput of each client?

What I'm thinking: if the producer no longer needs to maintain sequence number (by disabling idempotent production), can it handle context cancellation as expected?

[twmb]

But I'm not sure about the 1 vs 5 upper-bound. Does it mainly affect the produce throughput of each client?

Yes, but take a look at MaxProduceRequestsInflightPerBroker.

What I'm thinking: if the producer no longer needs to maintain sequence number (by disabling idempotent production), can it handle context cancellation as expected?

I've added this in #949.

[twmb]

Merged the PR; I hope to get 1.19 out in the next week or two. KIP-714 is the last big feature remaining.