MarvinCai commented on a change in pull request #4265:
[transaction][acknowledge] Introduce PENDING_ACK state in acknowledgement path
URL: https://github.com/apache/pulsar/pull/4265#discussion_r286323417
##########
File path:
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
##########
@@ -76,13 +90,38 @@
// for connected subscriptions, message expiry will be checked if the
backlog is greater than this threshold
private static final int MINIMUM_BACKLOG_FOR_EXPIRY_CHECK = 1000;
+ // Map to keep track of message ack by each txn.
+ private final ConcurrentOpenHashMap<TxnID,
ConcurrentOpenHashSet<Position>> pendingAckMessagesMap;
+
+ // Messages acked by ongoing transaction, pending transaction commit to
materialize the acks. For faster look up.
+ // Using hashset as a message should only be acked once by one transaction.
+ private final ConcurrentOpenHashSet<Position> pendingAckMessages;
+
+ // Message cumulative acked by ongoing transaction, pending transaction
commit to materialize the ack.
+ // Only one transaction can cumulative ack.
+ // This parameter only keep the the largest Position it cumulative ack,as
any Position smaller will also be covered.
+ private volatile Position pendingCumulativeAckMessage;
+
+ private static final AtomicReferenceFieldUpdater<PersistentSubscription,
Position> POSITION_UPDATER =
+
AtomicReferenceFieldUpdater.newUpdater(PersistentSubscription.class,
Position.class,
+ "pendingCumulativeAckMessage");
+
+ // Transaction currently using cumulative ack.
+ private volatile TxnID txnID;
Review comment:
agreed, that would be a more informative name.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
