[fix][broker] Reader stuck after call hasMessageAvailable when enable replicateSubscriptionState by shibd · Pull Request #22572 · apache/pulsar

shibd · 2024-04-24T09:14:28Z

Motivation

Analysis

When enabling replicateSubscriptionState will use topic to sync subscription state, and make these message metadata as Marker

These marker messages will not be sent to the consumer by the topic, and will automatically ack them.

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java

Lines 202 to 217 in fc393f6

    
           if (Markers.isReplicatedSubscriptionSnapshotMarker(msgMetadata)) { 
        
               final int readerIndex = metadataAndPayload.readerIndex(); 
        
               processReplicatedSubscriptionSnapshot(pos, metadataAndPayload); 
        
               metadataAndPayload.readerIndex(readerIndex); 
        
           } 
        
           // Deliver marker to __compaction cursor to avoid compaction task stuck, 
        
           // and filter out them when doing topic compaction. 
        
           if (msgMetadata == null || cursor == null 
        
                   || !cursor.getName().equals(Compactor.COMPACTION_SUBSCRIPTION)) { 
        
               entries.set(i, null); 
        
               entry.release(); 
        
               individualAcknowledgeMessageIfNeeded(Collections.singletonList(pos), 
        
                       Collections.emptyMap()); 
        
               continue; 
        
           }

But getLastMessageId will always return the last message position, regardless of whether the last message is marked or not. This will cause the reader stuck.

        while (reader.hasMessageAvailable()) {  // get true
              Message message reader.readNext();  // never can't receive msg.
        }

You can refer to this diagram to help understand this bug:

Modifications

Add asyncReverseFindPositionOneByOne method on ManagedLedger .
Add getLastCanDispatchPosition method on Topic, it will call asyncReverseFindPositionOneByOne to find the last position of entry that not is replistateSubscriptionState
Change the getLastMessageId implement of ServerCnx to use getLastCanDispatchPosition instead of getMaxReadPosition.

Verifying this change

Add ManagedLedgerTest.testReverseFindPositionOneByOne to cover ReverseFindPositionOneByOne method.
Add testReplicatedSubscriptionAcrossTwoRegionsGetLastMessage to cover this bug.

Documentation

doc
doc-required
doc-not-needed
doc-complete

Matching PR in forked repository

dao-jun

Overall looks good to me, just left some minor comments about the code style

… replicateSubscriptionState

coderzc

LGTM

codecov-commenter · 2024-04-28T11:56:01Z

Codecov Report

Attention: Patch coverage is 80.00000% with 9 lines in your changes are missing coverage. Please review.

Project coverage is 74.13%. Comparing base (bbc6224) to head (f094e04).
Report is 203 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #22572      +/-   ##
============================================
+ Coverage     73.57%   74.13%   +0.56%     
+ Complexity    32624     2747   -29877     
============================================
  Files          1877     1886       +9     
  Lines        139502   140653    +1151     
  Branches      15299    15462     +163     
============================================
+ Hits         102638   104278    +1640     
+ Misses        28908    28331     -577     
- Partials       7956     8044      +88

