[transaction][acknowledge] Introduce in-memory PENDING_ACK state in acknowledgement path #4265

MarvinCai · 2019-05-12T21:25:08Z

Master Issue: #2664

Motivation:
Add acknowledgeMessage, commit, abort for transaction in PersistentSubscription.

Changes:
Will put message in Pending_ACK status when acknowledgeMessage is called with TxnID.
No real status class introduced, only added collection to hold messages in Pending_ACK status.
Current PR only keep Pending_ACK state in memory, in subsequent PR will also persistent these pending acks so we can recover from broker failure.

Add commitTxn to put message to Deleted status.
Add abortTxn to put message to Pending status.

For normal acknowledgeMessage and redeliverUnacknowledgedMessages, will check to see if
message if message is in Pending_ACK first. If true, will ignore those acks/redeliverys.

Add unit test.

MarvinCai · 2019-05-12T21:28:01Z

@sijie @Tuten @zymap Can you help review this pr.

sijie · 2019-05-14T20:38:16Z

@MarvinCai thank you for picking this up! Great work!

I am traveling this week. I will probably get to it late this week.

sijie

@MarvinCai overall looks great!

I left some comments. PTAL

sijie · 2019-05-20T01:37:35Z

pulsar-broker/pom.xml

@@ -122,6 +128,12 @@
      <version>${project.version}</version>
    </dependency>

+    <dependency>


is this a duplicated dependency?

Yes, didn't see there's already one for pulsar-transaction-common, removed this one.

sijie · 2019-05-20T01:39:20Z

...broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

+                    "pendingCumulativeAckMessage");
+
+    // Transaction currently using cumulative ack.
+    private volatile TxnID txnID;


a better name might be pendingCumulativeAckTxnId?

agreed, that would be a more informative name.

sijie · 2019-05-20T01:46:59Z

...broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

