[fix] [issue 877] Fix ctx in partitionProducer.Send() is not performi…

…ng as expected (#1053)

Fixes #877

### Motivation

The original PR is #878. Because the original author @billowqiu has not continued to reply to the review comments for a long time, resubmit the fix here.

### Modifications

- Add select for ctx and doneCh in partitionProducer.Send()

---------

Co-authored-by: shenjiaqi.2769 <shenjiaqi.2769@bytedance.com>
(cherry picked from commit be35740)

pulsar/producer_partition.go

@@ -1102,9 +1102,13 @@ func (p *partitionProducer) Send(ctx context.Context, msg *ProducerMessage) (Mes
 	}, true)
 
 	// wait for send request to finish
-	<-doneCh
-
-	return msgID, err
+	select {
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	case <-doneCh:
+		// send request has been finished
+		return msgID, err
+	}
 }
 
 func (p *partitionProducer) SendAsync(ctx context.Context, msg *ProducerMessage,

pulsar/producer_test.go

@@ -2198,3 +2198,29 @@ func testSendMessagesWithMetadata(t *testing.T, disableBatch bool) {
 	assert.Equal(t, msg.OrderingKey, recvMsg.OrderingKey())
 	assert.Equal(t, msg.Properties, recvMsg.Properties())
 }
+
+func TestProducerSendWithContext(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+	assert.NoError(t, err)
+	defer client.Close()
+
+	topicName := newTopicName()
+	// create producer
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:           topicName,
+		DisableBatching: true,
+	})
+	assert.Nil(t, err)
+	defer producer.Close()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	// Make ctx be canceled to invalidate the context immediately
+	cancel()
+	_, err = producer.Send(ctx, &ProducerMessage{
+		Payload: make([]byte, 1024*1024),
+	})
+	//  producer.Send should fail and return err context.Canceled
+	assert.True(t, errors.Is(err, context.Canceled))
+}