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-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
