You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by cr...@apache.org on 2024/07/29 06:09:06 UTC
(pulsar-client-go) branch master updated: [Issue 1259][producer] Prevent panic when calling Flush on closed producer (#1260)
This is an automated email from the ASF dual-hosted git repository.
crossoverjie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar-client-go.git
The following commit(s) were added to refs/heads/master by this push:
new fb805c0a [Issue 1259][producer] Prevent panic when calling Flush on closed producer (#1260)
fb805c0a is described below
commit fb805c0aea3506ba9df87bb066221469e35005a0
Author: Gaylor Bosson <gb...@taurusgroup.ch>
AuthorDate: Mon Jul 29 08:09:00 2024 +0200
[Issue 1259][producer] Prevent panic when calling Flush on closed producer (#1260)
---
pulsar/producer_partition.go | 4 ++++
pulsar/producer_test.go | 15 +++++++++++++++
2 files changed, 19 insertions(+)
diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
index cd89862a..5c038aa5 100755
--- a/pulsar/producer_partition.go
+++ b/pulsar/producer_partition.go
@@ -1494,6 +1494,10 @@ func (p *partitionProducer) Flush() error {
}
func (p *partitionProducer) FlushWithCtx(ctx context.Context) error {
+ if p.getProducerState() != producerReady {
+ return ErrProducerClosed
+ }
+
flushReq := &flushRequest{
doneCh: make(chan struct{}),
err: nil,
diff --git a/pulsar/producer_test.go b/pulsar/producer_test.go
index 0e38d637..5b23182d 100644
--- a/pulsar/producer_test.go
+++ b/pulsar/producer_test.go
@@ -523,6 +523,21 @@ func TestFlushInPartitionedProducer(t *testing.T) {
assert.Equal(t, msgCount, numOfMessages/2)
}
+func TestProducerReturnsErrorOnFlushWhenClosed(t *testing.T) {
+ client, err := NewClient(ClientOptions{URL: serviceURL})
+ assert.NoError(t, err)
+ defer client.Close()
+
+ producer, err := client.CreateProducer(ProducerOptions{Topic: newTopicName()})
+ assert.NoError(t, err)
+ assert.NotNil(t, producer)
+
+ producer.Close()
+
+ err = producer.FlushWithCtx(context.Background())
+ assert.Error(t, err)
+}
+
func TestRoundRobinRouterPartitionedProducer(t *testing.T) {
topicName := "public/default/partition-testRoundRobinRouterPartitionedProducer"
numberOfPartitions := 5