Goroutines left behind after writing message even after closing writer

[fjagugar]

Describe the bug

While debugging a high CPU consumption in my project, I noticed that each time an event is sent to Kafka, the number of goroutines running in my project was increasing.

The code sending the event is:

func (k *Client) Push(ctx context.Context, topicName string, message *kafka.Message) error {
	writer := &kafka.Writer{
		Addr:  kafka.TCP(k.brokers...),
		Topic: topicName,
		Balancer: &kafka.Murmur2Balancer{},
		Transport: &kafka.Transport{
			TLS: k.dialer.TLS,
		},
	}

	defer func() {
		err := writer.Close()
		if err != nil {
			logger.Warnf("Error closing writer for topic %s: %v", topicName, err)
		}
	}()
	return writer.WriteMessages(ctx, *message)
}

After running that, I see in Intellij debug console that the number of goroutines increases:

The more events are sent, the more goroutines keep running.

Checking the code in the transport file I see that a pool of connections is initialized with a refc: 2 here. When calling to the ref() function, it is increasing that value here. When calling the unref() function, it decreases the value and if it's 0 it properly closes the context of that goroutine. The unref() function is properly called when closing the transport connections.

Adding breakpoints I see that, when writing an event, a pool is created, the code calls to unref(), ref(), unref(), ref(), unref(), leaving the value of refc as 1 but never properly canceling the context and leaving the discover goroutine running forever. On each write operation, there's a new pool created and a new discover goroutine left behind running.

I suspect that refc attribute is being used for tracking if the pool is being used or not. I don't fully understand why the code is using a uintptr for this instead of a int or any other type nor why it is being initialized with 2.

Kafka Version

• Kafka version: 3.7.0 (bitnami/kafka:3.7.0-debian-12-r7)
• kafka-go: v0.4.47

[petedannemann]

While there may be a goroutine leak I suspect the reason why this is actually problematic for you is because you are creating a new writer each time you are writing a message. If you create a single writer and reuse it I imagine this will be a non-issue

[tedwards-bw]

@fjagugar I'm seeing the same thing (I create a new writer for each batch I'm processing) - did you fork and fix it, or did you change your implementation to reduce the number of writers you create? Doesn't seem like a lot of fixes are happening these days.