Fix transaction coordinator client cann't reconnect to the broker (#1237

)

Fixes #1227

### Motivation

There are some issues with the `transactionCoordinatorClient` in the go client. When using the transaction, if there are any reconnection happens during the transaction operation. The connection to the transaction coordinator won't be reconnected. This causes all following operations to fail with the `connection closed` error.

### Modifications

- Introduced `transactionHandler` to manage reconnections and handle requests.
- Fix the tc client would crash if the broker doesn't enable the transaction
- Improved the error handling in the tc client.

pulsar/internal/connection.go

@@ -87,6 +87,7 @@ type Connection interface {
 	ID() string
 	GetMaxMessageSize() int32
 	Close()
+	WaitForClose() <-chan interface{}
 	IsProxied() bool
 }
 
@@ -1049,6 +1050,10 @@ func (c *connection) CheckIdle(maxIdleTime time.Duration) bool {
 	return time.Since(c.lastActive) > maxIdleTime
 }
 
+func (c *connection) WaitForClose() <-chan interface{} {
+	return c.closeCh
+}
+
 // Close closes the connection by
 // closing underlying socket connection and closeCh.
 // This also triggers callbacks to the ConnectionClosed listeners.

pulsar/internal/connection_pool.go

@@ -34,6 +34,9 @@ type ConnectionPool interface {
 	// GetConnection get a connection from ConnectionPool.
 	GetConnection(logicalAddr *url.URL, physicalAddr *url.URL) (Connection, error)
 
+	// GetConnections get all connections in the pool.
+	GetConnections() map[string]Connection
+
 	// Close all the connections in the pool
 	Close()
 }
@@ -124,6 +127,16 @@ func (p *connectionPool) GetConnection(logicalAddr *url.URL, physicalAddr *url.U
 	return conn, err
 }
 
+func (p *connectionPool) GetConnections() map[string]Connection {
+	p.Lock()
+	conns := make(map[string]Connection)
+	for k, c := range p.connections {
+		conns[k] = c
+	}
+	p.Unlock()
+	return conns
+}
+
 func (p *connectionPool) Close() {
 	p.Lock()
 	close(p.closeCh)