Fix transaction coordinator client cann't reconnect to the broker

pulsar/internal/connection.go

@@ -84,6 +84,7 @@ type Connection interface {
 	ID() string
 	GetMaxMessageSize() int32
 	Close()
+	WaitForClose() <-chan interface{}
 	IsProxied() bool
 }
 
@@ -995,6 +996,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)