Client SeekBroker Connection Leak

[BAN1ce]

Description

Running producer to produce messages. after that reset the net. use lsof will see some closed TCP connection. and they will never disappear until restart the process.

Sarama	Kafka	Go
release/1.45.1

Problem Description

The function LeastLoadedBroker, if the client.brokers is empty, then will use seedBroker

sarama/client.go

Line 770 in d6ca80a

_ = client.seedBrokers[0].Open(client.conf)

to connect the broker by create a new TCP. after that, the created TCP connection will never close. if net reset, the closed TCP will never release. So I found use RefreshBrokers func can fix the issue. but not every time. some closed TCP still can not release. My question is :

1. Is there connection leaked?
2. if so, anyway to fix it except RefreshBrokers func.

[github-actions]

Thank you for taking the time to raise this issue. However, it has not had any activity on it in the past 90 days and will be closed in 30 days if no updates occur.
Please check if the main branch has already resolved the issue since it was raised. If you believe the issue is still valid and you would like input from the maintainers then please comment to ask for it to be reviewed.

[github-actions]

Thank you for taking the time to raise this issue. However, it has not had any activity on it in the past 90 days and will be closed in 30 days if no updates occur.
Please check if the main branch has already resolved the issue since it was raised. If you believe the issue is still valid and you would like input from the maintainers then please comment to ask for it to be reviewed.

[Asphaltt]

Hi team,

I've verified that there are leaked TCP connections, even when the broken brokers are not configured as seed brokers.

I encountered a kernel issue that produced the following logs in dmesg:

[Thu Nov 13 16:25:05 2025] page_pool_release_retry() stalled pool shutdown 1 inflight 3020 sec
[Thu Nov 13 16:25:05 2025] page_pool_release_retry() stalled pool shutdown 1 inflight 3020 sec
[Thu Nov 13 16:25:05 2025] page_pool_release_retry() stalled pool shutdown 1 inflight 3020 sec

After investigating, I traced the root cause to leaked connections in this library. The issue chain is as follows:

sarama leaked connections
 → TCP sockets remain in TCP_CLOSE state indefinitely
   → Their sk_receive_queues retain a FINACK skb
     → The skb holds a page allocated from page_pool
       → page_pool_release_retry() cannot free the page because
          the skb is still referenced by the TCP stack
         → Thus page_pool_release_retry() stalls indefinitely

Verification

I used the following kernel module to verify the TCP_CLOSE sockets (tested on Linux kernel 6.6.102):

#include <linux/module.h>
#include <linux/pid.h>
#include <linux/sched.h>
#include <linux/fs.h>
#include <linux/file.h>
#include <linux/fdtable.h>
#include <net/sock.h>
#include <net/inet_sock.h>
#include <net/tcp.h>
#include <net/tcp_states.h>
#include <linux/ip.h>
#include <linux/tcp.h>
#include <linux/inet.h>
#include <net/ipv6.h>
#include <net/page_pool/helpers.h>

static pid_t target_pid;

static const char *tcp_state_str(int state)
{
	switch (state) {
	case TCP_ESTABLISHED: return "ESTABLISHED";
	case TCP_SYN_SENT:    return "SYN_SENT";
	case TCP_SYN_RECV:    return "SYN_RECV";
	case TCP_FIN_WAIT1:   return "FIN_WAIT1";
	case TCP_FIN_WAIT2:   return "FIN_WAIT2";
	case TCP_TIME_WAIT:   return "TIME_WAIT";
	case TCP_CLOSE:       return "CLOSE";
	case TCP_CLOSE_WAIT:  return "CLOSE_WAIT";
	case TCP_LAST_ACK:    return "LAST_ACK";
	case TCP_LISTEN:      return "LISTEN";
	case TCP_CLOSING:     return "CLOSING";
	case TCP_NEW_SYN_RECV: return "NEW_SYN_RECV";
	default:              return "UNKNOWN";
	}
}

