Fix: Potential data/write conflicts (#1336)

* rebase to master to merge #1343 * make consumer is not nil * revert test case * revert test case
apache · Mar 7, 2025 · 042bfcd · 042bfcd
1 parent 802aa69
commit 042bfcd
Showing 1 changed file with 6 additions and 16 deletions.
diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
@@ -411,26 +411,16 @@ func (c *connection) run() {
 	c.log.Debugf("Connection run starting with request capacity=%d queued=%d",
 		cap(c.incomingRequestsCh), len(c.incomingRequestsCh))
 
-	go func() {
-		for {
-			select {
-			case <-c.closeCh:
-				c.failLeftRequestsWhenClose()
-				return
-
-			case req := <-c.incomingRequestsCh:
-				if req == nil {
-					return // TODO: this never gonna be happen
-				}
-				c.internalSendRequest(req)
-			}
-		}
-	}()
-
 	for {
 		select {
 		case <-c.closeCh:
+			c.failLeftRequestsWhenClose()
 			return
+		case req := <-c.incomingRequestsCh:
+			if req == nil {
+				return // TODO: this never gonna be happen
+			}
+			c.internalSendRequest(req)
 		case data := <-c.writeRequestsCh:
 			if data == nil {
 				return