Fix:The producer is sometimes blocked when calling producer.SendAsync

[idehong]

Fixes #1365

Motivation

Go version: 1.23
pulsar-client-go verson: 0.14

When calling producer.SendAsync is blocked.

Modifications

When calling producer.SendAsync to produce a message, an error occurs. The producer.Close is called in the callback of SendAsync.

In the SendAsync function, the failed messages are traversed and a callback is called for each message, which will cause a deadlock with the close producer call.

	state := p.getProducerState()
	if state == producerClosing || state == producerClosed {
		return
	}

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

• Dependencies (does it add or upgrade a dependency): (yes / no)
• The public API: (yes / no)
• The schema: (yes / no / don't know)
• The default values of configurations: (yes / no)
• The wire protocol: (yes / no)

Documentation

• Does this pull request introduce a new feature? (yes / no)
• If yes, how is the feature documented? (not applicable / docs / GoDocs / not documented)
• If a feature is not applicable for documentation, explain why?
• If a feature is not documented yet in this PR, please create a followup issue for adding the documentation

[gunli]

Hi @idehong , You should NOT call producer.Close or some other producer's methods in the message callback, because the callback is called in the producer's goroutine, if you call producer.Close or some other producer's methods in this goroutine, it is like a recursive call, which may cause infinite loop or deadlock.

[crossoverJie]

I agree with @gunli ; you can refer to the usage here.

pulsar-client-go/pulsar/producer_test.go

Lines 153 to 167 in 06f4dd8

	for i := 0; i < 10; i++ {
	producer.SendAsync(context.Background(), &ProducerMessage{
	Payload: []byte("hello"),
	}, func(id MessageID, _ *ProducerMessage, e error) {
	if e != nil {
	log.WithError(e).Error("Failed to publish")
	errors.Put(e)
	} else {
	log.Info("Published message ", id)
	}
	wg.Done()
	})
	assert.NoError(t, err)
	}

[nodece]

@idehong, could you try the master branch? This issue may have been fixed by ##1422