You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "RobertIndie (via GitHub)" <gi...@apache.org> on 2025/04/18 09:39:04 UTC

[PR] Fix SendAsync won't be timeout during producer reconnection [pulsar-client-go]

RobertIndie opened a new pull request, #1356:
URL: https://github.com/apache/pulsar-client-go/pull/1356

   
   Fixes #1332
   
   
   ### Motivation
   
   The root cause is that the producer event loops become busy during reconnection, preventing messages in dataChan from timing out.
   
   SendAsync can wait until the runEventLoop processes it and pushes it into the pendingQueue or a batch, just like the Java client. Before entering the pendingQueue, SendAsync itself can check for timeouts and handle the callback. After entering the pendingQueue, `failTimeoutMessages` can manage the timeout.
   
   ### Modifications
   
   - Introduced a new channel `enqueued` to make SendAsync wait until the sendRequest is added to the pending queue.
   - Use `ctx` to check for timeouts and invoke the callback if a timeout occurs.
   
   ### Verifying this change
   
   The test `TestSendAsyncCouldTimeoutWhileReconnecting` is based on test from this PR: https://github.com/apache/pulsar-client-go/pull/1345
   
   ### 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
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Fix SendAsync won't be timeout during producer reconnection [pulsar-client-go]

Posted by "gunli (via GitHub)" <gi...@apache.org>.
gunli commented on PR #1356:
URL: https://github.com/apache/pulsar-client-go/pull/1356#issuecomment-3138430037

   我认为这个bug的根因是我们连接管理模块不太合理，它和业务逻辑层的边界不清晰，混合了太多生者/消费者的逻辑在里面，或者说生产者/消费者也参与了太多连接管理的职责，我们可以看到生产者/消费者里有很多类似的连接相关的代码，我们需要重构才能更好的解决问题。像 #1345  这样加入新的channel和再加一个goroutine事件循环，也只是为了解决问题和解决，治标不治本还增加了复杂度，生产者里的事件循环已经够复杂了，如果仅为了解决问题，还真不如 #1333 这样改。
   
   我认为连接管理模块只做好它的事情就行，创建连接、关闭连接、心跳、负载均衡、重连，对外提供获取连接、释放连接的接口，像个连接池，连接异常关闭，它在内部协程恢复连接就行，至于生产者、消费者，连接异常时，就去连接管理模块获取新连接，获取不到就不发数据好了，既然提供了异步的接口，对用户来说，除了队列满，就不能因为你SDK内部的问题影响（阻塞）我写数据，大不了最终失败，写进队列的数据，你发不出去该超时就超时，告诉我失败就行，你恢复了就继续发，现在连接异常，在生产者去死循环恢复连接就不应该是它的职责，同样的代码在消费者也有，很多都是重复的，有些bug生产者改了，消费者也要改一次，如果只了解生产者或者消费者，另一边的bug还不会改，我就提交过一些PR，有些是只在消费者改的，或者有些只
 有生产者改的，同样的问题需要都知情才会去改，这些都应该下沉到连接管理层去做。


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Fix SendAsync won't be timeout during producer reconnection [pulsar-client-go]

Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #1356:
URL: https://github.com/apache/pulsar-client-go/pull/1356#discussion_r2050421695


##########
pulsar/producer_partition.go:
##########
@@ -907,6 +907,13 @@ func (p *partitionProducer) writeData(buffer internal.Buffer, sequenceID uint64,
 			sequenceID:   sequenceID,
 			sendRequests: callbacks,
 		})
