From 1fd306731a076bec20705ceb9c18669a29e719b4 Mon Sep 17 00:00:00 2001 From: reichert Date: Thu, 11 Jun 2026 09:26:49 -0300 Subject: [PATCH] Fix batch acknowledgment for deserialized message IDs --- src/Pulsar.Client/Internal/ConsumerImpl.fs | 42 +++++++++++++++++----- tests/UnitTests/Common/MessageTests.fs | 27 ++++++++++++++ 2 files changed, 60 insertions(+), 9 deletions(-) diff --git a/src/Pulsar.Client/Internal/ConsumerImpl.fs b/src/Pulsar.Client/Internal/ConsumerImpl.fs index f6a45c37..2cf1765f 100644 --- a/src/Pulsar.Client/Internal/ConsumerImpl.fs +++ b/src/Pulsar.Client/Internal/ConsumerImpl.fs @@ -1,4 +1,4 @@ -namespace Pulsar.Client.Api +namespace Pulsar.Client.Api open System.Collections @@ -260,6 +260,8 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien let negativeAcksTracker = NegativeAcksTracker(prefix, consumerConfig.NegativeAckRedeliveryDelay, negativeAcksRedeliver) + let batchAckers = Dictionary() + let getConnectionState() = connectionHandler.ConnectionState let sendAckPayload (cnx: ClientCnx) payload = cnx.Send payload @@ -356,16 +358,23 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien let command = match messageId.Type with | 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 = match ackType with | Cumulative -> - batchAcker.AckCumulative(batchIndex) |> ignore - let bitSet = BitArray(batchAcker.GetBatchSize(), true) + acker.AckCumulative(batchIndex) |> ignore + let bitSet = BitArray(acker.GetBatchSize(), true) for i in 0 .. %batchIndex do bitSet[i] <- false bitSet | Individual -> - let bitSet = BitArray(batchAcker.GetBatchSize(), true) + let bitSet = BitArray(acker.GetBatchSize(), true) bitSet[%batchIndex] <- false bitSet let allBitsAreZero = @@ -382,7 +391,7 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien else ackSet |> toLongArray Commands.newAck consumerId messageId.LedgerId messageId.EntryId ackType properties adjustedSet - None (Some txnId) (Some requestId) (batchAcker.GetBatchSize() |> Some) + None (Some txnId) (Some requestId) (acker.GetBatchSize() |> Some) | _ -> Commands.newAck consumerId messageId.LedgerId messageId.EntryId ackType properties null None (Some txnId) (Some requestId) None @@ -402,10 +411,23 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien doTransactionAcknowledgeForResponse ackType properties txn.Id tcs messageId | None -> match messageId.Type with - | Batch batchDetails when not (markAckForBatchMessage messageId batchDetails ackType properties) -> - if consumerConfig.BatchIndexAcknowledgmentEnabled then - acksGroupingTracker.AddBatchIndexAcknowledgment(messageId, ackType, properties) - // other messages in batch are still pending ack. + | Batch batchDetails -> + 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 + if not (markAckForBatchMessage messageId struct(batchIndex, acker) ackType properties) then + if consumerConfig.BatchIndexAcknowledgmentEnabled then + acksGroupingTracker.AddBatchIndexAcknowledgment(messageId, ackType, properties) + else + batchAckers.Remove(key) |> ignore + sendAcknowledge messageId ackType properties + if Log.Logger.IsEnabled LogLevel.Debug then + Log.Logger.LogDebug("{0} acknowledged message - {1}, acktype {2}", prefix, messageId, ackType) | _ -> sendAcknowledge messageId ackType properties if Log.Logger.IsEnabled LogLevel.Debug then @@ -583,6 +605,7 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien interceptors.Close() statTimer.Stop() chunkTimer.Stop() + batchAckers.Clear() cleanup(this) let stopConsumer () = @@ -1382,6 +1405,7 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien default this.ReceiveIndividualMessagesFromBatch (rawMessage: RawMessage) schemaDecodeFunction isMessageUndecryptable = let batchSize = rawMessage.Metadata.NumMessages let acker = BatchMessageAcker(batchSize) + batchAckers[struct(rawMessage.MessageId.LedgerId, rawMessage.MessageId.EntryId)] <- acker let mutable skippedMessages = 0 let stream = rawMessage.Payload stream.Seek(0L, SeekOrigin.Begin) |> ignore diff --git a/tests/UnitTests/Common/MessageTests.fs b/tests/UnitTests/Common/MessageTests.fs index dcbdb0c9..f6d1c255 100644 --- a/tests/UnitTests/Common/MessageTests.fs +++ b/tests/UnitTests/Common/MessageTests.fs @@ -66,6 +66,33 @@ let tests = Expect.equal "" msgId deserialized } + test "Batch MessageId serialization acker isolation" { + let sharedAcker = BatchMessageAcker(2) + let msgId1 = { LedgerId = %1L; EntryId = %1L; Type = Batch(%0, sharedAcker); Partition = 1; TopicName = %""; ChunkMessageIds = None } + let msgId2 = { LedgerId = %1L; EntryId = %1L; Type = Batch(%1, sharedAcker); Partition = 1; TopicName = %""; ChunkMessageIds = None } + + let bytes1 = msgId1.ToByteArray() + let bytes2 = msgId2.ToByteArray() + + let deserialized1 = MessageId.FromByteArray bytes1 + let deserialized2 = MessageId.FromByteArray bytes2 + + match deserialized1.Type, deserialized2.Type with + | Batch (_, acker1), Batch (_, acker2) -> + // The deserialized ackers are different instances! + Expect.isFalse "Ackers must not be the same instance" (obj.ReferenceEquals(acker1, acker2)) + + // Let's ack individual indices + let acked1 = acker1.AckIndividual(%0) + let acked2 = acker2.AckIndividual(%1) + + // Neither one reports the whole batch is acknowledged + Expect.isFalse "Acker 1 should not be fully acked" acked1 + Expect.isFalse "Acker 2 should not be fully acked" acked2 + | _ -> + failwith "Deserialized message ID must be Batch type" + } + test "Message batching by count works correctly" { let messages = Messages(2, -1) let message = Message(MessageId.Earliest, [||], %"", false, EmptyProps, None, [||], %0L, [||], %0L, Nullable(), 0, "", "", Nullable(), fun () -> failwith "not implemented")