Repository: apache/pulsar-client-go

Issue:
Calling producer.Flush concurrently would hang indefinitely
#### Expected behavior

Calling `producer.Flush `concurrently should not hang forever

#### Actual behavior

Calling `producer.Flush` with multiple go-routines simutaneously, only one would succeed, others would hang forever.

#### Steps to reproduce

Assuming having a function that would send async messages and flush immediately.
Disclamer:
1. I know this is not a common usage but just to illustrate the situation
2. Normally we would add context with timeout for function calling, again, just to illustrate the situation, I've used the `context.Background()`

```
func test3(disableBatching bool) {
	fmt.Println("running test 3 with disableBatching", disableBatching)
	client, err := pulsarsdk.NewClient(pulsarsdk.ClientOptions{
		URL:            pulsarEndpoint,
		Authentication: pulsarsdk.NewAuthenticationToken(pulsarToken),
	})
	if err != nil {
		panic(err)
	}
	defer client.Close()
	producer, err := client.CreateProducer(pulsarsdk.ProducerOptions{
		Topic:           pulsarTopic,
		Name:            uuid.NewString(),
		DisableBatching: disableBatching,
	})
	if err != nil {
		panic(err)
	}
	defer producer.Close()

	ctx, cancel := registerExit()
	defer cancel()

	const n = 10
	var (
		wg sync.WaitGroup
	)
	wg.Add(n)
	for i := 0; i < n; i++ {
		go func(i int) {
			producer.SendAsync(ctx, &pulsarsdk.ProducerMessage{
				Payload: []byte(uuid.NewString()),
			}, func(id pulsarsdk.MessageID, message *pulsarsdk.ProducerMessage, err error) {
				wg.Done()
				if err != nil {
					panic(err)
				}
			})

			if err := producer.Flush(); err != nil {
				panic(err)
			}
		}(i)
	}

	wg.Wait()

```

With additional logs, you can see that one call of `producer.Flush` succeeds, while others hang indefinitely.

Check the `flush` related codes, first we can observe that the expected `case <-flushReq.doneCh` is never triggered.

```
func (p *partitionProducer) FlushWithCtx(ctx context.Context) error {
	if p.getProducerState() != producerReady {
		return ErrProducerClosed
	}

	flushReq := &flushRequest{
		doneCh: make(chan struct{}),
		err:    nil,
	}
	select {
	case <-ctx.Done():
		return ctx.Err()
	case p.cmdChan <- flushReq:
	}

	// wait for the flush request to complete
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-flushReq.doneCh:
		return flushReq.err
	}
}
```

The root cause might be here:

```
func (p *partitionProducer) internalFlush(fr *flushRequest) {
	p.clearPendingSendRequests()

	if !p.options.DisableBatching {
		p.internalFlushCurrentBatch()
	}

	pi, ok := p.pendingQueue.PeekLast().(*pendingItem)
	if !ok {
		close(fr.doneCh)
		return
	}

	// lock the pending request while adding requests
	// since the ReceivedSendReceipt func iterates over this list
	pi.Lock()
	defer pi.Unlock()

	if pi.isDone {
		// The last item in the queue has been completed while we were
		// looking at it. It's safe at this point to assume that every
		// message enqueued before Flush() was called are now persisted
		close(fr.doneCh)
		return
	}

	pi.flushCallback = func(err error) {
		fr.err = err
		close(fr.doneCh)
	}
}

```

`pi, ok := p.pendingQueue.PeekLast().(*pendingItem)` returns the same pi for all concurrent flush calls, causing `pi.flushCallback` to be overwritten. As a result, only the most recent `fr.doneCh` will be closed, causing all other flush requests to hang indefinitely.

#### Possible fix

I've applied a monkey patch of `pendingItem` by altering:
```
type pendingItem struct {
	sync.Mutex
	buffer        internal.Buffer
	sequenceID    uint64
	createdAt     time.Time
	sentAt        time.Time
	sendRequests  []interface{}
	isDone        bool
	flushCallback func(err error)
}
```

