dferstay commented on a change in pull request #694:
URL: https://github.com/apache/pulsar-client-go/pull/694#discussion_r776843156
##########
File path: pulsar/default_router.go
##########
@@ -80,32 +79,27 @@ func NewDefaultRouter(
// spread the data on different partitions but not necessarily
in a specific sequence.
var now int64
size := uint32(len(message.Payload))
- previousMessageCount := atomic.LoadUint32(&state.msgCounter)
- previousBatchingMaxSize :=
atomic.LoadUint32(&state.cumulativeBatchSize)
- previousLastChange :=
atomic.LoadInt64(&state.lastChangeTimestamp)
+ messageCount := atomic.AddUint32(&state.msgCounter, 1)
+ batchSize := atomic.AddUint32(&state.cumulativeBatchSize, size)
- messageCountReached := previousMessageCount >=
uint32(maxBatchingMessages-1)
- sizeReached := (size >=
uint32(maxBatchingSize)-previousBatchingMaxSize)
+ messageCountReached := messageCount%uint32(maxBatchingMessages)
== 0
Review comment:
@cckellogg ,
The initial approach taken in this commit was to only modify the
`messageCount` in the router state by using an atomic increment in order to
avoid races between when the counter is modified (on a new message) and when
the counter is reset (when the batch is completed). By doing this the count is
now a running total of all messages that have passed through the router, so we
use modulo arithmetic to determine when the `maxBatchingMessages` has been
reached.
--
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]
