Simplify connection close logic

[cckellogg]

Motivation

Simplify the connection close logic into one function. I thinks this makes the code easier to reason about and maintain.

[cckellogg]

We could also do something like

        var err error
	var host *url.URL
	var rpcResult *RPCResult
	backoff := Backoff{100 * time.Millisecond}
	var timeoutCh <-chan time.Time
	for {
		host, err = c.serviceNameResolver.ResolveHost()
		if err != nil {
			c.log.WithError(err).Errorf("rpc client failed to resolve host")
			return nil, err
		}
		rpcResult, err = c.Request(host, host, requestID, cmdType, message)
		// success we got a response
		if err == nil {
			break
		}

		retryTime := backoff.Next()
		sleepCh := time.After(retryTime)
		if timeoutCh == nil {
			timeoutCh = time.After(c.requestTimeout)
		}
		c.log.Debugf("Retrying request in {%v} with timeout in {%v}", retryTime, c.requestTimeout)
		select {
		case <- sleepCh:
			break
		case <- timeoutCh:
			return rpcResult, err
		}
	}

And we may want to think about adding a context (in another PR) if can we need to cancel this when shutting down producer or consumer on reconnect?

[wolfstudy]

Great Jobs

[wolfstudy]

Hello @cckellogg , In connection start, we first judge whether the TCP connection is established successfully.

if c.connect(){}

If the connection is established successfully but the handshake fails, we will branch else, but here we just change the connection status to connectionClosed without directly closing the TCP connection.

else {
	c.metrics.ConnectionsHandshakeErrors.Inc()
	c.changeState(connectionClosed)
}

But in the Close function, when the connection is in the connectionClosed state, it returns directly and does not perform the TCP close operation

func (c *connection) Close() {
	if c.getState() == connectionClosed {
		return
	}

So here in the reconnect process, it will cause multiple TCP connections to be created locally, resulting in TCP connection resources leakage.

Here for the 515 process, only one pulsar Client object is created. In theory, only one TCP connection should be established here, but during the reconnect process, it was found that multiple TCP connections were created locally, and there was a leak of TCP connection resources.

[wolfstudy]

cc / @freeznet @zymap @merlimat

[wolfstudy]

Hello @cckellogg Here I have confusion. When we change the connection status to connectionClosed, when did we actually close the TCP connection?

[wolfstudy]

For example, in https://github.com/apache/pulsar-client-go/blob/master/pulsar/internal/connection.go#L222-L224, When the handshake fails, we have created a new connection, but we directly changed the status to connectionClosed but did not actually close the connection

[cckellogg]

For example, in https://github.com/apache/pulsar-client-go/blob/master/pulsar/internal/connection.go#L222-L224, When the handshake fails, we have created a new connection, but we directly changed the status to connectionClosed but did not actually close the connection

Changed the code to call Close() instead of setting the state to connectionClosed. This should cleanup and close the connection.

[wolfstudy]

Thanks @cckellogg Great changes LGTM +1