to:
```
type pendingItem struct {
	sync.Mutex
	buffer        internal.Buffer
	sequenceID    uint64
	createdAt     time.Time
	sentAt        time.Time
	sendRequests  []interface{}
	isDone        bool
	flushCallbacks []func(err error)
}
```

And:
```
func (i *pendingItem) done(err error) {
	if i.isDone {
		return
	}
	i.isDone = true
	buffersPool.Put(i.buffer)
	if i.flushCallback != nil {
		i.flushCallback(err)
	}
}

```

to
```
func (i *pendingItem) done(err error) {
	if i.isDone {
		return
	}
	i.isDone = true
	buffersPool.Put(i.buffer)
	for _, cb := range i.flushCallback {
		cb(err)
	}
}

```

Same applies to `internalFlush`:
```
	pi.flushCallbacks = append(pi.flushCallbacks, func(err error) {
		fr.err = err
		close(fr.doneCh)
	})

```

And the `test3` would pass.

#### System configuration
**Pulsar version**:  3.2.4
**Pulsar-client-go version**: 0.14.0


Discussion:
@crossoverJie: Thanks for the report, could you submit a PR to fix this?@thomas-bousquet: I made the change in this PR https://github.com/apache/pulsar-client-go/pull/1409

Solution patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 510f182..b7aee02 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -50,13 +50,7 @@ jobs:
       matrix:
         go-version: [ '1.23', '1.24' ]
     steps:
-      - uses: actions/checkout@v3
-      - name: Check for Docker images
-        id: check_images
-        run: echo "::set-output name=images::$(docker images -q | wc -l)"
-      - name: Clean Docker cache if images exist
-        if: ${{ steps.check_images.outputs.images > 0 }}
-        run: docker rmi $(docker images -q) -f && df -h        
+      - uses: actions/checkout@v3      
       - uses: actions/setup-go@v3
         with:
           go-version: ${{ matrix.go-version }}
diff --git a/.gitignore b/.gitignore
index bfa04f0..91999bb 100644
--- a/.gitignore
+++ b/.gitignore
@@ -16,3 +16,4 @@ pulsar-perf
 bin
 
 vendor/
+logs/
diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
index 1d65ec8..9841222 100755
--- a/pulsar/producer_partition.go
+++ b/pulsar/producer_partition.go
@@ -839,15 +839,15 @@ func (p *partitionProducer) internalSingleSend(
 
 type pendingItem struct {
 	sync.Mutex
-	ctx           context.Context
-	cancel        context.CancelFunc
-	buffer        internal.Buffer
-	sequenceID    uint64
-	createdAt     time.Time
-	sentAt        time.Time
-	sendRequests  []interface{}
-	isDone        bool
-	flushCallback func(err error)
+	ctx            context.Context
+	cancel         context.CancelFunc
+	buffer         internal.Buffer
+	sequenceID     uint64
+	createdAt      time.Time
+	sentAt         time.Time
+	sendRequests   []interface{}
+	isDone         bool
+	flushCallbacks []func(err error)
 }
 
 func (p *partitionProducer) internalFlushCurrentBatch() {
@@ -1064,10 +1064,10 @@ func (p *partitionProducer) internalFlush(fr *flushRequest) {
 		return
 	}
 
-	pi.flushCallback = func(err error) {
+	pi.flushCallbacks = append(pi.flushCallbacks, func(err error) {
 		fr.err = err
 		close(fr.doneCh)
-	}
+	})
 }
 
 // clearPendingSendRequests makes sure to push forward previous sending requests
@@ -1749,8 +1749,8 @@ func (i *pendingItem) done(err error) {
 	i.isDone = true
 	// return the buffer to the pool after all callbacks have been called.
 	defer i.buffer.Release()
-	if i.flushCallback != nil {
-		i.flushCallback(err)
+	for _, callback := range i.flushCallbacks {
+		callback(err)
 	}
 
 	if i.cancel != nil {



Test patch:
diff --git a/pulsar/consumer_test.go b/pulsar/consumer_test.go
index 80ce0d2..8a4ea3f 100644
--- a/pulsar/consumer_test.go
+++ b/pulsar/consumer_test.go
@@ -5230,7 +5230,7 @@ func sendMessages(t *testing.T, client Client, topic string, startIndex int, num
 			}
 		}
 	}
-	assert.Nil(t, producer.Flush())
+	assert.Nil(t, producer.FlushWithCtx(ctx))
 }
 
 func receiveMessages(t *testing.T, consumer Consumer, numMessages int) []Message {
@@ -5276,10 +5276,10 @@ func TestAckResponseNotBlocked(t *testing.T) {
 			}
 		})
 		if i%100 == 99 {
-			assert.Nil(t, producer.Flush())
+			assert.Nil(t, producer.FlushWithCtx(ctx))
 		}
 	}