+		for _, cb := range callbacks {
+			if sr, ok := cb.(*sendRequest); ok {
+				if sr.enqueued != nil {

Review Comment:
   Consider ensuring that the enqueued channel is closed only once. Using a sync.Once in the sendRequest struct or another mechanism could help prevent any potential double-close panic.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Fix SendAsync won't be timeout during producer reconnection [pulsar-client-go]

Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie commented on PR #1356:
URL: https://github.com/apache/pulsar-client-go/pull/1356#issuecomment-3279957053

   > 我认为这个bug的根因是我们连接管理模块不太合理，它和业务逻辑层的边界不清晰，混合了太多生者/消费者的逻辑在里面，或者说生产者/消费者也参与了太多连接管理的职责，我们可以看到生产者/消费者里有很多类似的连接相关的代码，我们需要重构才能更好的解决问题。像 #1345 这样加入新的channel和再加一个goroutine事件循环，也只是为了解决问题和解决，治标不治本还增加了复杂度，生产者里的事件循环已经够复杂了，如果仅为了解决问题，还真不如 #1333 这样改。
   > 
   > 我认为连接管理模块只做好它的事情就行，创建连接、关闭连接、心跳、负载均衡、重连，对外提供获取连接、释放连接的接口，像个连接池，连接异常关闭，它在内部协程恢复连接就行，至于生产者、消费者，连接异常时，就去连接管理模块获取新连接，获取不到就不发数据好了，既然提供了异步的接口，对用户来说，除了队列满，就不能因为你SDK内部的问题影响（阻塞）我写数据，大不了最终失败，写进队列的数据，你发不出去该超时就超时，告诉我失败就行，你恢复了就继续发，现在连接异常，在生产者去死循环恢复连接就不应该是它的职责，同样的代码在消费者也有，很多都是重复的，有些bug生产者改了，消费者也要改一次，如果只了解生产者或者消费者，另一边的bug还不会改，我就提交过一些PR，有些是只在消费者改的，或者有些�
 ��有生产者改的，同样的问题需要都知情才会去改，这些都应该下沉到连接管理层去做。
   
   Thanks for your analysis. I agree that we should separate connection-related logic, such as reconnection, from the data and command event loop. However, the producer still needs to handle the reconnection logic because it is responsible for message resending. The connection layer only abstracts the underlying TCP connection for sending requests and receiving responses. Deciding which requests to send and what to do after reconnecting should be the producer's responsibility. Therefore, the current division of responsibilities looks good to me. We just need to ensure the connection logic and data logic do not run in the same event loop.
   
   Also, if we separate them into two event loops (a reconnection loop and a data loop), we must be very careful about message ordering. This is because both loops can now send messages to the connection. The reconnection loop will resend messages after a connection is re-established, while the data loop will send new messages. Otherwise, this could lead to the issue described here: https://github.com/apache/pulsar-client-go/issues/1027.
   
   I have proposed a new PR that fixes this in another way and also solves the issues mentioned above: https://github.com/apache/pulsar-client-go/pull/1422.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Fix SendAsync won't be timeout during producer reconnection [pulsar-client-go]

Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie commented on code in PR #1356:
URL: https://github.com/apache/pulsar-client-go/pull/1356#discussion_r2340081203


##########
pulsar/producer_partition.go:
##########
@@ -1353,6 +1367,17 @@ func (p *partitionProducer) internalSendAsync(
 	}
 
 	p.dataChan <- sr
+	select {

Review Comment:
   Yes. I agree. I have put a new PR https://github.com/apache/pulsar-client-go/pull/1422 which also address this issue. PTAL



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Fix SendAsync won't be timeout during producer reconnection [pulsar-client-go]

Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie closed pull request #1356: Fix SendAsync won't be timeout during producer reconnection
URL: https://github.com/apache/pulsar-client-go/pull/1356


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Fix SendAsync won't be timeout during producer reconnection [pulsar-client-go]

Posted by "gunli (via GitHub)" <gi...@apache.org>.
gunli commented on code in PR #1356:
URL: https://github.com/apache/pulsar-client-go/pull/1356#discussion_r2061982675


##########
pulsar/producer_partition.go:
##########
@@ -1353,6 +1367,17 @@ func (p *partitionProducer) internalSendAsync(
 	}
 
 	p.dataChan <- sr
+	select {

Review Comment:
   the `select` here will block `SendAsync()` when producer is in reconnecting while the pengding queue is not full, it is not good for those latency-sensitive applications such as game.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Fix SendAsync won't be timeout during producer reconnection [pulsar-client-go]

Posted by "gunli (via GitHub)" <gi...@apache.org>.
gunli commented on code in PR #1356:
URL: https://github.com/apache/pulsar-client-go/pull/1356#discussion_r2061981028


##########
pulsar/producer_partition.go:
##########
@@ -1353,6 +1367,17 @@ func (p *partitionProducer) internalSendAsync(
 	}
 
 	p.dataChan <- sr
+	select {
+	case <-sr.enqueued:
+	case <-ctx.Done():
+		err := ctx.Err()

Review Comment:
   Generally, timeout is not set to the ctx provided by application/user, may be we should update the ctx with context.WithTimeout(ctx, config.timeout)?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org