fix(consumer): remove old nsqd connections if addresses change

[jrkt]

Right now, in the event that nsqd addresses change, the old connections stay around in the consumer's connections map. This simply cleans up old connections from that map.

[jrkt]

@mreiferson could I get some thoughts on this?

[ploxiln]

it looks like these comment change are from go-1.19 gofmt changes - can you separate them into a different pull request?

[ploxiln]

These successfulNsqdAddrs from a single lookupd. It is expected that the consumer effectively connects to the union of the results from multiple lookupd. Often, all lookupd return the same results, but it is expected that while changes to the set of nsqlookupd propagate there will be inconsistencies, and that's OK because of the "union" behavior. The behavior is also supposed to be resilient to some temporary network disruptions. So we don't really want to remove/disconnect nsqds which are missing from a single lookupd response.

[jrkt]

But the polling lookup loop only happens on the first lookupd address.

go-nsq/consumer.go

Line 383 in b8bb2c5

if numLookupd == 1 {

. We are having issues when the upstream IPs change and simply looking at the code it seemed troublesome that if the IPs change, old connections are not closed.

[ploxiln]

When the first lookupd address is added, this queryLookupd() is called directly, then the lookupdLoop() is spawned to the background. That loop calls this queryLookupd() again periodically, while there usually are multiple lookupd addresses, and one of them is queried at a time by this function in a round-robin fashion.

[ploxiln]

Normally, connections are removed from Consumer.connections in Consumer.onConnClose(), which does various other cleanup tasks, to keep things consistent.

Are you having an issue where the tcp connection just hangs and is never detected to be closed?

[jrkt]

@ploxiln when does Consumer.onConnClose get called? It looks like it only gets called when you close the consumer but what happens to connections that don't point to a valid IP in the event that upstream IPs change?

[ploxiln]

When the connection has some error, including detecting TCP close, it goes through a sequence:
https://github.com/nsqio/go-nsq/blob/master/conn.go#L640
that eventually calls consumerConnDelegate.OnClose():
https://github.com/nsqio/go-nsq/blob/master/conn.go#L725
That calls Consumer.onConnClose():

go-nsq/delegates.go

Line 121 in 0e8d7a7

func (d *consumerConnDelegate) OnClose(c *Conn) { d.r.onConnClose(c) }

In cases where a server really just switches ip addresses abruptly, there is no TCP FIN or RST that can be sent from the original IP to close the connection, so it'll have to time-out eventually, but operating systems may not enable TCP "keep-alives" by default. But if you do enable them, they should cause the zombie connection to be closed in a few minutes.
golang/go@5bd7e9c
https://tldp.org/HOWTO/TCP-Keepalive-HOWTO/usingkeepalive.html