static void log_pp_recycled_skb(struct sock *sk, unsigned long ino,
				unsigned int qlen, struct sk_buff *skb)
{
	struct inet_sock *inet = inet_sk(sk);
	struct tcp_sock *tp = tcp_sk(sk);
	struct skb_shared_info *shinfo = skb_shinfo(skb);
	char src_ip[64] = {0};
	char dst_ip[64] = {0};
	u16 src_port = 0, dst_port = 0;
	unsigned int i;

	if (sk->sk_family == AF_INET) {
		snprintf(src_ip, sizeof(src_ip), "%pI4", &inet->inet_saddr);
		snprintf(dst_ip, sizeof(dst_ip), "%pI4", &inet->inet_daddr);
		src_port = ntohs(inet->inet_sport);
		dst_port = ntohs(inet->inet_dport);
#if IS_ENABLED(CONFIG_IPV6)
	} else if (sk->sk_family == AF_INET6) {
		snprintf(src_ip, sizeof(src_ip), "%pI6c", &sk->sk_v6_rcv_saddr);
		snprintf(dst_ip, sizeof(dst_ip), "%pI6c", &sk->sk_v6_daddr);
		src_port = ntohs(inet->inet_sport);
		dst_port = ntohs(inet->inet_dport);
#endif
	}

	pr_info("pid %d socket 0x%llx ino=%lu state=%u/%s shutdown=0x%x err=%d(soft=%d) linger2=%d dead=%u: receive_queue(qlen=%u): skb 0x%llx pp_recycle=%d tcp_flags=0x%02x cloned=%u head_frag=%u %s:%u->%s:%u len=%u data_len=%u nr_frags=%u\n",
		target_pid, (unsigned long long)sk, ino, sk->sk_state, tcp_state_str(sk->sk_state),
		sk->sk_shutdown, sk->sk_err, sk->sk_err_soft, tp->linger2, sock_flag(sk, SOCK_DEAD),
		qlen, (unsigned long long)skb, skb->pp_recycle,
		TCP_SKB_CB(skb)->tcp_flags,
		skb->cloned, skb->head_frag,
		src_ip, src_port, dst_ip, dst_port, skb->len, skb->data_len, shinfo->nr_frags);

	/* Log linear buffer head page if present */
	if (skb->head) {
		struct page *head_page = virt_to_head_page(skb->head);
		dma_addr_t head_dma = page_pool_get_dma_addr(head_page);
		pr_info("  pid %d head: page=0x%llx pfn=%lu data_off=%u headlen=%u dma=0x%llx\n",
			target_pid, (unsigned long long)head_page, page_to_pfn(head_page),
			(unsigned int)(skb->data - skb->head), skb_headlen(skb),
			(unsigned long long)head_dma);
	}

	for (i = 0; i < shinfo->nr_frags; i++) {
		const skb_frag_t *frag = &shinfo->frags[i];
		struct page *page = skb_frag_page(frag);
		dma_addr_t dma = page_pool_get_dma_addr(page);

		pr_info("  pid %d frag[%u]: page=0x%llx pfn=%lu off=%u size=%u dma=0x%llx\n",
			target_pid, i, (unsigned long long)page, page_to_pfn(page),
			skb_frag_off(frag), skb_frag_size(frag),
			(unsigned long long)dma);
	}
}

static void inspect_socket(struct socket *sock, unsigned long ino)
{
	struct sock *sk = sock->sk;
	struct sk_buff *skb;
	unsigned int qlen;
	int owned_before;

	if (!sk || sock->type != SOCK_STREAM || sk->sk_protocol != IPPROTO_TCP || sk->sk_state != TCP_CLOSE)
		return;

	owned_before = sock_owned_by_user(sk);
	lock_sock(sk);

	/* Log if we had to wait for lock */
	if (owned_before)
		pr_info("pid %d socket %p ino=%lu: WAITED for lock (was owned=%d)\n",
			target_pid, sk, ino, owned_before);

	qlen = skb_queue_len(&sk->sk_receive_queue);

	/* Check if there's backlog that will be processed */
	if (sk->sk_backlog.tail) {
		pr_info("pid %d socket %p ino=%lu: HAS BACKLOG (tail=0x%llx)\n",
			target_pid, sk, ino, (unsigned long long)sk->sk_backlog.tail);
	}

	skb_queue_walk(&sk->sk_receive_queue, skb) {
		log_pp_recycled_skb(sk, ino, qlen, skb);
	}
	release_sock(sk);
}

static bool inspect_fd_entry(struct files_struct *files, unsigned int fd)
{
	struct fdtable *fdt;
	struct file __rcu **fdentry;
	struct file *file;
	struct socket *sock;
	unsigned long ino = 0;

	rcu_read_lock();
	fdt = files_fdtable(files);
	if (fd >= fdt->max_fds) {
		rcu_read_unlock();
		return false;
	}

	fdentry = fdt->fd + array_index_nospec(fd, fdt->max_fds);
	file = rcu_dereference_raw(*fdentry);
	if (!file || !get_file_rcu(file)) {
		rcu_read_unlock();
		return true;
	}
	if (unlikely(files_fdtable(files) != fdt) ||
	    unlikely(rcu_dereference_raw(*fdentry) != file)) {
		fput(file);
		rcu_read_unlock();
		return true;
	}
	rcu_read_unlock();

	if (file_inode(file))
		ino = file_inode(file)->i_ino;

	sock = sock_from_file(file);
	if (sock)
		inspect_socket(sock, ino);
	fput(file);

	return true;
}

static void walk_task_sockets(struct files_struct *files)
{
	unsigned int fd;

	for (fd = 0; inspect_fd_entry(files, fd); fd++)
		;
}

static int __init qinspect_init(void)
{
	struct pid *pid;
	struct task_struct *task;
	struct files_struct *files = NULL;
	int ret = 0;

	pid = find_get_pid(target_pid);
	if (!pid)
		return -ESRCH;

	task = get_pid_task(pid, PIDTYPE_PID);
	put_pid(pid);
	if (!task)
		return -ESRCH;

	task_lock(task);
	files = task->files;
	if (files)
		atomic_inc(&files->count);
	task_unlock(task);

	if (!files) {
		ret = -ENOENT;
		goto out_put_task;
	}

	walk_task_sockets(files);

	atomic_dec(&files->count);

out_put_task:
	put_task_struct(task);
	return ret;
}

static void __exit qinspect_exit(void) { }

module_param(target_pid, int, 0444);
module_init(qinspect_init);
module_exit(qinspect_exit);
MODULE_LICENSE("GPL");
MODULE_AUTHOR("Leon Hwang <leon.hwang@linux.dev>");
MODULE_DESCRIPTION("TCP Socket sk_receive_queue Inspection Module");

Makefile

KDIR ?= /lib/modules/$(shell uname -r)/build

KMOD_NAME := tcp_rxq
KMOD := $(KMOD_NAME).ko

obj-m := $(KMOD_NAME).o
$(KMOD_NAME)-objs := $(KMOD_NAME).kmod.o

.PHONY: all clean

all:
	$(MAKE) -C $(KDIR) M=$(CURDIR) modules

clean:
	$(MAKE) -C $(KDIR) M=$(CURDIR) clean

run:
	if [ -z $(PID) ]; then exit 1; fi
	insmod $(KMOD) target_pid=$(PID)
	rmmod $(KMOD_NAME)
	dmesg | grep $(PID)

Results

This helped me identify the leaked connections in TCP_CLOSE state using make run PID=$(pidof xxx-agent):

