Skip to content

Commit

Permalink
[fix] [issue 877] Fix ctx in partitionProducer.Send() is not performi…
Browse files Browse the repository at this point in the history
…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>
  • Loading branch information
Gleiphir2769 and shenjiaqi.2769 committed Jul 11, 2023
1 parent dd920ef commit be35740
Show file tree
Hide file tree
Showing 2 changed files with 33 additions and 3 deletions.
10 changes: 7 additions & 3 deletions pulsar/producer_partition.go
Original file line number Diff line number Diff line change
Expand Up @@ -1095,9 +1095,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,
Expand Down
26 changes: 26 additions & 0 deletions pulsar/producer_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -2024,3 +2024,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))
}

0 comments on commit be35740

Please sign in to comment.