Flag	Coverage Δ
inttests	`27.23% <57.77%> (+2.64%)`	⬆️
systests	`24.48% <57.77%> (+0.15%)`	⬆️
unittests	`73.44% <80.00%> (+0.59%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files	Coverage Δ
...sar/broker/service/persistent/PersistentTopic.java	`78.66% <100.00%> (+0.20%)`	⬆️
...va/org/apache/pulsar/broker/service/ServerCnx.java	`72.47% <93.33%> (+0.33%)`	⬆️
...n/java/org/apache/pulsar/broker/service/Topic.java	`34.78% <0.00%> (-1.59%)`	⬇️
...ookkeeper/mledger/util/ManagedLedgerImplUtils.java	`69.56% <69.56%> (ø)`

... and 278 files with indirect coverage changes

… replicateSubscriptionState (apache#22572) (cherry picked from commit a761b97)

… replicateSubscriptionState (#22572) (cherry picked from commit a761b97)

dao-jun · 2024-04-28T14:24:55Z

The PR handled the case of ServerOnlyMarker, but it looks we also need to handle txn aborted messages.
I created a PR for improvement, PTAL
#22610

… replicateSubscriptionState (apache#22572) (cherry picked from commit a761b97) (cherry picked from commit 1dacca5)

… replicateSubscriptionState (apache#22572)

…osition implementation for compaction (streamnative#606) ### Motivation apache/pulsar#22891 introduces some changes for the PositionImpl. Now we need to use the Position interface instead of the PositionImpl. ### Modifications - Refactor to use Position instead of PositionImpl - apache/pulsar#22572 uses `getLastDispatchablePosition` instead of `getMaxPosition` to return the last message id, this patch implements the new method for `KopPersistentTopic`. - apache/pulsar#22838 changed the result for `checkTopicExists`. This PR also adapts to changes. - apache/pulsar#22882 removed the method `org.apache.pulsar.broker.service.Producer.updateRates(int numOfMessages, long msgSizeInBytes)`. We need to use `producer.getStats().recordMsgIn` instead. --------- Co-authored-by: Yunze Xu <xyzinfernity@163.com>

shibd requested review from Technoboy-, codelipenghui, congbobo184, liangyepianzhou and poorbarcode April 24, 2024 09:14

shibd self-assigned this Apr 24, 2024

shibd added type/bug The PR fixed a bug or issue reported a bug ready-to-test labels Apr 24, 2024

github-actions Bot added the doc-not-needed Your PR changes do not impact docs label Apr 24, 2024

lhotari reviewed Apr 24, 2024

View reviewed changes

Comment thread managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java Outdated

dao-jun reviewed Apr 24, 2024

View reviewed changes

Comment thread ...r/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java Outdated

dao-jun approved these changes Apr 25, 2024

View reviewed changes

dao-jun reviewed Apr 25, 2024

View reviewed changes

Comment thread ...r/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java Outdated

shibd requested a review from lhotari April 25, 2024 07:22

shibd force-pushed the geo_replication branch from 6431d8c to c396054 Compare April 26, 2024 00:05

codelipenghui reviewed Apr 28, 2024

View reviewed changes

coderzc reviewed Apr 28, 2024

View reviewed changes

Comment thread pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java Outdated

shibd force-pushed the geo_replication branch from c396054 to 99032e8 Compare April 28, 2024 05:13

[fix][broker] Reader stuck after call hasMessageAvailable when enable…

d78603f

… replicateSubscriptionState

shibd force-pushed the geo_replication branch from 99032e8 to d78603f Compare April 28, 2024 05:13

shibd requested review from BewareMyPower, RobertIndie, codelipenghui, coderzc and dao-jun April 28, 2024 05:20

codelipenghui approved these changes Apr 28, 2024

View reviewed changes

codelipenghui reviewed Apr 28, 2024

View reviewed changes

Comment thread pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorSubscriptionTest.java Outdated

coderzc reviewed Apr 28, 2024

View reviewed changes

Comment thread pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

shibd added 2 commits April 28, 2024 15:13

address code reivews.

ab3a433

Address code reviews.

f094e04

shibd requested a review from coderzc April 28, 2024 07:42

coderzc approved these changes Apr 28, 2024

View reviewed changes

codelipenghui merged commit a761b97 into apache:master Apr 28, 2024

codelipenghui added release/2.10.7 release/2.11.5 release/3.2.3 release/3.0.5 labels Apr 28, 2024

codelipenghui pushed a commit to codelipenghui/incubator-pulsar that referenced this pull request Apr 28, 2024

[fix][broker] Reader stuck after call hasMessageAvailable when enable…

b1f1580

… replicateSubscriptionState (apache#22572) (cherry picked from commit a761b97)

codelipenghui pushed a commit to codelipenghui/incubator-pulsar that referenced this pull request Apr 28, 2024

[fix][broker] Reader stuck after call hasMessageAvailable when enable…

1dacca5

… replicateSubscriptionState (apache#22572) (cherry picked from commit a761b97)

codelipenghui pushed a commit that referenced this pull request Apr 28, 2024

[fix][broker] Reader stuck after call hasMessageAvailable when enable…

ad00315

… replicateSubscriptionState (#22572) (cherry picked from commit a761b97)

codelipenghui added cherry-picked/branch-2.11 cherry-picked/branch-3.0 cherry-picked/branch-3.2 labels Apr 28, 2024

dao-jun mentioned this pull request Apr 28, 2024

[fix] Fix Reader can be stuck from transaction aborted messages. #22610

Merged

15 tasks

Demogorgon314 reviewed Apr 29, 2024

View reviewed changes

Comment thread managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/ManagedLedgerImplUtils.java

lhotari reviewed Apr 29, 2024

View reviewed changes

Comment thread managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/ManagedLedgerImplUtils.java

Comment thread managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/ManagedLedgerImplUtils.java

dao-jun mentioned this pull request May 19, 2024

Reader stuck after reading all the messages on a topic that were each published via a separate transaction. #11877

Closed

hanmz pushed a commit to hanmz/pulsar that referenced this pull request Feb 12, 2025

[fix][broker] Reader stuck after call hasMessageAvailable when enable…

4dfb742

… replicateSubscriptionState (apache#22572)

	if (Markers.isReplicatedSubscriptionSnapshotMarker(msgMetadata)) {
	final int readerIndex = metadataAndPayload.readerIndex();
	processReplicatedSubscriptionSnapshot(pos, metadataAndPayload);
	metadataAndPayload.readerIndex(readerIndex);
	}

	// Deliver marker to __compaction cursor to avoid compaction task stuck,
	// and filter out them when doing topic compaction.
	if (msgMetadata == null \|\| cursor == null
	\|\| !cursor.getName().equals(Compactor.COMPACTION_SUBSCRIPTION)) {
	entries.set(i, null);
	entry.release();
	individualAcknowledgeMessageIfNeeded(Collections.singletonList(pos),
	Collections.emptyMap());
	continue;
	}

Conversation

shibd commented Apr 24, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation

Analysis

Modifications

Verifying this change

Documentation

Matching PR in forked repository

Uh oh!

Uh oh!

Uh oh!

dao-jun left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

coderzc left a comment

Choose a reason for hiding this comment

Uh oh!

codecov-commenter commented Apr 28, 2024

Codecov Report

Uh oh!

dao-jun commented Apr 28, 2024

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

9 participants

shibd commented Apr 24, 2024 •

edited

Loading