[102694.540568] pid 286275 socket 0xff110002e0c53600 ino=4741242 state=7/CLOSE shutdown=0x3 err=32(soft=0) linger2=0 dead=0: receive_queue(qlen=1): skb 0xff110002b6aa6100 pp_recycle=1 tcp_flags=0x11 cloned=0 head_frag=1 10.x.x.x:43542->10.y.y.y:9093 len=0 data_len=0 nr_frags=0
[102694.540576]   pid 286275 head: page=0xffd400000f7bd040 pfn=4058945 data_off=118 headlen=0 dma=0x3def41000
[102694.540599] pid 286275 socket 0xff1100010d5c6c00 ino=4836518 state=7/CLOSE shutdown=0x3 err=32(soft=0) linger2=0 dead=0: receive_queue(qlen=1): skb 0xff11000126213300 pp_recycle=1 tcp_flags=0x11 cloned=0 head_frag=1 10.x.x.x:46590->10.y.y.y:9093 len=0 data_len=0 nr_frags=0
[102694.540602]   pid 286275 head: page=0xffd400000fb088c0 pfn=4112931 data_off=118 headlen=0 dma=0x3ec223000
[102694.540605] pid 286275 socket 0xff110002d565d100 ino=4785632 state=7/CLOSE shutdown=0x3 err=32(soft=0) linger2=0 dead=0: receive_queue(qlen=1): skb 0xff110003fc307100 pp_recycle=1 tcp_flags=0x11 cloned=0 head_frag=1 10.x.x.x:61656->10.y.y.y:9093 len=0 data_len=0 nr_frags=0
[102694.540607]   pid 286275 head: page=0xffd4000009d60240 pfn=2578441 data_off=118 headlen=0 dma=0x275809000
[102694.540609] pid 286275 socket 0xff1100019376da00 ino=4837511 state=7/CLOSE shutdown=0x3 err=32(soft=0) linger2=0 dead=0: receive_queue(qlen=1): skb 0xff110001267b1d00 pp_recycle=1 tcp_flags=0x11 cloned=0 head_frag=1 10.x.x.x:23242->10.y.y.y:9093 len=0 data_len=0 nr_frags=0
[102694.540610]   pid 286275 head: page=0xffd40000048382c0 pfn=1183243 data_off=118 headlen=0 dma=0x120e0b000
[102694.540612] pid 286275 socket 0xff110001fe136c00 ino=4884828 state=7/CLOSE shutdown=0x3 err=32(soft=0) linger2=0 dead=0: receive_queue(qlen=1): skb 0xff110001258af300 pp_recycle=1 tcp_flags=0x11 cloned=0 head_frag=1 10.x.x.x:30476->10.y.y.y:9093 len=0 data_len=0 nr_frags=0
[102694.540614]   pid 286275 head: page=0xffd400000bf6d0c0 pfn=3136323 data_off=118 headlen=0 dma=0x2fdb43000
[102694.540616] pid 286275 socket 0xff11000114d89b00 ino=4911491 state=7/CLOSE shutdown=0x3 err=32(soft=0) linger2=0 dead=0: receive_queue(qlen=1): skb 0xff11000127d16500 pp_recycle=1 tcp_flags=0x11 cloned=0 head_frag=1 10.x.x.x:60284->10.y.y.y:9093 len=0 data_len=0 nr_frags=0
[102694.540617]   pid 286275 head: page=0xffd4000007c029c0 pfn=2031783 data_off=118 headlen=0 dma=0x1f00a7000

Analysis

There are few TCP connections to be closed when fetching metadata from the server:

client/metadata got error from broker 1193 while fetching metadata: write tcp 10.x.x.x:24180->10.y.y.y.y:9093: write: broken pipe

However, connections in TCP_CLOSE state are never freed when those brokers are not selected for metadata fetching.

Proposed Solution

To prevent leaked connections, we should close them when errors occur to the tcp sockets. Please see PR #3384 for the implementation.

[github-actions]

Thank you for taking the time to raise this issue. However, it has not had any activity on it in the past 90 days and will be closed in 30 days if no updates occur.
Please check if the main branch has already resolved the issue since it was raised. If you believe the issue is still valid and you would like input from the maintainers then please comment to ask for it to be reviewed.