[PIP-165] Auto release idle connections

pulsar/client_impl.go

@@ -30,12 +30,12 @@ import (
 )
 
 const (
-	defaultConnectionTimeout     = 10 * time.Second
-	defaultOperationTimeout      = 30 * time.Second
-	defaultKeepAliveInterval     = 30 * time.Second
-	defaultMemoryLimitBytes      = 64 * 1024 * 1024
-	defaultConnectionMaxIdleTime = 60 * time.Second
-	minConnectionMaxIdleTime     = 60 * time.Second
+	defaultConnectionTimeout = 10 * time.Second
+	defaultOperationTimeout  = 30 * time.Second
+	defaultKeepAliveInterval = 30 * time.Second
+	defaultMemoryLimitBytes  = 64 * 1024 * 1024
+	defaultConnMaxIdleTime   = 180 * time.Second
+	minConnMaxIdleTime       = 60 * time.Second
 )
 
 type client struct {
@@ -60,10 +60,10 @@ func newClient(options ClientOptions) (Client, error) {
 
 	connectionMaxIdleTime := options.ConnectionMaxIdleTime
 	if connectionMaxIdleTime == 0 {
-		connectionMaxIdleTime = defaultConnectionMaxIdleTime
-	} else if connectionMaxIdleTime > 0 && connectionMaxIdleTime < minConnectionMaxIdleTime {
+		connectionMaxIdleTime = defaultConnMaxIdleTime
+	} else if connectionMaxIdleTime > 0 && connectionMaxIdleTime < minConnMaxIdleTime {
 		return nil, newError(InvalidConfiguration, fmt.Sprintf("Connection max idle time should be at least %f "+
-			"seconds", minConnectionMaxIdleTime.Seconds()))
+			"seconds", minConnMaxIdleTime.Seconds()))
 	} else {
 		logger.Debugf("Disable auto release idle connections")
 	}

pulsar/client_impl_test.go

@@ -1131,7 +1131,7 @@ func TestConfigureConnectionMaxIdleTime(t *testing.T) {
 		ConnectionMaxIdleTime: 1 * time.Second,
 	})
 
-	assert.Error(t, err, "Should be failed when the connectionMaxIdleTime is less than minConnectionMaxIdleTime")
+	assert.Error(t, err, "Should be failed when the connectionMaxIdleTime is less than minConnMaxIdleTime")
 
 	cli, err := NewClient(ClientOptions{
 		URL:                   serviceURL,

pulsar/internal/connection.go

@@ -936,23 +936,29 @@ func (c *connection) ResetLastActive() {
 }
 
 func (c *connection) isIdle() bool {
-	c.pendingLock.Lock()
-	if len(c.pendingReqs) != 0 {
-		return false
+	{
+		c.pendingLock.Lock()
+		defer c.pendingLock.Unlock()
+		if len(c.pendingReqs) != 0 {
+			return false
+		}
 	}
-	c.pendingLock.Unlock()
 
-	c.listenersLock.RLock()
-	if len(c.listeners) != 0 {
-		return false
+	{
+		c.listenersLock.RLock()
+		defer c.listenersLock.RUnlock()
+		if len(c.listeners) != 0 {
+			return false
+		}
 	}
-	c.listenersLock.RUnlock()
 
-	c.consumerHandlersLock.Lock()
-	if len(c.consumerHandlers) != 0 {
-		return false
+	{
+		c.consumerHandlersLock.Lock()
+		defer c.consumerHandlersLock.Unlock()
+		if len(c.consumerHandlers) != 0 {
+			return false
+		}
 	}
-	c.consumerHandlersLock.Unlock()
 
 	if len(c.incomingRequestsCh) != 0 || len(c.writeRequestsCh) != 0 {
 		return false

pulsar/internal/connection_pool.go

@@ -74,7 +74,7 @@ func NewConnectionPool(
 		metrics:               metrics,
 		closeCh:               make(chan struct{}),
 	}
-	go p.cleanConnections(connectionMaxIdleTime)
+	go p.checkAndCleanIdleConnections(connectionMaxIdleTime)
 	return p
 }
 
@@ -142,7 +142,7 @@ func (p *connectionPool) getMapKey(addr *url.URL) string {
 	return fmt.Sprint(addr.Host, '-', idx)
 }
 
-func (p *connectionPool) cleanConnections(maxIdleTime time.Duration) {
+func (p *connectionPool) checkAndCleanIdleConnections(maxIdleTime time.Duration) {
 	if maxIdleTime < 0 {
 		return
 	}

pulsar/internal/helper.go

@@ -19,10 +19,10 @@ package internal
 
 import "time"
 
-// These method should only used by tests
+// These method should only be used by tests
 
 func StartCleanConnectionsTask(p *ConnectionPool, connectionMaxIdleTime time.Duration) {
-	go (*p).(*connectionPool).cleanConnections(connectionMaxIdleTime)
+	go (*p).(*connectionPool).checkAndCleanIdleConnections(connectionMaxIdleTime)
 }
 
 func GetConnectionsCount(p *ConnectionPool) int {