RobertIndie opened a new pull request, #1394: URL: https://github.com/apache/pulsar-client-go/pull/1394
Fixes https://github.com/apache/pulsar-client-go/issues/1371 ### Motivation The issue is that the buffer is sent to the connection when the data is flushed at this line: https://github.com/apache/pulsar-client-go/blob/0ab28c229e4dab2adb505a135325b6ede6e0e4f4/pulsar/producer_partition.go#L910 Because of this, both the data request in the connection and the pending item hold onto this buffer. However, if the pending item fails due to a timeout or the producer closing, the buffer is returned to the buffer pool at this line: https://github.com/apache/pulsar-client-go/blob/0ab28c229e4dab2adb505a135325b6ede6e0e4f4/pulsar/producer_partition.go#L1747 Yet, the connection still has this buffer. When the connection tries to handle the data request, it will try to use a buffer that has already been returned. This causes a data race. This fix implements proper reference counting to ensure buffers are only returned to the pool when all references are released, preventing the data race. ### Modifications - Added `Retain()` and `Release()` methods to the `Buffer` interface - Implemented atomic reference counting using `atomic.Int64` - Buffers are only returned to the pool when reference count reaches zero - Add new metric `pulsar_client_sending_buffers_count` to show the sending buffers count. - This PR also fixes serval minor issues like data/request cleanup when closing the connection ### 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
