Skip to content

[issue 447] double check message size when client enable compress #806

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

Closed
cocktail828 wants to merge 3 commits into from
Closed

[issue 447] double check message size when client enable compress #806

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
6395bbc
double check message size when client enable compress
Jul 12, 2022
bcb0a78
Update pulsar/producer_partition.go
cocktail828 Jul 14, 2022
aba369d
check message before flush batchs to avoid compress twice
cocktail828 Jul 16, 2022
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
8 changes: 8 additions & 0 deletions pulsar/internal/batch_builder.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,9 @@ type BatchBuilder interface {
// IsFull check if the size in the current batch exceeds the maximum size allowed by the batch
IsFull() bool

// compress the payload
Compress(uncompressedPayload []byte) []byte

// Add will add single message to batch.
Add(
metadata *pb.SingleMessageMetadata, sequenceIDGenerator *uint64,
Expand Down Expand Up @@ -152,6 +155,11 @@ func NewBatchBuilder(
return &bc, nil
}

// compress the payload
func (bc *batchContainer) Compress(uncompressedPayload []byte) []byte {
return bc.compressionProvider.Compress(nil, uncompressedPayload)
}

// IsFull checks if the size in the current batch meets or exceeds the maximum size allowed by the batch
func (bc *batchContainer) IsFull() bool {
return bc.numMessages >= bc.maxMessages || bc.buffer.ReadableBytes() >= uint32(bc.maxBatchSize)
Expand Down
3 changes: 2 additions & 1 deletion pulsar/producer_partition.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -520,7 +520,8 @@ func (p *partitionProducer) internalSend(request *sendRequest) {
}

// if msg is too large
if len(payload) > int(p._getConn().GetMaxMessageSize()) {
if len(payload) > int(p._getConn().GetMaxMessageSize()) &&
Copy link
Member

@RobertIndie RobertIndie Jul 14, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Seems that in this PR we do the compression twice when sending the message. The batch builder will do the compression here: https://github.com/apache/pulsar-client-go/blob/master/pulsar/internal/commands.go#L251
Could we move the compression from the batch builder to here?

Copy link
Author

@cocktail828 cocktail828 Jul 15, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yep, this modification just an translation the implement of https://github.com/apache/pulsar/blob/782132561ac9fc8430ae3ef12913999e5871d3d2/pulsar-client-cpp/lib/ProducerImpl.cc#L431.
It indeed do compress twice.

// Wire format
// [TOTAL_SIZE] [CMD_SIZE][CMD] [MAGIC_NUMBER][CHECKSUM] [PAYLOAD]
// |
// compressed or uncompressed {[METADATA_SIZE_0] [METADATA_0] [PAYLOAD_0]} ...
Acording to the wire format above, SDK cannot compress messages one by one but should compress the whole batchs together.

Another way to avoid 'twice-compress', I think is flushing batchs immediately before and after add big message to batchBuilder.

len(p.batchBuilder.Compress(payload)) > int(p._getConn().GetMaxMessageSize()) {
p.publishSemaphore.Release()
request.callback(nil, request.msg, errMessageTooLarge)
p.log.WithError(errMessageTooLarge).
Expand Down