[Issue 456] Support chunking for big messages. #805

Gleiphir2769 · 2022-07-10T08:02:13Z

Contribution Checklist

Master Issue: #456

Motivation

Make pulsar go client support chunking to produce/consume big messages. The earlier implementation (#717) didn't take into account many details, so I decided to reimplement it.

Modifications

Add internalSingleSend to send message without batch because batch message will not be received by chunk.
Moved BlockIfQueueFull check from internalSendAsync to internalSend (canAddQueue) to ensure the normal block in chunking.
Make producer send big messages by chunking.
Add chunkedMsgCtxMap to store chunked messages meta and data.
Make consumer can obtain chunks and consume the big message.

Verifying this change

Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

Add TestProducerChunking to verify send big message by chunking.
Add message_chunking_test to verify message chunking.

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

Dependencies (does it add or upgrade a dependency): no
The public API: no
The schema: no
The default values of configurations: no
The wire protocol: no

Documentation

Does this pull request introduce a new feature? yes
If yes, how is the feature documented? not yet

zzzming

Based on Java implementation, ConsumerImpl.java, Chunking also requires changes at the consumer side to be able to assembly chunks into the original message. Are you going to add support on the consumer side?

Gleiphir2769 · 2022-07-27T14:07:41Z

Based on Java implementation, ConsumerImpl.java, Chunking also requires changes at the consumer side to be able to assembly chunks into the original message. Are you going to add support on the consumer side?

The consumer side implement is already planned. It will be commited in another PR.

…g of sequenceID generate; fix the bug where batch messages are compressed twice; add internal error handle, fix inaccurate PublishErrorsMsgTooLarge metric; fix the incorrect producer queueFullBlock

…ecause it's no need

Gleiphir2769 · 2022-08-06T08:28:47Z

Based on Java implementation, ConsumerImpl.java, Chunking also requires changes at the consumer side to be able to assembly chunks into the original message. Are you going to add support on the consumer side?

Hi, the consumer side is implemented here. Looking forward to your review. @zzzming

…atchIDX to pass test. Add the lisence headers to timewheel.go/timewheel_test.go/message_chunking_test.go

pulsar/producer.go

pulsar/producer_partition.go

…e. 2. Add the unit test for chunk size. 3. fix some other bugs.

pulsar/consumer.go

…exConsumer

…he chunkMessageID.Serialize

Gleiphir2769 · 2022-10-16T03:27:25Z

This PR will be possible to fix this issue #447.

pulsar/consumer_partition.go

Gleiphir2769 · 2022-10-24T02:20:32Z

/pulsarbot run-failure-checks

nodece

Your PR seems to have made improvements unrelated to chunking, if so, I suggest you make a new PR to improve.

nodece · 2022-10-24T02:31:43Z

pulsar/consumer_impl.go

-
-	if mid.consumer != nil {
-		return mid.Ack()
+	if err := c.checkMsgIDPartition(msgID); err != nil {


Missed covert the msgID from the MessageID to the trackingMessageID type, I'm not sure if we need this.

Why not use messageID(), what did I miss?

trackingMessageID does not records chunking infomation.

For example, Ack() a big message need to ack all the chunks of it. Using trackingMessageID can not figure out which chunk (messageId) need to be ack.

trackingMessageID is designed to tracking batch messages so it shoud not be the messageId type accepted by the method exposed by partitionConsumer. I think the better way would be to accept MessageID as the messageId type in partitionConsumer methods. However, only the necessary interfaces have been modified (Ack, NAck and Seed) for the least changes

pulsar/consumer_impl.go

pulsar/consumer_partition.go

nodece · 2022-10-24T04:45:39Z

pulsar/internal/batch_builder.go

@@ -164,7 +167,11 @@ func (bc *batchContainer) hasSpace(payload []byte) bool {
 		return true
 	}
 	msgSize := uint32(len(payload))
-	return bc.numMessages+1 <= bc.maxMessages && bc.buffer.ReadableBytes()+msgSize <= uint32(bc.maxBatchSize)
+	expectedSize := bc.buffer.ReadableBytes() + msgSize


Please revert this code(170-174).

L171 compared the bc.buffer.ReadableBytes() + msgSize with bc.maxMessageSize and the original code does not.

It used to make sure that one batch size does not exceed the maxMessageSize. It's a part of correctly calculation whether the message is too large.

By the way, the compare code is too long to be inline.

Good catch, but using inline looks like more clear, so like:

return bc.numMessages+1 <= bc.maxMessages && expectedSize <= uint32(bc.maxBatchSize) && expectedSize <= bc.maxMessageSize

@Gleiphir2769

@nodece This PR introduces the maxMessageSize. I think these changes should be related to this PR.

Good catch, but using inline looks like more clear, so like:

return bc.numMessages+1 <= bc.maxMessages && expectedSize <= uint32(bc.maxBatchSize) && expectedSize <= bc.maxMessageSize

Done, thx.

pulsar/producer_partition.go

Gleiphir2769 · 2022-10-24T09:57:46Z

/pulsarbot run-failure-checks

maraiskruger1980 · 2023-05-08T09:30:34Z

Does the chunking work on shared subscription introduced in Pulsar 2.11?

Gleiphir2769 · 2023-05-08T13:09:12Z

Does the chunking work on shared subscription introduced in Pulsar 2.11

Hi @maraiskruger1980, This PR is finished when pulsar 2.11 has not been released. So it doesn't support shared subscription chunking.

I think I can take some time on it. Welcome to follow the progress.

maraiskruger1980 · 2023-05-08T13:59:42Z

That will be great if it can support shared subscription

Gleiphir2769 · 2023-07-08T07:15:41Z

That will be great if it can support shared subscription

Hi @maraiskruger1980. After I checked, it does no limit in consumer when subscription is shared. Which means you can safely consume chunking messages in shared subscribtion if your pulsar version >= 2.11.

Gleiphir2769 added 3 commits July 3, 2022 21:50

feat: add internalSingleSend to send single message

cd183c2

feat: support chunking for producer

342e8e5

Merge branch 'master' into chunking_issue_456

ff60f94

Gleiphir2769 mentioned this pull request Jul 14, 2022

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

Closed

zzzming reviewed Jul 26, 2022

View reviewed changes

Gleiphir2769 added 3 commits August 6, 2022 15:48

fix: fix the bug of Send() block when SingleSend() failed; fix the bu…

3a90ce5

…g of sequenceID generate; fix the bug where batch messages are compressed twice; add internal error handle, fix inaccurate PublishErrorsMsgTooLarge metric; fix the incorrect producer queueFullBlock

feat: Implemented chunking on the consumer side

ed7915d

test: add message_chunking_test.go for chunk consume/produce test

0456630

Gleiphir2769 force-pushed the chunking_issue_456 branch from bf39e92 to 0456630 Compare August 6, 2022 07:58

refector: do some clean, removed unAckedChunkedMessageIdSequenceMap b…

a4769e8

…ecause it's no need

Gleiphir2769 changed the title ~~[Issue 456] Support chunking to produce big messages.~~ [Issue 456] Support chunking for big messages. Aug 6, 2022

Gleiphir2769 added 7 commits August 10, 2022 15:15

fix: fix the bug of uuid generation

64b9a20

fix: fix the Seek() for chunk message id

c8e083a

test: move chunk tests to message_chunking_test

7cc168f

refactor: add licence header

db7537f

fix: Revert the change about trackingMessageID. Fix the conflict of b…

2537cf4

…atchIDX to pass test. Add the lisence headers to timewheel.go/timewheel_test.go/message_chunking_test.go

fix: fix the license header of timewheel.go.

4dc9bb7

fix: fix the bug of data race for availablePermits

0db443e

RobertIndie reviewed Sep 2, 2022

View reviewed changes

Gleiphir2769 added 3 commits September 6, 2022 21:01

fix: 1. fix the comments of EnableChunking and rename the MaxChunkSiz…

92fbff2

…e. 2. Add the unit test for chunk size. 3. fix some other bugs.

fix: 1. fix the bug of seq ID. 2. make code pass golangci

d23d1cd

fix: fix the test function TestChunkSize()

5e098d2

nodece reviewed Sep 9, 2022

View reviewed changes

pulsar/consumer.go Outdated Show resolved Hide resolved

Gleiphir2769 added 3 commits September 9, 2022 15:45

fix: revert negativeAcksTracker.Add()

203a1fc

fix: modify the comment of ExpireTimeOfIncompleteChunk

2a1342d

fix: fix the ack/nack for chunking

fda6997

RobertIndie assigned Gleiphir2769 Sep 13, 2022

fix: fix the chunking message ack/nack for multiTopicConsumer and Reg…

7fd7ffa

…exConsumer

Gleiphir2769 added 3 commits September 30, 2022 15:14

fix: remove the ExpireTimeOfIncompleteChunk limit and do some refactor

3aeb764

fix: fix the nil pointer error of discardOldestChunkMessage and fix t…

4b53e33

…he chunkMessageID.Serialize

update chunking_issue_456 and resolve conflict

50f4959

Gleiphir2769 added 4 commits October 16, 2022 11:49

test: restart workflow

e80fe90

test: restart workflow

84c02fe

test: restart workflow

23570cf

test: restart workflow

56ef453

RobertIndie reviewed Oct 21, 2022

View reviewed changes

pulsar/consumer_partition.go Show resolved Hide resolved

fix: add logs when chunkMsgCtxMap has closed

a42de10

RobertIndie approved these changes Oct 24, 2022

View reviewed changes

nodece requested changes Oct 24, 2022

View reviewed changes

Gleiphir2769 added 3 commits October 24, 2022 15:45

fix: modify by review comment

b3b304f

fix: inline compare and fix a data race error

6a801cb

fix: fix nil pointer error

00e9e83

nodece approved these changes Oct 24, 2022

View reviewed changes

nodece requested review from wolfstudy and merlimat October 24, 2022 09:50

RobertIndie merged commit 77c7ccb into apache:master Oct 25, 2022

flowchartsman mentioned this pull request Dec 2, 2022

Compression broken if batching is off. #899

Closed

Gleiphir2769 mentioned this pull request Dec 6, 2022

[Issue 899] Fix the compression broken when batching is disabled #902

Merged

1 task

RobertIndie added this to the v0.10.0 milestone Mar 27, 2023

ericsyh mentioned this pull request Aug 18, 2023

[feat][sink][pulsar]: add chunk config in the pulsar sink to support large message pingcap/tiflow#9598

Open

geniusjoe mentioned this pull request Jul 11, 2024

Support Large Message Size (PIP-37 / "chunking") #456

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Issue 456] Support chunking for big messages. #805

[Issue 456] Support chunking for big messages. #805

Gleiphir2769 commented Jul 10, 2022 •

edited by RobertIndie

Loading

zzzming left a comment

Gleiphir2769 commented Jul 27, 2022

Gleiphir2769 commented Aug 6, 2022

Gleiphir2769 commented Oct 16, 2022 •

edited

Loading

Gleiphir2769 commented Oct 24, 2022

nodece left a comment

nodece Oct 24, 2022

Gleiphir2769 Oct 24, 2022

nodece Oct 24, 2022

Gleiphir2769 Oct 24, 2022

nodece Oct 24, 2022 •

edited

Loading

nodece Oct 24, 2022

RobertIndie Oct 24, 2022

Gleiphir2769 Oct 24, 2022 •

edited

Loading

Gleiphir2769 commented Oct 24, 2022

maraiskruger1980 commented May 8, 2023

Gleiphir2769 commented May 8, 2023

maraiskruger1980 commented May 8, 2023

Gleiphir2769 commented Jul 8, 2023

[Issue 456] Support chunking for big messages. #805

[Issue 456] Support chunking for big messages. #805

Conversation

Gleiphir2769 commented Jul 10, 2022 • edited by RobertIndie Loading

Contribution Checklist

Motivation

Modifications

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

zzzming left a comment

Choose a reason for hiding this comment

Gleiphir2769 commented Jul 27, 2022

Gleiphir2769 commented Aug 6, 2022

Gleiphir2769 commented Oct 16, 2022 • edited Loading

Gleiphir2769 commented Oct 24, 2022

nodece left a comment

Choose a reason for hiding this comment

nodece Oct 24, 2022

Choose a reason for hiding this comment

Gleiphir2769 Oct 24, 2022

Choose a reason for hiding this comment

nodece Oct 24, 2022

Choose a reason for hiding this comment

Gleiphir2769 Oct 24, 2022

Choose a reason for hiding this comment

nodece Oct 24, 2022 • edited Loading

Choose a reason for hiding this comment

nodece Oct 24, 2022

Choose a reason for hiding this comment

RobertIndie Oct 24, 2022

Choose a reason for hiding this comment

Gleiphir2769 Oct 24, 2022 • edited Loading

Choose a reason for hiding this comment

Gleiphir2769 commented Oct 24, 2022

maraiskruger1980 commented May 8, 2023

Gleiphir2769 commented May 8, 2023

maraiskruger1980 commented May 8, 2023

Gleiphir2769 commented Jul 8, 2023

Gleiphir2769 commented Jul 10, 2022 •

edited by RobertIndie

Loading

Gleiphir2769 commented Oct 16, 2022 •

edited

Loading

nodece Oct 24, 2022 •

edited

Loading

Gleiphir2769 Oct 24, 2022 •

edited

Loading