Fix race

Author: RobertIndie

pulsar/client_impl_test.go

@@ -1163,53 +1163,53 @@ func TestAutoCloseIdleConnection(t *testing.T) {
 	topic := "TestAutoCloseIdleConnection"
 
 	// create consumer
-	consumer, err := cli.Subscribe(ConsumerOptions{
+	consumer1, err := cli.Subscribe(ConsumerOptions{
 		Topic:            topic,
 		SubscriptionName: "my-sub",
 	})
 	assert.Nil(t, err)
 
 	// create producer
-	producer, err := cli.CreateProducer(ProducerOptions{
+	producer1, err := cli.CreateProducer(ProducerOptions{
 		Topic:           topic,
 		DisableBatching: false,
 	})
 	assert.Nil(t, err)
 
-	testSendAndReceive(t, producer, consumer)
+	testSendAndReceive(t, producer1, consumer1)
 
 	pool := cli.(*client).cnxPool
 
-	producer.Close()
-	consumer.Close()
+	producer1.Close()
+	consumer1.Close()
 
 	assert.NotEqual(t, 0, internal.GetConnectionsCount(&pool))
 
 	internal.StartCleanConnectionsTask(&pool, 2*time.Second) // Enable auto idle connections release manually
 
-	time.Sleep(4 * time.Second) // Need to wait at least 2 * ConnectionMaxIdleTime
+	time.Sleep(6 * time.Second) // Need to wait at least 3 * ConnectionMaxIdleTime
 
 	assert.Equal(t, 0, internal.GetConnectionsCount(&pool))
 
 	// create consumer
-	consumer, err = cli.Subscribe(ConsumerOptions{
+	consumer2, err := cli.Subscribe(ConsumerOptions{
 		Topic:            topic,
 		SubscriptionName: "my-sub",
 	})
 	assert.Nil(t, err)
 
 	// create producer
-	producer, err = cli.CreateProducer(ProducerOptions{
+	producer2, err := cli.CreateProducer(ProducerOptions{
 		Topic:           topic,
 		DisableBatching: false,
 	})
 	assert.Nil(t, err)
 
 	// Ensure the client still works
-	testSendAndReceive(t, producer, consumer)
+	testSendAndReceive(t, producer2, consumer2)
 
-	producer.Close()
-	consumer.Close()
+	producer2.Close()
+	consumer2.Close()
 
 	cli.Close()
 }

pulsar/internal/connection.go

@@ -926,15 +926,35 @@ func (c *connection) ResetLastActive() {
 	c.lastActive = time.Now()
 }
 
-func (c *connection) CheckIdle(maxIdleTime time.Duration) bool {
-	c.Lock()
-	defer c.Unlock()
+func (c *connection) isIdle() bool {
+	c.pendingLock.Lock()
+	if len(c.pendingReqs) != 0 {
+		return false
+	}
+	c.pendingLock.Unlock()
+
+	c.listenersLock.RLock()
+	if len(c.listeners) != 0 {
+		return false
+	}
+	c.listenersLock.RUnlock()
 
-	if len(c.pendingReqs) != 0 ||
-		len(c.incomingRequestsCh) != 0 ||
-		len(c.writeRequestsCh) != 0 ||
-		len(c.listeners) != 0 ||
-		len(c.consumerHandlers) != 0 {
+	c.consumerHandlersLock.Lock()
+	if len(c.consumerHandlers) != 0 {
+		return false
+	}
+	c.consumerHandlersLock.Unlock()
+
+	if len(c.incomingRequestsCh) != 0 || len(c.writeRequestsCh) != 0 {
+		return false
+	}
+	return true
+}
+
+func (c *connection) CheckIdle(maxIdleTime time.Duration) bool {
+	// We don't need to lock here because this method should only be
+	// called in a single goroutine of the connectionPool
+	if !c.isIdle() {
 		c.lastActive = time.Now()
 	}
 	return time.Since(c.lastActive) > maxIdleTime