Fix stuck when reconnect broker

[wolfstudy]

Signed-off-by: xiaolongran rxl@apache.org

Fixes #697

Motivation

As #697 said, In Go SDK, when the reconnection logic is triggered under certain conditions, the reconnection will not succeed due to request timeout.

Comparing the implementation of the Java SDK, we can see that each time the reconnection logic is triggered, the original connection will be closed and a new connection will be created.

So in this pr, we introduced a new reconnectFlag field in the connection struct to mark the reconnection state. When the broker actively informs the client to close the connection to trigger the reconnection logic, we will store it from the connections cache of the connectionPool. The old connection object is deleted, and a new connection is created to complete the reconnection

Modifications

• Add reconnectFlag in connection struct

[cckellogg]

How does the broker notify that the connection is closed?

[wolfstudy]

The action of the broker's notification to close the connection will be processed by the two commands CommandCloseProducer and CommandCloseConsumer. Generally, it may trigger the automatic load balancing of the topic or the unload operation will trigger the active closing of the broker.

[cckellogg]

This variable is a little confusing to me. What is this flag suppose to indicate? And if it's set is the connection in an unusable state?

[wolfstudy]

The calling path of the entire code is:

1. The broker notifies the client to close the connection and sends CommandCloseProducer or CommandCloseConsumer cmd to the client
2. The client receives this command and starts to enter the logic of handleCloseConsumer or handleCloseProducer
3. The client starts to trigger the ConnectionClosed action, and then enters the reconnection logic
4. In the reconnection logic, the GetConnection method will be called to obtain a connection

Previously in GetConnection(), we obtained the old connection, and now we want to change this behavior, expecting to remove the old connection object from the connections map when the ConnectionClosed action is triggered, and then from the GetConnection() method to get a new connection

[wolfstudy]

Based on the current wrapper implementation of Connection and ConnectionPool, we have two elegant ways to handle this:

1. In my current implementation, a flag is introduced into the struct of connection to mark the state of reconnection
2. When using changeState() to trigger handleCloseProducer, set the state of the current connection to close

[wolfstudy]

@cckellogg Choose to reuse the current code logic for processing. When receiving the CommandCloseProducer or CommandCloseConsumer cmd, set the current connection state to connectionClosed, and then delete the connection in the closed state through the following check-in GetConnection().

		if conn.closed() {
			delete(p.connections, key)
			p.log.Infof("Removed connection from pool key=%s logical_addr=%+v physical_addr=%+v",
				key, conn.logicalAddr, conn.physicalAddr)
			conn = nil // set to nil so we create a new one
		}

[lhotari]

Comparing the implementation of the Java SDK, we can see that each time the reconnection logic is triggered, the original connection will be closed and a new connection will be created.

@wolfstudy I think that might be an incorrect observation. In the Java SDK, the concepts (classes) are fairly misleading, and something that might seem to be a connection close isn't one. The ClientHandler is really a reference to a connection in the Java SDK. The ClientCnx isn't closed in the Java SDK in ClientHandler.connectionClosed
https://github.com/apache/pulsar/blob/04aa9e8e51869d1621a7e25402a656084eebfc09/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java#L122-L141