[fix] [client] fix same producer/consumer use more than one connection per broker

[shibd]

Motivation

This is a catch up for apache/pulsar#21144

When a producer or consumer reconnects, a random number will be generated as the key suffix in ConnectionPool to create or get the Connection object from the pool.

pulsar-client-go/pulsar/internal/connection_pool.go

Lines 159 to 160 in ffba2a8

	idx := cnt % p.maxConnectionsPerHost
	return fmt.Sprint(logicalAddr.Host, "-", physicalAddr.Host, "-", idx)

Modifications

• Change GetConnection() of connection_pool and add params: keySuffix
• Generate once keySuffix for consumer/producer when create it.
• When reconnecting, use this keySuffix to ensure obtaining the same connection

Verifying this change

• Add TestSelectConnectionForSameConsumer and TestSelectConnectionForSameProducer to cover this.

[BewareMyPower]

I see the ACK requests still use pc._getConn() and RequestOnCnx or RequestOnCnxNoWait in some places, should they use RequestWithCnxKeySuffix as well?

Or at least should we call _setConn(nil) when the consumer unregisters itself from the connection? For example, https://github.com/apache/pulsar/blob/5a3a1f169a7f90181bd5c213c8e9f479bc74f0f2/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java#L194

[shibd]

I see the ACK requests still use pc._getConn() and RequestOnCnx or RequestOnCnxNoWait in some places, should they use RequestWithCnxKeySuffix as well?

No need, in grabConn will reset conx

[RobertIndie]

cnt = -cnt is incorrect. I think you're trying to handle the case where uint32 overflow results in a negative number. However, if cnt overflows to math.MinInt32, negating it will still result in a negative number.

Could you make roundRobinCnt as uint32, so that we can simplify this method. as the following:

Suggested change

	func (p *connectionPool) GenerateRoundRobinIndex() int32 {
	cnt := atomic.AddInt32(&p.roundRobinCnt, 1)
	if cnt < 0 {
	cnt = -cnt
	}
	return cnt % p.maxConnectionsPerHost
	}
	func (p *connectionPool) GenerateRoundRobinIndex() int32 {
	return atomic.AddUint32(&p.roundRobinCnt, 1) % p.maxConnectionsPerHost
	}

[BewareMyPower]

I'm wondering why not just generate a random integer?

[shibd]

Could you make roundRobinCnt as uint32, so that we can simplify this method. as the following:

Done

I'm wondering why not just generate a random integer?

Here just keep original logic:

pulsar-client-go/pulsar/internal/connection_pool.go

Lines 154 to 161 in ffba2a8

	func (p *connectionPool) getMapKey(logicalAddr *url.URL, physicalAddr *url.URL) string {
	cnt := atomic.AddInt32(&p.roundRobinCnt, 1)
	if cnt < 0 {
	cnt = -cnt
	}
	idx := cnt % p.maxConnectionsPerHost
	return fmt.Sprint(logicalAddr.Host, "-", physicalAddr.Host, "-", idx)
	}

[BewareMyPower]

Yes, the consumer's connection is called after the Subscribe RPC is done.

When a partition consumer (pc) receives a CLOSE_CONSUMER command,

1. In connection.handleCloseConsumer, it calls consumer.ConnectionClosed(closeConsumer), which sends a connectionClosed event to pc.connectCloseCh.
2. In pc.runEventsLoop, reconnectToBroker is called, all requests sent via pc._getConn() will wait until this method is done.

What I have concern is, before the reconnectToBroker is done, the pc._getConn() will return a non-nil outdated connection. If _getConn() is not called in runEventsLoop(), it might return an outdated connection.

There is another possible case that internalXxx could be called before reconnectToBroker is called, i.e. pc.eventsCh receives an event before pc.connectClosedCh.

Ideally, I think the best solution is to call pc._setConn(nil) before sending the connectionClosed event in ConnectionClosed and check if _getConn is nil in RequestOnCnx or RequestOnCnxNoWait.

[shibd]

hi, @BewareMyPower

1. Even if we set the connection to null when receive cmd ConsumerClosed, we cannot completely guarantee that other API calls won't use the old connection to send to the old broker. Setting the connection to null and performing checks can help mitigate this issue to some extent.
2. If the connection is null when processing API requests (e.g., getLastMessageId), we should have an internal retry mechanism instead of throwing an exception to the client, as connection reconnection should be transparent to the caller.

In this PR, I will throw an error when the connection is null. In a future PR, I will implement the retry logic.