fix producer block (#326)

Co-authored-by: 灰柯 <huike@tuya.com>
apache · Jul 24, 2020 · c75aa62 · c75aa62
1 parent 65cb19a
commit c75aa62
Show file tree

Hide file tree

Showing 3 changed files with 6 additions and 3 deletions.
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
@@ -180,7 +180,9 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
 		dlq:                  dlq,
 		log:                  log.WithField("topic", options.topic),
 	}
-	pc.log = pc.log.WithField("name", pc.name).WithField("subscription", options.subscription)
+	pc.log = pc.log.WithField("name", pc.name).
+		WithField("subscription", options.subscription).
+		WithField("consumerID", pc.consumerID)
 	pc.nackTracker = newNegativeAcksTracker(pc, options.nackRedeliveryDelay)
 
 	err := pc.grabConn()

diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
@@ -636,7 +636,7 @@ func (c *connection) handleCloseConsumer(closeConsumer *pb.CommandCloseConsumer)
 
 	if consumer, ok := c.consumerHandler(consumerID); ok {
 		consumer.ConnectionClosed()
-		delete(c.listeners, consumerID)
+		c.DeleteConsumeHandler(consumerID)
 	} else {
 		c.log.WithField("consumerID", consumerID).Warnf("Consumer with ID not found while closing consumer")
 	}

diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
@@ -150,7 +150,8 @@ func newPartitionProducer(client *client, topic string, options *ProducerOptions
 		return nil, err
 	}
 
-	p.log = p.log.WithField("producer_name", p.producerName)
+	p.log = p.log.WithField("producer_name", p.producerName).
+		WithField("producerID", p.producerID)
 	p.log.WithField("cnx", p.cnx.ID()).Info("Created producer")
 	atomic.StoreInt32(&p.state, producerReady)