You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "dao-jun (via GitHub)" <gi...@apache.org> on 2024/05/14 06:28:08 UTC

[PR] [improve][broker] Optimize getLastDispatchablePosition [pulsar]

dao-jun opened a new pull request, #22707:
URL: https://github.com/apache/pulsar/pull/22707

   ### Motivation
   
   [PersistentTopic#getLastDispatchablePosition](https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java#L3776-L3788) is using by 
   [Reader#hasMessageAvailable](https://github.com/apache/pulsar/blob/master/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Reader.java#L116) , [ConsumerImpl#hasMessageAvailable](https://github.com/apache/pulsar/blob/master/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L2440-L2448), [Consumer#getLastMessageIdAsync](https://github.com/apache/pulsar/blob/master/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Consumer.java#L591-L615). 
   
   The current implementation is read entries from Bookkeeper(or sth else), which leads to low throughput, high latency and heavy load, this PR is for the purpose of optimization.
   
   ### Modifications
   
   <!-- Describe the modifications you've done. -->
   
   ### Verifying this change
   
   - [ ] Make sure that the change passes the CI checks.
   
   *(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:)*
     - *Added integration tests for end-to-end deployment with large payloads (10MB)*
     - *Extended integration test for recovery after broker failure*
   
   ### Does this pull request potentially affect one of the following parts:
   
   <!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
   
   *If the box was checked, please highlight the changes*
   
   - [ ] Dependencies (add or upgrade a dependency)
   - [ ] The public API
   - [ ] The schema
   - [ ] The default values of configurations
   - [ ] The threading model
   - [ ] The binary protocol
   - [ ] The REST endpoints
   - [ ] The admin CLI options
   - [ ] The metrics
   - [ ] Anything that affects deployment
   
   ### Documentation
   
   <!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
   
   - [ ] `doc` <!-- Your PR contains doc changes. -->
   - [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
   - [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
   - [ ] `doc-complete` <!-- Docs have been already added -->
   
   ### Matching PR in forked repository
   
   PR in forked repository: 
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] [improve][broker] Optimize PersistentTopic.getLastDispatchablePosition [pulsar]

Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #22707:
URL: https://github.com/apache/pulsar/pull/22707#issuecomment-2157397197

   ## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/22707?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
   Attention: Patch coverage is `73.17073%` with `11 lines` in your changes missing coverage. Please review.
   > Project coverage is 73.29%. Comparing base [(`bbc6224`)](https://app.codecov.io/gh/apache/pulsar/commit/bbc62245c5ddba1de4b1e7cee4ab49334bc36277?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`780000b`)](https://app.codecov.io/gh/apache/pulsar/commit/780000b99f0c26bee8604ba2517554003688db81?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
   > Report is 366 commits behind head on master.
   
   <details><summary>Additional details and impacted files</summary>
   
   
   [![Impacted file tree graph](https://app.codecov.io/gh/apache/pulsar/pull/22707/graphs/tree.svg?width=650&height=150&src=pr&token=acYqCpsK9J&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)](https://app.codecov.io/gh/apache/pulsar/pull/22707?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
   
   ```diff
   @@             Coverage Diff              @@
   ##             master   #22707      +/-   ##
   ============================================
   - Coverage     73.57%   73.29%   -0.29%     
   - Complexity    32624    33015     +391     
   ============================================
     Files          1877     1891      +14     
     Lines        139502   141969    +2467     
     Branches      15299    15571     +272     
   ============================================
   + Hits         102638   104051    +1413     
   - Misses        28908    29893     +985     
   - Partials       7956     8025      +69     
   ```
   
   | [Flag](https://app.codecov.io/gh/apache/pulsar/pull/22707/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
   |---|---|---|
   | [inttests](https://app.codecov.io/gh/apache/pulsar/pull/22707/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `27.21% <56.09%> (+2.62%)` | :arrow_up: |
   | [systests](https://app.codecov.io/gh/apache/pulsar/pull/22707/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `24.70% <39.02%> (+0.37%)` | :arrow_up: |
   | [unittests](https://app.codecov.io/gh/apache/pulsar/pull/22707/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `72.30% <73.17%> (-0.55%)` | :arrow_down: |
   
   Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
   
   | [Files](https://app.codecov.io/gh/apache/pulsar/pull/22707?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
   |---|---|---|
   | [...ransaction/buffer/impl/TopicTransactionBuffer.java](https://app.codecov.io/gh/apache/pulsar/pull/22707?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Ftransaction%2Fbuffer%2Fimpl%2FTopicTransactionBuffer.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci90cmFuc2FjdGlvbi9idWZmZXIvaW1wbC9Ub3BpY1RyYW5zYWN0aW9uQnVmZmVyLmphdmE=) | `88.21% <100.00%> (+0.46%)` | :arrow_up: |
   | [...sar/broker/service/persistent/PersistentTopic.java](https://app.codecov.io/gh/apache/pulsar/pull/22707?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2Fpersistent%2FPersistentTopic.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL3BlcnNpc3RlbnQvUGVyc2lzdGVudFRvcGljLmphdmE=) | `79.12% <90.90%> (+0.66%)` | :arrow_up: |
   | [...nsaction/buffer/impl/TransactionBufferDisable.java](https://app.codecov.io/gh/apache/pulsar/pull/22707?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Ftransaction%2Fbuffer%2Fimpl%2FTransactionBufferDisable.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci90cmFuc2FjdGlvbi9idWZmZXIvaW1wbC9UcmFuc2FjdGlvbkJ1ZmZlckRpc2FibGUuamF2YQ==) | `57.57% <71.42%> (+1.05%)` | :arrow_up: |
   | [...ransaction/buffer/impl/InMemTransactionBuffer.java](https://app.codecov.io/gh/apache/pulsar/pull/22707?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Ftransaction%2Fbuffer%2Fimpl%2FInMemTransactionBuffer.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci90cmFuc2FjdGlvbi9idWZmZXIvaW1wbC9Jbk1lbVRyYW5zYWN0aW9uQnVmZmVyLmphdmE=) | `54.85% <0.00%> (-2.72%)` | :arrow_down: |
   
   ... and [407 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/22707/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
   
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] [improve][broker] Optimize PersistentTopic.getLastDispatchablePosition [pulsar]

Posted by "dao-jun (via GitHub)" <gi...@apache.org>.
dao-jun commented on code in PR #22707:
URL: https://github.com/apache/pulsar/pull/22707#discussion_r1632581874


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java:
##########
@@ -307,6 +307,7 @@ public CompletableFuture<Void> commitTxn(TxnID txnID, long lowWaterMark) {
                 txnBuffer.commitAt(committedAtLedgerId, committedAtEntryId);
                 addTxnToTxnIdex(txnID, committedAtLedgerId);
             }
+            updateLastDispatchablePosition(null);

Review Comment:
   OK, resolved



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] [improve][broker] Optimize PersistentTopic.getLastDispatchablePosition [pulsar]

Posted by "dao-jun (via GitHub)" <gi...@apache.org>.
dao-jun commented on code in PR #22707:
URL: https://github.com/apache/pulsar/pull/22707#discussion_r1626825099


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java:
##########
@@ -377,8 +379,11 @@ public boolean isTxnAborted(TxnID txnID, PositionImpl readPosition) {
 
     @Override
     public void syncMaxReadPositionForNormalPublish(PositionImpl position, boolean isMarkerMessage) {
-        if (!isMarkerMessage && maxReadPositionCallBack != null) {
-            maxReadPositionCallBack.maxReadPositionMovedForward(null, position);
+        if (!isMarkerMessage) {
+            updateLastDispatchablePosition(position);

Review Comment:
   About this point, 
   ```java
          if (!isMarkerMessage) {
               updateLastDispatchablePosition(position);
               if (maxReadPositionCallBack != null) {
                   maxReadPositionCallBack.maxReadPositionMovedForward(null, position);
               }
           }
   ```
   I don't understand why this write approach is strange, could you please explain why?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] [improve][broker] Optimize PersistentTopic.getLastDispatchablePosition [pulsar]

Posted by "dao-jun (via GitHub)" <gi...@apache.org>.
dao-jun commented on code in PR #22707:
URL: https://github.com/apache/pulsar/pull/22707#discussion_r1626823163


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java:
##########
@@ -307,6 +307,7 @@ public CompletableFuture<Void> commitTxn(TxnID txnID, long lowWaterMark) {
                 txnBuffer.commitAt(committedAtLedgerId, committedAtEntryId);
                 addTxnToTxnIdex(txnID, committedAtLedgerId);
             }
+            updateLastDispatchablePosition(null);

Review Comment:
   > Another point is that the transaction in InMemTransactionBuffer  does not effect the lastDispatchablePosition.
   
   It is because of I don't want to introduce `Separated` solution to `InMemoryTxnBuffer`, just make it consistent with `TopicTxnBuffer`, let it easier to maintain. 



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] [improve][broker] Optimize PersistentTopic.getLastDispatchablePosition [pulsar]

Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #22707:
URL: https://github.com/apache/pulsar/pull/22707#discussion_r1623371207


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java:
##########
@@ -377,8 +379,11 @@ public boolean isTxnAborted(TxnID txnID, PositionImpl readPosition) {
 
     @Override
     public void syncMaxReadPositionForNormalPublish(PositionImpl position, boolean isMarkerMessage) {
-        if (!isMarkerMessage && maxReadPositionCallBack != null) {
-            maxReadPositionCallBack.maxReadPositionMovedForward(null, position);
+        if (!isMarkerMessage) {
+            updateLastDispatchablePosition(position);

Review Comment:
   This write approach is strange. Please follow the previous way.



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java:
##########
@@ -331,6 +332,7 @@ public CompletableFuture<Void> abortTxn(TxnID txnID, long lowWaterMark) {
             TxnBuffer txnBuffer = getTxnBufferOrThrowNotFoundException(txnID);
             txnBuffer.abort();
             buffers.remove(txnID, txnBuffer);
+            updateLastDispatchablePosition(null);

Review Comment:
   We should not update the position to null even if failed to commit a transaction.



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java:
##########
@@ -307,6 +307,7 @@ public CompletableFuture<Void> commitTxn(TxnID txnID, long lowWaterMark) {
                 txnBuffer.commitAt(committedAtLedgerId, committedAtEntryId);
                 addTxnToTxnIdex(txnID, committedAtLedgerId);
             }
+            updateLastDispatchablePosition(null);

Review Comment:
   Why do you update this position to null?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] [improve][broker] Optimize PersistentTopic.getLastDispatchablePosition [pulsar]

Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #22707:
URL: https://github.com/apache/pulsar/pull/22707#discussion_r1624635007


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java:
##########
@@ -307,6 +307,7 @@ public CompletableFuture<Void> commitTxn(TxnID txnID, long lowWaterMark) {
                 txnBuffer.commitAt(committedAtLedgerId, committedAtEntryId);
                 addTxnToTxnIdex(txnID, committedAtLedgerId);
             }
+            updateLastDispatchablePosition(null);

Review Comment:
   Especially for  `InMemTransactionBuffer.java.`



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] [improve][broker] Optimize PersistentTopic.getLastDispatchablePosition [pulsar]

Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #22707:
URL: https://github.com/apache/pulsar/pull/22707#discussion_r1626233626


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java:
##########
@@ -307,6 +307,7 @@ public CompletableFuture<Void> commitTxn(TxnID txnID, long lowWaterMark) {
                 txnBuffer.commitAt(committedAtLedgerId, committedAtEntryId);
                 addTxnToTxnIdex(txnID, committedAtLedgerId);
             }
+            updateLastDispatchablePosition(null);

Review Comment:
   Ok. More understand achieved.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] [improve][broker] Optimize PersistentTopic.getLastDispatchablePosition [pulsar]

Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #22707:
URL: https://github.com/apache/pulsar/pull/22707#discussion_r1624603252


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java:
##########
@@ -307,6 +307,7 @@ public CompletableFuture<Void> commitTxn(TxnID txnID, long lowWaterMark) {
                 txnBuffer.commitAt(committedAtLedgerId, committedAtEntryId);
                 addTxnToTxnIdex(txnID, committedAtLedgerId);
             }
+            updateLastDispatchablePosition(null);

Review Comment:
   I just promote a question here, right? There is no point in me arguing with you just to waste my time. What we do together in our spare time is make Pulsar better.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] [improve][broker] Optimize PersistentTopic.getLastDispatchablePosition [pulsar]

Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou merged PR #22707:
URL: https://github.com/apache/pulsar/pull/22707


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] [improve][broker] Optimize PersistentTopic.getLastDispatchablePosition [pulsar]

Posted by "dao-jun (via GitHub)" <gi...@apache.org>.
dao-jun commented on code in PR #22707:
URL: https://github.com/apache/pulsar/pull/22707#discussion_r1624343253


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java:
##########
@@ -307,6 +307,7 @@ public CompletableFuture<Void> commitTxn(TxnID txnID, long lowWaterMark) {
                 txnBuffer.commitAt(committedAtLedgerId, committedAtEntryId);
                 addTxnToTxnIdex(txnID, committedAtLedgerId);
             }
+            updateLastDispatchablePosition(null);

Review Comment:
   please see the changes carefully.



##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java:
##########
@@ -307,6 +307,7 @@ public CompletableFuture<Void> commitTxn(TxnID txnID, long lowWaterMark) {
                 txnBuffer.commitAt(committedAtLedgerId, committedAtEntryId);
                 addTxnToTxnIdex(txnID, committedAtLedgerId);
             }
+            updateLastDispatchablePosition(null);

Review Comment:
   please see the changes carefully. @liangyepianzhou 



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] [improve][broker] Optimize PersistentTopic.getLastDispatchablePosition [pulsar]

Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #22707:
URL: https://github.com/apache/pulsar/pull/22707#discussion_r1624604509


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java:
##########
@@ -307,6 +307,7 @@ public CompletableFuture<Void> commitTxn(TxnID txnID, long lowWaterMark) {
                 txnBuffer.commitAt(committedAtLedgerId, committedAtEntryId);
                 addTxnToTxnIdex(txnID, committedAtLedgerId);
             }
+            updateLastDispatchablePosition(null);

Review Comment:
   You record a maximum stable position that can be used for consumption, but you set this position to null when a transaction is committed. This behavior puzzles me.
   Because transaction committed will not change the previously recorded stable consumable position into a non-consumable position.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] [improve][broker] Optimize PersistentTopic.getLastDispatchablePosition [pulsar]

Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #22707:
URL: https://github.com/apache/pulsar/pull/22707#discussion_r1635176114


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java:
##########
@@ -459,6 +464,8 @@ void removeTxnAndUpdateMaxReadPosition(TxnID txnID) {
         } else {
             updateMaxReadPosition((PositionImpl) topic.getManagedLedger().getLastConfirmedEntry(), false);
         }
+        // Update the last dispatchable position to null if there is a TXN finished.
+        updateLastDispatchablePosition(null);

Review Comment:
   Oh, Sorry, I overlooked this. If you do not set it to null here, but set a flag that requires reading BK, you can reduce the response time of the next request.  I mentioned [it](https://github.com/apache/pulsar/pull/22707/#discussion_r1626241035) before, but maybe not in detail.
   
   1. end txn.
   2. set a flag 'readFromBK' as true
   3. update normal message
   4. set a flag 'readFromBK' as false
   5. when call `getLastDispatchablePosition`, if the flag == true, the `lastDispatchablePosition ` is valid to return. And read from BK after return.
   6. Call `getLastDispatchablePosition` again, it can get the last `getLastDispatchablePosition`.
   
   How do you think about this?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] [improve][broker] Optimize PersistentTopic.getLastDispatchablePosition [pulsar]

Posted by "dao-jun (via GitHub)" <gi...@apache.org>.
dao-jun commented on code in PR #22707:
URL: https://github.com/apache/pulsar/pull/22707#discussion_r1624839739


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java:
##########
@@ -307,6 +307,7 @@ public CompletableFuture<Void> commitTxn(TxnID txnID, long lowWaterMark) {
                 txnBuffer.commitAt(committedAtLedgerId, committedAtEntryId);
                 addTxnToTxnIdex(txnID, committedAtLedgerId);
             }
+            updateLastDispatchablePosition(null);

Review Comment:
   I introduced a filed `lastDispatchablePosition` to store the last position which is valid to dispatch, in `PersistentTopic`. And before return `lastDispatchablePosition`, check if `lastDispatchablePosition` is null, otherwise, search the position in BK.
   About why set `lastDispatchablePosition` to null when commit txn, it is because of it's hard to determine which position to set, so, set to null to make it fallback to search the position from BK.
   
   > Because transaction committed will not change the previously recorded stable consumable position into a
   > non-consumable position.
   
   I don't agree with this point, after commit a txn, the `lastDispatchablePosition` may have a chance to move forward, right? But, we don't know which position to set, and, even though we record the last entry position of the txn, what if non-txn messages published? 
   Say, `txn1` contains `1, 2, 3` messages, and we record all the positions. Before commit `txn1`, there is a new non-txn message `4` published, when commit `txn1`, what value can we set to `lastDispatchablePosition`? So I set `lastDispatchablePosition` to null, we can get the `lastDispatchablePosition` by searching BK.
   
   My  point is, the situations here is very complex, right? After commit a txn, the `lastDispatchablePosition` can be moved forward, or not. A topic contains `txn messages`, `non-txn messages`, `Marker`; `txn messages` contains `abort txn`, `ongoing txn`, `commit txn`. 
   Set `lastDispatchablePosition` to null after commit/abort txn is the easiest way to make it consistent, 



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] [improve][broker] Optimize PersistentTopic.getLastDispatchablePosition [pulsar]

Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #22707:
URL: https://github.com/apache/pulsar/pull/22707#discussion_r1626241035


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java:
##########
@@ -307,6 +307,7 @@ public CompletableFuture<Void> commitTxn(TxnID txnID, long lowWaterMark) {
                 txnBuffer.commitAt(committedAtLedgerId, committedAtEntryId);
                 addTxnToTxnIdex(txnID, committedAtLedgerId);
             }
+            updateLastDispatchablePosition(null);

Review Comment:
   >About why set lastDispatchablePosition to null when commit txn, it is because of it's hard to determine which position to set, so, set to null to make it fallback to search the position from BK.
   
   This is a good way for cache lastDispatchablePosition. There are some ways to do that. 
    1. your way
    2. When committing the transaction, keep the previous position because it is still the correct position. Then start a task to update, or mark it as needing to be updated from BK. Then the next time it is read, return to the previous position that is somewhat lagging but still valid. Then read from BK.
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] [improve][broker] Optimize getLastDispatchablePosition [pulsar]

Posted by "dao-jun (via GitHub)" <gi...@apache.org>.
dao-jun closed pull request #22707: [improve][broker] Optimize getLastDispatchablePosition
URL: https://github.com/apache/pulsar/pull/22707


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] [improve][broker] Optimize PersistentTopic.getLastDispatchablePosition [pulsar]

Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #22707:
URL: https://github.com/apache/pulsar/pull/22707#discussion_r1626242709


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java:
##########
@@ -307,6 +307,7 @@ public CompletableFuture<Void> commitTxn(TxnID txnID, long lowWaterMark) {
                 txnBuffer.commitAt(committedAtLedgerId, committedAtEntryId);
                 addTxnToTxnIdex(txnID, committedAtLedgerId);
             }
+            updateLastDispatchablePosition(null);

Review Comment:
   Another point is that the transaction in `InMemTransactionBuffer ` does not effect the `lastDispatchablePosition`.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] [improve][broker] Optimize PersistentTopic.getLastDispatchablePosition [pulsar]

Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #22707:
URL: https://github.com/apache/pulsar/pull/22707#discussion_r1635176114


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java:
##########
@@ -459,6 +464,8 @@ void removeTxnAndUpdateMaxReadPosition(TxnID txnID) {
         } else {
             updateMaxReadPosition((PositionImpl) topic.getManagedLedger().getLastConfirmedEntry(), false);
         }
+        // Update the last dispatchable position to null if there is a TXN finished.
+        updateLastDispatchablePosition(null);

Review Comment:
   Oh, Sorry, I overlooked this. If you do not set it to null here, but set a flag that requires reading BK, you can reduce the response time of the next request.  I mentioned [here](https://github.com/apache/pulsar/pull/22707/#discussion_r1626241035) before, but maybe not in detail.
   
   1. end txn.
   2. set a flag 'readFromBK' as true
   3. update normal message
   4. set a flag 'readFromBK' as false
   5. when call `getLastDispatchablePosition`, if the flag == true, the `lastDispatchablePosition ` is valid to return. And read from BK after return.
   6. Call `getLastDispatchablePosition` again, it can get the last `getLastDispatchablePosition`.
   
   How do you think about this?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] [improve][broker] Optimize PersistentTopic.getLastDispatchablePosition [pulsar]

Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #22707:
URL: https://github.com/apache/pulsar/pull/22707#discussion_r1632203580


##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java:
##########
@@ -307,6 +307,7 @@ public CompletableFuture<Void> commitTxn(TxnID txnID, long lowWaterMark) {
                 txnBuffer.commitAt(committedAtLedgerId, committedAtEntryId);
                 addTxnToTxnIdex(txnID, committedAtLedgerId);
             }
+            updateLastDispatchablePosition(null);

Review Comment:
   You should do nothing here.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


Re: [PR] [improve][broker] Optimize PersistentTopic.getLastDispatchablePosition [pulsar]

Posted by "dao-jun (via GitHub)" <gi...@apache.org>.
dao-jun commented on PR #22707:
URL: https://github.com/apache/pulsar/pull/22707#issuecomment-2114197101

   @coderzc PTAL


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org