-
Notifications
You must be signed in to change notification settings - Fork 3.5k
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.
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
[Transaction] Support consume transaction messages. #7781
Conversation
/pulsarbot run-failure-checks |
1 similar comment
/pulsarbot run-failure-checks |
private void read(ConcurrentLinkedQueue<TxnID> transactionQueue, | ||
int readMessageNum, Object ctx, | ||
AsyncCallbacks.ReadEntriesCallback readEntriesCallback) { | ||
final TxnID txnID = transactionQueue.peek(); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It's might peek null here.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Ok, I'll add a check.
private void read(ConcurrentLinkedQueue<TxnID> transactionQueue, | ||
int readMessageNum, Object ctx, | ||
AsyncCallbacks.ReadEntriesCallback readEntriesCallback) { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Is the read txn message count is the size of the transactionQueue
? I did not understand this method, is the transactionQueue only stores the first txn ID? I noticed you have only peek one item from this queue.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I'll adjust this logic.
87cb7dd
to
9f213b7
Compare
/pulsarbot run-failure-checks |
8bd38d9
to
5007a2a
Compare
/pulsarbot run-failure-checks |
Master Issue: apache#2664 Fix https://github.com/streamnative/pulsar/issues/1304 ### Motivation Currently, the consumer can't receive transaction messages. ### Modifications Support process the commit marker in the topic partition and fetch transaction messages from TransactionBuffer.
Master Issue: apache#2664 Fix https://github.com/streamnative/pulsar/issues/1304 ### Motivation Currently, the consumer can't receive transaction messages. ### Modifications Support process the commit marker in the topic partition and fetch transaction messages from TransactionBuffer.
Master Issue: apache#2664 Fix https://github.com/streamnative/pulsar/issues/1304 ### Motivation Currently, the consumer can't receive transaction messages. ### Modifications Support process the commit marker in the topic partition and fetch transaction messages from TransactionBuffer.
Master Issue: apache#2664 Fix https://github.com/streamnative/pulsar/issues/1304 ### Motivation Currently, the consumer can't receive transaction messages. ### Modifications Support process the commit marker in the topic partition and fetch transaction messages from TransactionBuffer.
Master Issue: #2664
Fix https://github.com/streamnative/pulsar/issues/1304
Motivation
Currently, the consumer can't receive transaction messages.
Modifications
Support process the commit marker in the topic partition and fetch transaction messages from TransactionBuffer.
Verifying this change
This change added tests and can be verified as follows:
org.apache.pulsar.broker.transaction.TransactionConsumeTest
Does this pull request potentially affect one of the following parts:
If
yes
was chosen, please highlight the changesDocumentation