+            if (this.txnID != null) {
+                log.warn("[{}][{}] An ongoing transaction:{}{} is doing cumulative ack, " +
+                                "new cumulative ack is not allowed till the transaction is committed.",
+                        this.txnID.getMostSigBits(), this.txnID.getLeastSigBits(), topicName, subName);


nit: can we just have a toString implementation on TxnID? so we don't need to logging two parts separately.

Also I think the argument sequence in this log statement should be topicName, subName, txnID, no?

sijie · 2019-05-20T02:04:07Z

...broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

+
+                    // If single ack is within range of cumulative ack of an ongoing transaction, fail the ack.
+                    if (null != this.pendingCumulativeAckMessage &&
+                            ((PositionImpl) position).compareTo((PositionImpl) this.pendingCumulativeAckMessage) < 0) {


do we need to check if position is an instance of PositionImpl?

Alternatively, maybe we can consider making Position extending Comparable interface?

I thought we mostly check type in catch clause or when passed in parameter is an Object, and PositionImpl is only implementation class of Position so I skip the check.
But seems I do find some places in BK code base that we're checking if it's a PositionImpl instance. Also added here, just for safe.

sijie · 2019-05-20T02:04:39Z

...broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

+
+                    // If single ack is within range of cumulative ack of an ongoing transaction, fail the ack.
+                    if (null != this.pendingCumulativeAckMessage &&
+                            ((PositionImpl) position).compareTo((PositionImpl) this.pendingCumulativeAckMessage) < 0) {


should it be <=?

yes, markDelete include the last position, so should be <=.

sijie · 2019-05-20T02:16:06Z

...broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

+                pendingAckMessage.add(position);
+                this.pendingAckMessages.add(position);
+            }
+        }


I think we need to persistent these pending acks to cursor. so when a broker crashes, we can restore/recover those pending acks back. otherwise if a broker crashes, all the pending acks are gone.

This can probably be done in a subsequent change though. We can consider updating the description of this PR to introduce in-memory PENDING_ACK state.

hmmm, you're right, I missed this part, will add in next pr.

sijie · 2019-05-20T02:20:37Z

...broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

+            String errorMsg = "[" + topicName + "][" + subName + "] Transaction with id:" + txnID.getMostSigBits()
+                            + txnID.getLeastSigBits() + " not found.";
+            log.error(errorMsg);
+            throw new IllegalArgumentException(errorMsg);


throw TransactionNotFoundException?

yes, it's a better one, I'll merge mainline and update.

sijie · 2019-05-20T02:24:26Z

...broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

+     * @throws TransactionConflictException
+     */
+    public synchronized void commitTxn(TxnID txnID, Map<String,Long> properties) throws TransactionConflictException {
+        if (!this.pendingAckMessagesMap.containsKey(txnID)) {


Actually this check can be moved down to line 890. so we just do remove and check if the TxnID can be be removed. so it is one operation to prevent any concurrent commits on same TxnID.

ConcurrentOpenHashSet<Position> pendingAckMessageCurTxn = this.pendingAckMessageMap.remove(txnID); if (null == pendingAckMessageCurTxn) { throw new TransactionNotFoundException(...); }

make sense, will update.

sijie · 2019-05-20T02:25:44Z

...broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

+     * @throws IllegalArgumentException
+     * @throws TransactionConflictException
+     */
+    public synchronized void commitTxn(TxnID txnID, Map<String,Long> properties) throws TransactionConflictException {


I think we should return CompletableFuture<Void> here, because a transaction is only committed when those positions are deleted.

sijie · 2019-05-20T02:27:12Z

...broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

+     * @param txnID  {@link TxnID} to identify the transaction.
+     * @throws IllegalArgumentException
+     */
+    public synchronized void abortTxn(TxnID txnID) {


similar comments as commitTxn - return CompletableFuture<Void>.

In this PR you can handle in-memory pending_ack state, but in subsequent PRs, we need to look into persistenting those states into cursor ledger for handling broker crashes.

MarvinCai · 2019-05-24T05:03:50Z

...broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

+            throw new IllegalArgumentException(errorMsg);
+        }
+
+        // This shouldn't happen.


After reconsideration, I think this check here doesn't really make sense.
When client commit a transaction, it doesn't necessary to be the ongoing transaction that did cumulative ack on this subscription before.
But we could add timeout check here, so when a transaction did cumulative ack but did commit within time, it's position will be removed. So other subsequent transaction can still use cumulative ack.

MarvinCai · 2019-05-28T05:18:56Z

@sijie Can you take another look, thanks!

sijie · 2019-06-01T13:23:35Z

pulsar-transaction/common/src/main/java/org/apache/pulsar/transaction/impl/common/TxnID.java

@@ -45,4 +45,8 @@
     */
    private final long leastSigBits;

+    @Override
+    public String toString() {
+        return String.valueOf(mostSigBits) + leastSigBits;


nit: maybe "(" + mostSigBits + "," + leastSigBits + ")"?

sijie · 2019-06-01T13:26:12Z

...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;


actually one last comment:

@MarvinCai : can you make pendingAckMessagesMap and pendingAckMessages non-final and initialize them when they are needed? so we don't need to construct these two structures if people don't use these features, especially when transaction is a feature under development.

Sure, changed to only initialize them when client start ack message for transaction, and add null check for other places using them.

sijie · 2019-06-01T13:26:54Z

@MarvinCai

Can you rebase this pull request to latest master and also address my last comment?

Make pendingAckMessagesMap and pendingAckMessages set non-final to we only initialize them when needed.

…review

MarvinCai · 2019-06-03T16:40:06Z

rerun java8 tests
rerun cpp tests
rerun integration tests

MarvinCai · 2019-06-03T17:43:01Z

rerun java8 tests

…review

A message should be redeliver if the map and set holding in-memory pending_ack message is null which meas consumer haven't start transaction and ack message yet.

MarvinCai · 2019-06-05T13:18:21Z

rerun cpp tests

MarvinCai · 2019-06-05T14:00:10Z

rerun cpp tests

…test to fail. Bug is when get unacked message from consumer and check if message is in pending_ack status for some transaction, it's not constructing position correctly, so causing ledger read exception. Also update java test case so it covers this scenario.

…review

MarvinCai · 2019-06-07T09:31:06Z

run java8 tests

sijie · 2019-06-07T19:15:38Z

@merlimat @rdhabalia @jiazhai this change is mostly isolated from existing code. but since it adds a few fields in the acknowledge path for transaction support. the code path will not be activated until transaction is enabled. I would like you guys also to have a look at this path as well.

sijie · 2019-06-11T02:07:23Z

ping @merlimat @rdhabalia @jiazhai

sijie

@MarvinCai missed a few review comments. PTAL

sijie · 2019-06-12T01:30:57Z

...broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

    }

    @Override
    public synchronized void redeliverUnacknowledgedMessages(Consumer consumer, List<PositionImpl> positions) {
-        dispatcher.redeliverUnacknowledgedMessages(consumer, positions);
+        // Check if message is in pending_ack status.
+        List<PositionImpl> pendingPositions = new ArrayList<>();


@MarvinCai

I missed this in the previous reviews. We should avoid creating a new array list when this is called when transaction is not used.

So the logic should be something like:

List<PositionImpl> pendingPositions = positions; if (pendingAckMessages != null || cumulativeAckPosition != null) { pendingPositions = new ArrayList<>(); // process the logic here }

sijie · 2019-06-12T01:32:06Z

...broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

+        ConcurrentLongLongPairHashMap positionMap = consumer.getPendingAcks();
+        // Check if message is in pending_ack status.
+        if (null != positionMap) {
+            List<PositionImpl> pendingPositions = new ArrayList<>();


same comment as what I made in the other redeliverUnacknowledgedMessages method.

sijie · 2019-06-12T01:33:42Z

...broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

@@ -232,6 +276,27 @@ public void acknowledgeMessage(List<Position> positions, AckType ackType, Map<St
            if (log.isDebugEnabled()) {
                log.debug("[{}][{}] Individual acks on {}", topicName, subName, positions);
            }
+
+            synchronized (PersistentSubscription.this) {


add this block under an if check statement. so this code will not be processed when transaction is not used.

if (pendingAckMessages != null || pendingCumulativeAckMessage != null) { // process the logic }

…ge we only check if message is acked by ongoing transaction when there is an ongoing transaction.

…review

MarvinCai · 2019-06-12T17:09:23Z

@sijie Addressed your comment

codelipenghui

@MarvinCai
Overall looks good to me, just few problem i'm not sure it is correct.

codelipenghui · 2019-06-13T09:57:40Z

...broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

+        // It's valid to create transaction then commit without doing any operation, which will cause
+        // pendingAckMessagesMap to be null.
+        ConcurrentOpenHashSet<Position> pendingAckMessageForCurrentTxn = pendingAckMessagesMap != null ?
+                                                this.pendingAckMessagesMap.remove(txnId) : new ConcurrentOpenHashSet();


Is it necessary to create a new Set here?

you're right, it's actually unnecessary to create this empty set, will update in next commit.

codelipenghui · 2019-06-13T10:19:43Z

...broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

+     * @param txnId  {@link TxnID} to identify the transaction.
+     * @throws IllegalArgumentException if given {@link TxnID} is not found in this subscription.
+     */
+    public synchronized CompletableFuture<Void> abortTxn(TxnID txnId) {


is abortTxn need to redeliver messages the txnId contains? because if broker handle
redeliverUnacknowledgedMessages will skip the messages which is in a transaction, if abortTxn do not redeliver these messages, i think message will never be redelivery.

I think need to call dispatcher.redeliverUnacknowledgedMessages(consumer, pendingAckMessagesMap.get(txnId))

As discussed, redelivery all un acked messages after aborting a transaction is more desired behavior for most transaction use case. So will update to call dispatcher.redeliverUnacknowledgedMessages after abort a transaction.

codelipenghui · 2019-06-13T10:35:25Z

...broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

+                        if (pendingAckMessages != null && this.pendingAckMessages.contains(position)) {
+                            log.warn("[{}][{}] Invalid acks position conflict with an ongoing transaction:{}.",
+                                    topicName, subName, this.pendingCumulativeAckTxnId.toString());
+                            return;


if consumer ack 1,2,3,4,5 and 1 is acknowledged by ongoing transaction, is 2,3,4,5 can not be acked?

2, 3, 4, 5 should be acked, return in foreach will just skip to next item.
Current logic here isn't actually handling the case correctly, will update.

codelipenghui · 2019-06-13T10:38:30Z

...broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

+            for (Position position : positions) {
+                // If try to ack message already acked by some transaction(can be itself), throw exception.
+                // Acking single message within range of cumulative ack(if exist) is considered valid operation.
+                if (this.pendingAckMessages.contains(position)) {


If ack message already acked by consumer without transaction, shall we need to abort the transaction?

Yes, it should raise conflict, will update.

1. Remove creation of unnecessary set when committing a transaction. 2. Correctly skip message already acked by ongoing transaction when doing normal ack. 3. When acking message for transaction, also check if messge has been acked. 4. After aborting a transaction, redeliver all messages delivery to a consumer during that transaction.

…review

MarvinCai · 2019-06-15T15:58:07Z

run java8 tests

MarvinCai · 2019-06-16T02:01:34Z

run java8 tests

MarvinCai · 2019-06-16T03:59:56Z

@codelipenghui thanks for your comments, I've addressed them, can you take another look?

codelipenghui

LGTM +1

MarvinCai added 5 commits May 12, 2019 14:03

Add acknowledgeMessage for txn.

af03979

Add commitTxn and abortTxn.

353c28a

Check if message is in pending_ack status before redeliver them.

f3762c0

Add unit test.

2a3d1b6

Fix maven structure, add dependency on transaction.

f80ca23

jiazhai requested a review from sijie May 13, 2019 02:14

jiazhai assigned MarvinCai May 13, 2019

jiazhai added area/broker area/client labels May 13, 2019

Use + operation instead of String.format to construct error message.

1509dc1

sijie requested changes May 20, 2019

View reviewed changes

MarvinCai commented May 24, 2019

View reviewed changes

Address pr comments: https://github.com/apache/pulsar/pull/4265/files

0063390

sijie approved these changes Jun 1, 2019

View reviewed changes

sijie reviewed Jun 1, 2019

View reviewed changes

MarvinCai added 2 commits June 2, 2019 23:34

Update strig format for TxnId.

fa0cddf

Make pendingAckMessagesMap and pendingAckMessages set non-final to we only initialize them when needed.

Merge remote-tracking branch 'apache-pulsar/master' into zxc-txn-ack-…

f26f3bc

…review

MarvinCai added 3 commits June 4, 2019 10:52

Fix construcor in PersistentSubscriptionTest after merge in mainline.

4cd7abd

Merge remote-tracking branch 'apache-pulsar/master' into zxc-txn-ack-…

20f7f2d

…review

Fix condition check for deciding whether a message should be redeliver.

113bd6b

A message should be redeliver if the map and set holding in-memory pending_ack message is null which meas consumer haven't start transaction and ack message yet.

MarvinCai added 2 commits June 6, 2019 23:58

Merge remote-tracking branch 'apache-pulsar/master' into zxc-txn-ack-…

8a31c14

…review

sijie approved these changes Jun 7, 2019

View reviewed changes

sijie requested review from merlimat, rdhabalia and jiazhai June 7, 2019 19:14

sijie added this to the 2.5.0 milestone Jun 7, 2019

sijie reviewed Jun 12, 2019

View reviewed changes

MarvinCai added 2 commits June 11, 2019 21:13

Add condition check so when redeliverUnAckMessage or acknowledgeMessa…

0bf338a

…ge we only check if message is acked by ongoing transaction when there is an ongoing transaction.

Merge remote-tracking branch 'apache-pulsar/master' into zxc-txn-ack-…

afacf32

…review

codelipenghui changed the title ~~[transaction][acknowledge] Introduce PENDING_ACK state in acknowledgement path~~ [transaction][acknowledge] Introduce in-memory PENDING_ACK state in acknowledgement path Jun 13, 2019

codelipenghui reviewed Jun 13, 2019

View reviewed changes

MarvinCai added 3 commits June 14, 2019 23:32

Merge remote-tracking branch 'apache-pulsar/master' into zxc-txn-ack-…

089e644

…review

Add method mock for mocked cursor to solve NPE from testcase.

8e3d3cf

codelipenghui approved these changes Jun 17, 2019

View reviewed changes

sijie modified the milestones: 2.5.0, 2.4.0 Jun 17, 2019

sijie merged commit e66ba27 into apache:master Jun 17, 2019

kuangye098 mentioned this pull request Apr 15, 2020

PIP-31: Transactional Streaming #2664

Closed

sijie mentioned this pull request Aug 10, 2020

ISSUE-2664: PIP-31: Transactional Streaming streamnative/pulsar-archived#106

Closed

50 tasks

gaoran10 mentioned this pull request Nov 21, 2020

Transactional Streaming V1 #8657

Closed

32 tasks

[transaction][acknowledge] Introduce in-memory PENDING_ACK state in acknowledgement path #4265

[transaction][acknowledge] Introduce in-memory PENDING_ACK state in acknowledgement path #4265

Conversation

MarvinCai commented May 12, 2019 • edited Loading

MarvinCai commented May 12, 2019

sijie commented May 14, 2019

sijie left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

MarvinCai May 23, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

MarvinCai May 24, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

MarvinCai commented May 28, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sijie commented Jun 1, 2019

MarvinCai commented Jun 3, 2019

MarvinCai commented Jun 3, 2019

MarvinCai commented Jun 5, 2019

MarvinCai commented Jun 5, 2019

MarvinCai commented Jun 7, 2019

sijie commented Jun 7, 2019

sijie commented Jun 11, 2019

sijie left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

MarvinCai commented Jun 12, 2019

codelipenghui left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codelipenghui Jun 13, 2019 • edited Loading

Choose a reason for hiding this comment

MarvinCai Jun 15, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

MarvinCai Jun 14, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

MarvinCai commented Jun 15, 2019

MarvinCai commented Jun 16, 2019

MarvinCai commented Jun 16, 2019

codelipenghui left a comment

Choose a reason for hiding this comment

MarvinCai commented May 12, 2019 •

edited

Loading

MarvinCai May 23, 2019 •

edited

Loading

MarvinCai May 24, 2019 •

edited

Loading

codelipenghui Jun 13, 2019 •

edited

Loading

MarvinCai Jun 15, 2019 •

edited

Loading

MarvinCai Jun 14, 2019 •

edited

Loading