You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "nodece (via GitHub)" <gi...@apache.org> on 2026/05/20 07:32:18 UTC
[PR] fix: prevent sendRequest double-done panic [pulsar-client-go]
nodece opened a new pull request, #1501:
URL: https://github.com/apache/pulsar-client-go/pull/1501
## Motivation
Under timeout/reconnect races, the same sendRequest can hit done() more than once from different paths. Since done() clears and returns the object to sendRequestPool, a later duplicate call can access cleared fields and panic with nil pointer dereference.
## Modifications
- Add an idempotency guard (doneFlag) in sendRequest.done() so duplicate invocations on the same instance return immediately.
- Keep the guard raised after reset/put to block stale-pointer calls in the reset-to-reinit window.
- Introduce newSendRequest(...) and newChunkSendRequest(...) helpers and route request creation through them, centralizing initialization for pooled objects.
- Update tests to use constructor helpers where sendRequest is created in tests.
- Add unit test TestSendRequestDoneIsIdempotentAfterPutToPool to verify a second done() call on the same pointer does not panic after pool return.
--
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: prevent sendRequest double-done panic [pulsar-client-go]
Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie commented on PR #1501:
URL: https://github.com/apache/pulsar-client-go/pull/1501#issuecomment-4591501137
Could you help take a look the comments from Copilot? @nodece
--
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: prevent sendRequest double-done panic [pulsar-client-go]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on PR #1501:
URL: https://github.com/apache/pulsar-client-go/pull/1501#issuecomment-4601585575
@RobertIndie Done.
--
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: prevent sendRequest double-done panic [pulsar-client-go]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #1501:
URL: https://github.com/apache/pulsar-client-go/pull/1501#discussion_r3309813776
##########
pulsar/producer_partition.go:
##########
@@ -1700,6 +1721,8 @@ func (sr *sendRequest) done(msgID MessageID, err error) {
if pool != nil {
// reset all the fields
*sr = sendRequest{}
+ // Keep the guard raised until the object is reinitialized from the pool.
+ sr.doneFlag.Store(true)
Review Comment:
`done()` resets the entire struct via `*sr = sendRequest{}` after setting `doneFlag` at the top. That assignment temporarily clears `doneFlag` back to false, reopening a race window where a concurrent second `done()` call can pass the CAS and run on a partially-reset object. To keep idempotency under concurrent calls, avoid overwriting `doneFlag` during reset (e.g., clear fields individually or use a reset helper that preserves `doneFlag` as true throughout reset+Put).
##########
pulsar/producer_partition.go:
##########
@@ -1648,7 +1609,67 @@ type sendRequest struct {
maxMessageSize int32
}
+func newSendRequest(
+ ctx context.Context,
+ p *partitionProducer,
+ msg *ProducerMessage,
+ callback func(MessageID, *ProducerMessage, error),
+ flushImmediately bool,
+) *sendRequest {
+ sr := sendRequestPool.Get().(*sendRequest)
+ *sr = sendRequest{
+ pool: sendRequestPool,
+ ctx: ctx,
+ msg: msg,
+ producer: p,
+ callback: callback,
+ callbackOnce: &sync.Once{},
+ flushImmediately: flushImmediately,
+ publishTime: time.Now(),
+ chunkID: -1,
+ }
+ return sr
+}
+
+func newChunkSendRequest(p *sendRequest, chunkID int, uuid string, cr *chunkRecorder, reservedMem int64) *sendRequest {
+ sr := sendRequestPool.Get().(*sendRequest)
+ *sr = sendRequest{
+ pool: sendRequestPool,
+ ctx: p.ctx,
+ msg: p.msg,
+ producer: p.producer,
+ callback: p.callback,
+ callbackOnce: p.callbackOnce,
+ publishTime: p.publishTime,
+ flushImmediately: p.flushImmediately,
+ totalChunks: p.totalChunks,
+ chunkID: chunkID,
+ uuid: uuid,
+ chunkRecorder: cr,
+ transaction: p.transaction,
+ memLimit: p.memLimit,
+ semaphore: p.semaphore,
+ reservedMem: reservedMem,
+ sendAsBatch: p.sendAsBatch,
+ schema: p.schema,
+ schemaVersion: p.schemaVersion,
+ uncompressedPayload: p.uncompressedPayload,
+ uncompressedSize: p.uncompressedSize,
+ compressedPayload: p.compressedPayload,
+ compressedSize: p.compressedSize,
+ payloadChunkSize: p.payloadChunkSize,
+ mm: p.mm,
+ deliverAt: p.deliverAt,
+ maxMessageSize: p.maxMessageSize,
+ }
Review Comment:
`newChunkSendRequest` also uses `*sr = sendRequest{...}` to initialize a pooled object containing `atomic.Bool`. This both violates the atomic-copying constraint and can clear `doneFlag` to false during initialization, reintroducing the stale-pointer race the guard is meant to prevent. Initialize fields without overwriting the atomic field and only clear the guard (`doneFlag.Store(false)`) as the final step once the object is fully initialized.
##########
pulsar/producer_partition.go:
##########
@@ -1648,7 +1609,67 @@ type sendRequest struct {
maxMessageSize int32
}
+func newSendRequest(
+ ctx context.Context,
+ p *partitionProducer,
+ msg *ProducerMessage,
+ callback func(MessageID, *ProducerMessage, error),
+ flushImmediately bool,
+) *sendRequest {
+ sr := sendRequestPool.Get().(*sendRequest)
+ *sr = sendRequest{
+ pool: sendRequestPool,
+ ctx: ctx,
+ msg: msg,
+ producer: p,
+ callback: callback,
+ callbackOnce: &sync.Once{},
+ flushImmediately: flushImmediately,
+ publishTime: time.Now(),
+ chunkID: -1,
+ }
+ return sr
+}
+
+func newChunkSendRequest(p *sendRequest, chunkID int, uuid string, cr *chunkRecorder, reservedMem int64) *sendRequest {
+ sr := sendRequestPool.Get().(*sendRequest)
+ *sr = sendRequest{
+ pool: sendRequestPool,
+ ctx: p.ctx,
+ msg: p.msg,
+ producer: p.producer,
+ callback: p.callback,
+ callbackOnce: p.callbackOnce,
+ publishTime: p.publishTime,
+ flushImmediately: p.flushImmediately,
+ totalChunks: p.totalChunks,
+ chunkID: chunkID,
+ uuid: uuid,
+ chunkRecorder: cr,
+ transaction: p.transaction,
+ memLimit: p.memLimit,
+ semaphore: p.semaphore,
+ reservedMem: reservedMem,
+ sendAsBatch: p.sendAsBatch,
+ schema: p.schema,
+ schemaVersion: p.schemaVersion,
+ uncompressedPayload: p.uncompressedPayload,
+ uncompressedSize: p.uncompressedSize,
+ compressedPayload: p.compressedPayload,
+ compressedSize: p.compressedSize,
+ payloadChunkSize: p.payloadChunkSize,
+ mm: p.mm,
+ deliverAt: p.deliverAt,
+ maxMessageSize: p.maxMessageSize,
+ }
Review Comment:
`newSendRequest` reinitializes pooled objects with `*sr = sendRequest{...}`. Since `sendRequest` now contains an `atomic.Bool`, copying/overwriting it via struct assignment is unsafe (atomic types must not be copied after first use) and also drops the guard to false early during initialization, allowing stale-pointer `done()` calls to proceed if they race with pool reuse. Prefer preserving `doneFlag` as true while filling fields, then explicitly `Store(false)` as the last init step, without struct-assigning over the atomic field.
--
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