Fix: Potential data/write conflicts

[gunli]

Fixes #1335

Master Issue: #1335

Motivation

Sync all the write calls of a connection into one goroutine to avoid potential data/write conflicts

Modifications

Sync all the write calls of a connection into one goroutine.

Verifying this change

• Make sure that the change passes the CI checks.

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

No

@shibd @nodece @RobertIndie @BewareMyPower @crossoverJie
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? (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]

This bug has been discussed in #1333.
@shibd @nodece @RobertIndie @BewareMyPower @crossoverJie

[nodece]

LGTM

[shibd]

thanks for your contribution.

I don't understand what the issue is with different goroutines calling ctx.write. Didn't we add a lock before writing?

pulsar-client-go/pulsar/internal/connection.go

Lines 518 to 519 in 024e230

	c.writeBufferLock.Lock()
	defer c.writeBufferLock.Unlock()

If there is a issue, a single client may have multiple consumer and producer objects(many goroutines), which might use the same connection. How can we ensure they are all in the same goroutine?

BTW: Would it be possible to add unit tests to help reproduce the issue instead of guessing?

[gunli]

I don't understand what the issue is with different goroutines calling ctx.write

It is the same as writing a file in different threads, it will lead to data conflict.

[nodece]

@shibd Writing data and listening for the close event in separate goroutines can lead to inconsistencies in the response. This way may cause incorrect responses due to the race condition between the broker's successful response and the connection close.

For example:

• Time1: The broker responds with success.
• Time2: The connection is closed.
• Result: Depending on timing, you may receive a 'success' response or an 'error: closed' response, which can lead to unexpected behavior."

@nodece Thanks for explaining, I see.

I will revert my requested change, but I prefer to have a unit test to cover this.

[BewareMyPower]

In addition to @nodece's explanation, currently the internalWriteData method could be called in two goroutines, which might have some underlying issues.

	go func() {
		for {
			select {
			/* ... */
			case req := <-c.incomingRequestsCh:
				/* ... */
				c.internalSendRequest(req) // [1]
			}
		}
	}()

	for {
		select {
		/* ... */
		case data := <-c.writeRequestsCh:
			/* ... */
			c.internalWriteData(data) // [2]

[gunli]

IT failed again, but it passed on my host, I think it is nothing to do with this PR.

[nodece]

@gunli Please check CI.

[nodece]

@gunli The failure of TestAvailablePermitsLeak occurs because when a consumer with ConsumerCryptoFailureActionDiscard receives a message, it sends an acknowledgment command to the broker. However, since both read and write operations are handled within the same goroutine, the write operation can block the read operation. This results in a deadlock-like situation where the read operation cannot proceed, ultimately leading to a context.DeadlineExceeded error.

To resolve this issue, read and write operations should be handled in separate goroutines to avoid blocking and ensure smooth message processing.

[gunli]

@gunli The failure of TestAvailablePermitsLeak occurs because when a consumer with ConsumerCryptoFailureActionDiscard receives a message, it sends an acknowledgment command to the broker. However, since both read and write operations are handled within the same goroutine, the write operation can block the read operation. This results in a deadlock-like situation where the read operation cannot proceed, ultimately leading to a context.DeadlineExceeded error.

To resolve this issue, read and write operations should be handled in separate goroutines to avoid blocking and ensure smooth message processing.

OK, I will create a new PR to fix this, it seems connection.incomingCmdCh is redundant, because reading is run in a seperate goroutine: go c.reader.readFromConnection().

[gunli]

Please merge #1343 first, I will rebase and update this PR after that.

[gunli]

mark it as draft until #1343 is merged, I will rebase and update this PR after that.