Skip to content
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

Jump to bottom

[issue 513] fix batch size limit validation #528

Merged
merged 4 commits into from
Oct 25, 2021
Merged

[issue 513] fix batch size limit validation #528

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
790977d
fix batch size limit validation
zzzming May 28, 2021
ece7731
respect either of one batch size and number of messages limit
zzzming May 28, 2021
379e584
correct hasSpace logic
zzzming May 29, 2021
6e18a3f
fix key-based batch builder
zzzming May 29, 2021
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
4 changes: 2 additions & 2 deletions pulsar/internal/batch_builder.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -157,7 +157,7 @@ func (bc *batchContainer) IsFull() bool {

func (bc *batchContainer) hasSpace(payload []byte) bool {
msgSize := uint32(len(payload))
return bc.numMessages > 0 && (bc.buffer.ReadableBytes()+msgSize) > uint32(bc.maxBatchSize)
return bc.numMessages+1 < bc.maxMessages || (bc.buffer.ReadableBytes()+msgSize) < uint32(bc.maxBatchSize)
}

// Add will add single message to batch.
Expand All @@ -174,7 +174,7 @@ func (bc *batchContainer) Add(
// There's already a message with cluster replication list. need to flush before next
// message can be sent
return false
} else if bc.hasSpace(payload) {
} else if !bc.hasSpace(payload) {
// The current batch is full. Producer has to call Flush() to
return false
}
Expand Down
4 changes: 2 additions & 2 deletions pulsar/internal/key_based_batch_builder.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -117,7 +117,7 @@ func (bc *keyBasedBatchContainer) IsMultiBatches() bool {

func (bc *keyBasedBatchContainer) hasSpace(payload []byte) bool {
msgSize := uint32(len(payload))
return bc.numMessages > 0 && (bc.buffer.ReadableBytes()+msgSize) > uint32(bc.maxBatchSize)
return bc.numMessages+1 < bc.maxMessages || (bc.buffer.ReadableBytes()+msgSize) < uint32(bc.maxBatchSize)
}

// Add will add single message to key-based batch with message key.
Expand All @@ -134,7 +134,7 @@ func (bc *keyBasedBatchContainer) Add(
// There's already a message with cluster replication list. need to flush before next
// message can be sent
return false
} else if bc.hasSpace(payload) {
} else if !bc.hasSpace(payload) {
// The current batch is full. Producer has to call Flush() to
return false
}
Expand Down