-	producer.Flush()
+	producer.FlushWithCtx(ctx)
 	producer.Close()
 
 	// Set a small receiver queue size to trigger ack response blocking if the internal `queueCh`
diff --git a/pulsar/producer_test.go b/pulsar/producer_test.go
index 9a3b99f..02b4b67 100644
--- a/pulsar/producer_test.go
+++ b/pulsar/producer_test.go
@@ -449,6 +449,78 @@ func TestFlushInProducer(t *testing.T) {
 	assert.Equal(t, msgCount, numOfMessages)
 }
 
+// TestConcurrentFlushInProducer validates that concurrent flushes don't create a deadlock
+func TestConcurrentFlushInProducer(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: serviceURL,
+	})
+	assert.NoError(t, err)
+	defer client.Close()
+
+	topicName := "test-concurrent-flushes-in-producer"
+	subName := "subscription-name"
+	ctx := context.Background()
+
+	// set batch message number numOfMessages, and max delay 10s
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:           topicName,
+		DisableBatching: false,
+	})
+	assert.Nil(t, err)
+	defer producer.Close()
+
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:            topicName,
+		SubscriptionName: subName,
+	})
+	assert.Nil(t, err)
+	defer consumer.Close()
+
+	expectedMsgCount := 100
+
+	var wg sync.WaitGroup
+
+	wg.Add(expectedMsgCount)
+
+	errs := make(chan error, expectedMsgCount*2)
+
+	// Each message in sent and flushed concurrently
+	for range expectedMsgCount {
+		go func() {
+			defer wg.Done()
+			producer.SendAsync(ctx, &ProducerMessage{
+				Payload: []byte("anythingWorksInThatPayload"),
+			}, func(_ MessageID, _ *ProducerMessage, e error) {
+				errs <- e
+			})
+
+			errs <- producer.FlushWithCtx(ctx)
+		}()
+	}
+
+	// Wait for all concurrent async publications and flushes to complete
+	wg.Wait()
+
+	// Make sure that there were no error publishing or flushing
+	close(errs)
+	var errElementCount int
+	for e := range errs {
+		errElementCount++
+		assert.Nil(t, e)
+	}
+	assert.Equal(t, errElementCount, expectedMsgCount*2)
+
+	// Make sure all messages were processed successfully
+	var receivedMsgCount int
+	for range expectedMsgCount {
+		_, err := consumer.Receive(ctx)
+		assert.Nil(t, err)
+		receivedMsgCount++
+	}
+
+	assert.Equal(t, receivedMsgCount, expectedMsgCount)
+}
+
 func TestFlushInPartitionedProducer(t *testing.T) {
 	topicName := "public/default/partition-testFlushInPartitionedProducer"
 
diff --git a/pulsaradmin/pkg/admin/subscription_test.go b/pulsaradmin/pkg/admin/subscription_test.go
index 087a4a3..1e4a7b2 100644
--- a/pulsaradmin/pkg/admin/subscription_test.go
+++ b/pulsaradmin/pkg/admin/subscription_test.go
@@ -126,7 +126,7 @@ func TestPeekMessageForPartitionedTopic(t *testing.T) {
 			Payload: []byte(fmt.Sprintf("hello-%d", i)),
 		}, nil)
 	}
-	err = producer.Flush()
+	err = producer.FlushWithCtx(ctx)
 	if err != nil {
 		return
 	}