Skip to content

Fix batch acknowledgment for deserialized message IDs #359

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
MatheusReichert wants to merge 1 commit into
base: develop
Choose a base branch
from
Open

Fix batch acknowledgment for deserialized message IDs #359

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 33 additions & 9 deletions src/Pulsar.Client/Internal/ConsumerImpl.fs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
namespace Pulsar.Client.Api
namespace Pulsar.Client.Api

open System.Collections

Expand Down Expand Up @@ -260,6 +260,8 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien

let negativeAcksTracker = NegativeAcksTracker(prefix, consumerConfig.NegativeAckRedeliveryDelay, negativeAcksRedeliver)

let batchAckers = Dictionary<struct(LedgerId*EntryId), BatchMessageAcker>()

let getConnectionState() = connectionHandler.ConnectionState
let sendAckPayload (cnx: ClientCnx) payload = cnx.Send payload

Expand Down Expand Up @@ -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 =
Comment on lines 360 to 368
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 =
Expand All @@ -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
Expand All @@ -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
Comment on lines +415 to +422
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
Expand Down Expand Up @@ -583,6 +605,7 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien
interceptors.Close()
statTimer.Stop()
chunkTimer.Stop()
batchAckers.Clear()
cleanup(this)

let stopConsumer () =
Expand Down Expand Up @@ -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
Comment on lines 1406 to 1409
let stream = rawMessage.Payload
stream.Seek(0L, SeekOrigin.Begin) |> ignore
Expand Down
27 changes: 27 additions & 0 deletions tests/UnitTests/Common/MessageTests.fs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down
Loading