Producer messages stuck flushing even though there weren't any writes.

[ylhan]

Description

Scenario:

1. I have a producer running for 4 days
2. No writes at all
3. Calling flush on this producer (timeout=10 seconds) results in 2 unflushed messages

I'm at wits' end here. I didn't write anything using this producer yet it complains that two messages are not flushed.

I dug into the library code a bit and perhaps this could be an issue with the Len() method? Why does this method add up the lengths of 3 different queues? Why do I have unflushed messages when I did not call write on the producer even once?

How to reproduce

Instantiate a kafka producer with the configuration map below, do not write anything, and flush the producer.

Checklist

Please provide the following information:

• [x] confluent-kafka-go and librdkafka version (LibraryVersion()): v2.3.1-0.20240315214844-f1230c0e9dd4
• [x] Apache Kafka broker version: 3.2.0
• [x] Client configuration: ConfigMap{...}:

  &kafka.ConfigMap{
  "client.id":         "blah",
  "bootstrap.servers": ..., 
  "sasl.mechanism": "SCRAM-SHA-512",
  "security.protocol": "SASL_SSL",
  "sasl.username": ...,
  "sasl.password": ...,
  "acks": "all",
  "enable.idempotence": "true",
  "max.in.flight.requests.per.connection": 5,
  "linger.ms": 100,
  "retries": 5,
  "batch.size": 2000,
  "request.timeout.ms": 2000,
  }

• [x] Operating system: docker container - base image: golang:1.21-bullseye
• [x] Provide client logs (with "debug": ".." as necessary) I wrapped confluent kafka's flush like so:

  func (p *Producer) Close(context.Context) error {
  defer p.writer.Close()
  unflushed := p.writer.Flush(p.flushTimeoutMs)
  if unflushed > 0 {
      p.log.Error("Failed to flush messages", zap.Int("unflushed", unflushed))
      return errors.Errorf("failed to flush %d messages", unflushed)
  }
  return nil
  }

  When the client shutdown, I get the following error:

  failed to flush 2 messages

• [x] Provide broker log excerpts: attached. There are multiple services connected to the broker but service is called market-connector. Nothing very telling in the logs from my review.
• [x] Critical issue: No

extract-2024-05-20T21_40_38.964Z.csv

[milindl]

Hi @ylhan , Is there a goroutine or something that is reading from the producer.Events() channel? That channel needs to be read from continuously, as there might be some administrative/control events on that channel, and until that channel is drained, Flush would complain.

It's necessary to read from this channel while running a producer, even if you choose to set per-message delivery channels.

[ylhan]

Yes we have a goroutine running in the background consuming any result messages asynchronously.

[guotie]

same problem

when I produce message, and recv kafka.ErrQueueFull error, I do flush(), but I found it stucked.

[guotie]

If I produce message as following:

            err = p.Produce(&kafka.Message{
                TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: 0},
                Value:          buf,
            }, delivery)
            if err == nil {
                <-delivery
            }

then, When I do p.Flush(10000), it will sucks.

And if I set delivery to nil, then Flush is Ok.

Why?