Fix batch acknowledgment for deserialized message IDs#359
Fix batch acknowledgment for deserialized message IDs#359MatheusReichert wants to merge 1 commit into
Conversation
There was a problem hiding this comment.
Pull request overview
This PR addresses a Pulsar batch-acknowledgment correctness issue when MessageIds from batched messages are serialized and later deserialized for manual acknowledgment, by introducing shared BatchMessageAcker resolution inside ConsumerImpl keyed by (LedgerId, EntryId).
Changes:
- Added
batchAckersmap to shareBatchMessageAckerinstances across deserializedMessageIds for the same batch entry. - Updated acknowledgment paths to resolve the shared acker from
batchAckers, and clear the map during consumer shutdown. - Added a unit test documenting that
MessageId(de)serialization yields isolatedBatchMessageAckerinstances.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 5 comments.
| File | Description |
|---|---|
| src/Pulsar.Client/Internal/ConsumerImpl.fs | Adds and uses a batchAckers map to coordinate batch-acker state across deserialized MessageIds; clears map on close; stores acker during batch parsing. |
| tests/UnitTests/Common/MessageTests.fs | Adds a unit test around batch MessageId (de)serialization acker-instance isolation. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| let struct(batchIndex, batchAcker) = batchDetails | ||
| let key = struct(messageId.LedgerId, messageId.EntryId) | ||
| let acker = | ||
| match batchAckers.TryGetValue(key) with | ||
| | true, sharedAcker -> sharedAcker | ||
| | false, _ -> | ||
| batchAckers[key] <- batchAcker | ||
| batchAcker |
| | Batch (batchIndex, batchAcker) -> | ||
| let key = struct(messageId.LedgerId, messageId.EntryId) | ||
| let acker = | ||
| match batchAckers.TryGetValue(key) with | ||
| | true, sharedAcker -> sharedAcker | ||
| | false, _ -> | ||
| batchAckers[key] <- batchAcker | ||
| batchAcker | ||
| let ackSet = |
| let batchSize = rawMessage.Metadata.NumMessages | ||
| let acker = BatchMessageAcker(batchSize) | ||
| batchAckers[struct(rawMessage.MessageId.LedgerId, rawMessage.MessageId.EntryId)] <- acker | ||
| let mutable skippedMessages = 0 |
|
|
||
| let negativeAcksTracker = NegativeAcksTracker(prefix, consumerConfig.NegativeAckRedeliveryDelay, negativeAcksRedeliver) | ||
|
|
||
| let batchAckers = Dictionary<struct(LedgerId*EntryId), BatchMessageAcker>() |
| Expect.equal "" msgId deserialized | ||
| } | ||
|
|
||
| test "Batch MessageId serialization acker isolation" { |
|
Hi, thanks for raising the issue, however given the referenced Java issue and unmerged PR, I'm not sure that this should be fixed at all, it rather looks like unsupported case by pulsar. |
See also apache/pulsar#19030 and apache/pulsar#19031
Motivation
Currently, when a message ID of a batched message is serialized (e.g., stored in a database or outbox envelope) and later deserialized to be acknowledged manually via
AcknowledgeAsync(orNegativeAcknowledge), the acknowledgment fails to propagate to the broker.This happens because the deserialized
MessageIdgets its own newly instantiated, isolated instance ofBatchMessageAcker. Since the individual messages in the batch no longer share the same acker instance, callingAckIndividualon each isolated acker never drives the outstanding count to0for any of the instances. Thus, the consumer never sends the final batch acknowledgment command to the broker, causing all messages in the batch to be redelivered once the acknowledgment timeout expires.This PR fixes the issue by introducing a
batchAckersmap insideConsumerImplto track and shareBatchMessageAckerinstances across deserialized message IDs associated with the same batch entry(LedgerId, EntryId).Modifications
batchAckersdictionary inConsumerImpl.fsto track activeBatchMessageAckers.batchAckersmap during batch parsing inReceiveIndividualMessagesFromBatch.trySendAcknowledgeanddoTransactionAcknowledgeForResponseto resolve the sharedBatchMessageAckerfrom the map.batchAckersdictionary incloseConsumerTasksto release resources and prevent memory leaks.MessageTests.fsto verify batch message ID acker isolation behavior during serialization and deserialization.