fix: release pending requests when the connection is close

apache · Aug 8, 2024 · 4359d31 · 4359d31
1 parent 745536b
commit 4359d31
Showing 1 changed file with 7 additions and 0 deletions.
diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
@@ -1070,6 +1070,13 @@ func (c *connection) Close() {
 
 		close(c.closeCh)
 
+		c.pendingLock.Lock()
+		for id, req := range c.pendingReqs {
+			delete(c.pendingReqs, id)
+			req.callback(nil, errors.New("connection closed"))
+		}
+		c.pendingLock.Unlock()
+
 		listeners := make(map[uint64]ConnectionListener)
 		c.listenersLock.Lock()
 		for id, listener := range c.listeners {