-
Notifications
You must be signed in to change notification settings - Fork 3.6k
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
[fix][broker] Avoid compaction task stuck when the last message to compact is a marker #21718
Conversation
pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java
Show resolved
Hide resolved
We need to cherry to 2.9 ~ 3.1 right ? |
Yes. |
Codecov Report
Additional details and impacted files@@ Coverage Diff @@
## master #21718 +/- ##
============================================
- Coverage 73.50% 73.33% -0.17%
+ Complexity 32818 32754 -64
============================================
Files 1893 1893
Lines 140721 140768 +47
Branches 15502 15512 +10
============================================
- Hits 103432 103228 -204
- Misses 29200 29427 +227
- Partials 8089 8113 +24
Flags with carried forward coverage won't be shown. Click here to find out more.
|
@@ -192,7 +195,7 @@ public int filterEntriesForConsumer(@Nullable MessageMetadata[] metadataArray, i | |||
} | |||
} | |||
|
|||
if (msgMetadata == null || Markers.isServerOnlyMarker(msgMetadata)) { | |||
if (msgMetadata == null || (Markers.isServerOnlyMarker(msgMetadata) && !Markers.isTxnMarker(msgMetadata))) { |
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 think we should pass all the markers and filter them at the client side when doing compaction.
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.
Fixed.
@@ -262,7 +266,10 @@ private void phaseTwoLoop(RawReader reader, MessageId to, Map<String, MessageId> | |||
MessageId id = m.getMessageId(); | |||
Optional<RawMessage> messageToAdd = Optional.empty(); | |||
mxBean.addCompactionReadOp(reader.getTopic(), m.getHeadersAndPayload().readableBytes()); | |||
if (RawBatchConverter.isReadableBatch(m)) { | |||
MessageMetadata metadata = Commands.parseMessageMetadata(m.getHeadersAndPayload()); | |||
if (Markers.isTxnMarker(metadata)) { |
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.
Why only check the txn marker here? Should we use Markers.isServerOnlyMarker(metadata)
?
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.
Yes, I will fix it.
@@ -1849,4 +1850,62 @@ public void testReadCommittedWithReadCompacted() throws Exception{ | |||
Assert.assertEquals(messages, List.of("V2", "V3")); | |||
} | |||
|
|||
|
|||
@Test | |||
public void testReadCommittedWithCompaction() throws Exception{ |
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.
Should we add a test to send a replicated subscription snapshot marker and test the compaction works?
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 add test testReplicatedSubscriptionWithCompaction
, PLAT~
56ba96a
to
55d929c
Compare
c4965c0
to
3ce2f31
Compare
…mpact is a marker (#21718)
…mpact is a marker (apache#21718) (cherry picked from commit 1f99568) Conflicts: pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java Fixed duplicate namespace created in transactionTest
…mpact is a marker (apache#21718) (cherry picked from commit 1f99568) Conflicts: pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java Fixed duplicate namespace created in transactionTest
…mpact is a marker (#21718)
Motivation
If the last message to compact is a marker, then the compaction task will stuck until timeout fails. The root cause is that the marker has been filtered out by the dispatcher, and the compaction reader will not read the last message.
Modifications
For
__compaction
internal cursor, we don't need to filter out marker messages and delete marker messages during compaction.Verifying this change
(Please pick either of the following options)
This change is a trivial rework / code cleanup without any test coverage.
(or)
This change is already covered by existing tests, such as (please describe tests).
(or)
This change added tests and can be verified as follows:
(example:)
Does this pull request potentially affect one of the following parts:
If the box was checked, please highlight the changes
Documentation
doc
doc-required
doc-not-needed
doc-complete
Matching PR in forked repository
PR in forked repository: