[Transaction] Transaction pending ack lazy init. by congbobo184 · Pull Request #11091 · apache/pulsar

congbobo184 · 2021-06-25T07:49:35Z

Motivation

now, in broker.conf transactionCoordinatorEnabled=true MLPendingAck will init manageLedger, some ack will not use transaction, so don't need to init manageLedger. When this sub use transaction, we can lazy init PendingAckHandle.

implement

When this sub use transaction, we can lazy init PendingAckHandle.

Verifying this change

Add the tests for it

Does this pull request potentially affect one of the following parts:
If yes was chosen, please highlight the changes

Dependencies (does it add or upgrade a dependency): (no)
The public API: (no)
The schema: (no)
The default values of configurations: (no)
The wire protocol: (no)
The rest endpoints: (no)
The admin cli options: (no)
Anything that affects deployment: (no)

eolivelli

very interesting work !
it is a big patch. I will do a second pass tomorrow

...src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java

...rc/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleState.java

hangc0276 · 2021-08-05T07:23:09Z

@merlimat @codelipenghui @eolivelli Would you please help review this PR ?

eolivelli

Lgtm

hangc0276 · 2021-08-06T04:12:27Z

move to 2.8.2.

…_ack_lazy_load # Conflicts: # managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerFactory.java # managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java # managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStore.java # managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStoreImpl.java # pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/TransactionBufferClientTest.java

codelipenghui · 2021-09-02T14:22:41Z

...src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java

+                    initPendingAckStore();
+                    return completableFuture;
+                } else if (checkIfReady()) {
+


Miss something here?

When ready, we should let the code continue to execute downward

codelipenghui · 2021-09-02T14:26:21Z

...src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java

+            synchronized (PendingAckHandleImpl.this) {
+                if (!acceptQueue.isEmpty()) {
+                    CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+                    addIndividualAcknowledgeMessageRequest(txnID, positions, completableFuture);


Why we can't add to the queue if the queue is empty?

We need to make sure that Ack is in order

codelipenghui · 2021-09-02T14:28:14Z

...src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java

+            }
+        }
+
+        if (!acceptQueue.isEmpty() && !isInCacheRequest) {


Is it able to refine the code here? looks like the above part has check isInCacheRequest but here check it again.

optimized, please review again

codelipenghui · 2021-09-02T14:32:51Z

...ain/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckReplyCallBack.java

+
+            if (pendingAckHandle.changeToReadyState()) {
+                pendingAckHandle.completeHandleFuture();
+                pendingAckHandle.handleCacheRequest();


We will hold the lock for a long time here.

In order to ensure the order of ack, I think we should continue to acquire this lock

gaoran10

Great work! Left some comments.

It seems that the transaction pending ack lay initialization increases the logic complexity, does this is necessary?

gaoran10 · 2021-09-02T18:37:41Z

...ain/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckReplyCallBack.java

+            if (pendingAckHandle.changeToReadyState()) {
+                pendingAckHandle.completeHandleFuture();
+                pendingAckHandle.handleCacheRequest();
+                log.info("Topic name : [{}], SubName : [{}] pending ack state reply success!",


It seems that this log is the same as the above.

gaoran10 · 2021-09-02T18:50:17Z

...ain/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckReplyCallBack.java

+                log.info("Topic name : [{}], SubName : [{}] pending ack state reply success!",
+                        pendingAckHandle.getTopicName(), pendingAckHandle.getSubName());
+            } else {
+                log.error("Topic name : [{}], SubName : [{}] pending ack state reply fail!",


If the pending ack handle relay failed, how to reply again?

if change ready state fail, represent that the pending ack has already close

gaoran10 · 2021-09-02T18:54:19Z

...src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java


+    private void initPendingAckStore() {
+        if (changeToInitializingState()) {
+            synchronized (PendingAckHandleImpl.this) {


I'm not sure why need this lock, it seems that only one thread will perform the replay operation because only one thread could change the state successfully.

if change Initializing state success, in this time, we can change the state to closed, when change state to closed, we don't need to do replay op.

gaoran10 · 2021-09-02T19:04:31Z

...src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java

    @Override
    public synchronized CompletableFuture<Void> commitTxn(TxnID txnID, Map<String, Long> properties,
-                                                          long lowWaterMark) {
+                                                          long lowWaterMark, boolean isInCacheRequest) {


It seems that the param isInCacheRequest isn't used, do we need to add this param?

good catch, we should add the commit request !isInCacheRequest in to a queue.

gaoran10 · 2021-09-02T19:06:35Z

...src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java

    @Override
-    public synchronized CompletableFuture<Void> abortTxn(TxnID txnId, Consumer consumer, long lowWaterMark) {
+    public synchronized CompletableFuture<Void> abortTxn(TxnID txnId, Consumer consumer,
+                                                         long lowWaterMark, boolean isInCacheRequest) {


It seems that the param isInCacheRequest isn't used, do we need to add this param?

good catch, we should add the abort request !isInCacheRequest in to a queue.

gaoran10 · 2021-09-02T19:08:17Z

...src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java

    public CompletableFuture<Void> close() {
-        return this.pendingAckStoreFuture.thenAccept(PendingAckStore::closeAsync);
+        changeToCloseState();
+        synchronized (PendingAckHandleImpl.this) {


Do we need this lock?

same with init

gaoran10 · 2021-09-02T19:20:20Z

...src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java

+
+                } else {
+                    return FutureUtil.failedFuture(
+                            new ServiceUnitNotReadyException("PendingAckHandle not replay complete!"));


Does this error log is right? It seems that the state is Error or Close.

gaoran10 · 2021-09-02T19:26:39Z

...src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java

+        if (!isInCacheRequest) {
+            if (!checkIfReady()) {
+                synchronized (PendingAckHandleImpl.this) {
+                    if (state == State.Initializing) {


It seems that this check is repeated many times, maybe we could add a method to make this check.

such as

public boolean needCacheRequest() { if (state == State.Initializing ||) { return true; } else if (state == State.None) { initPendingAckStore(); return true; } else if (checkIfReady()) { // do nothing } else { return FutureUtil.failedFuture( new ServiceUnitNotReadyException("PendingAckHandle replay failed!")); } }

in different state, commit, ack and abort will do different op, so this method return what?

If the return value is true, we could cache the request in the queue.

gaoran10 · 2021-09-02T19:38:17Z

...src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java

+                                                                boolean isInCacheRequest) {
+        if (!isInCacheRequest) {
+            if (!checkIfReady()) {
+                synchronized (PendingAckHandleImpl.this) {


Does this lock is used to make sure the state doesn't be changed?

congbobo184 · 2021-09-03T02:47:05Z

@gaoran10 thanks for you review, fix some comment, please review again.

# Conflicts: # pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/TransactionMetricsTest.java

## Motivation now, in `broker.conf` `transactionCoordinatorEnabled=true` MLPendingAck will init manageLedger, some ack will not use transaction, so don't need to init manageLedger. When this sub use transaction, we can lazy init `PendingAckHandle`. ## implement When this sub use transaction, we can lazy init `PendingAckHandle`.

[Transaction] Transaction pending ack lasy init.

c736bbb

congbobo184 requested review from codelipenghui, eolivelli, gaoran10 and merlimat June 25, 2021 07:49

fix code checkstyle

07e4a1c

codelipenghui assigned congbobo184 Jun 25, 2021

codelipenghui added this to the 2.9.0 milestone Jun 25, 2021

codelipenghui added release/2.8.1 component/transactions area/transaction and removed component/transactions labels Jun 25, 2021

codelipenghui mentioned this pull request Jun 30, 2021

Only create managed ledger for the pending ack state if there are messages ack with transactions streamnative/pulsar-archived#2671

Closed

eolivelli reviewed Jun 30, 2021

View reviewed changes

...src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java Outdated Show resolved Hide resolved

...rc/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleState.java Show resolved Hide resolved

congbo added 8 commits July 1, 2021 13:35

Fix some comment

a613363

fix some test

a7c36c8

Fix some test

c3a22d9

Fix some test

8356b88

Merge branch 'master' into congbobo184_pending_ack_lazy_load

02d69d6

fix some comment.

863b4e6

Merge branch 'master' into congbobo184_pending_ack_lazy_load

10d3353

Fix some test

b425572

eolivelli approved these changes Aug 5, 2021

View reviewed changes

hangc0276 added release/2.8.2 and removed release/2.8.1 labels Aug 6, 2021

codelipenghui requested changes Sep 2, 2021

View reviewed changes

Fix some comment

707d7c2

gaoran10 reviewed Sep 2, 2021

View reviewed changes

congbo added 3 commits September 3, 2021 10:48

Fix some comment

1f5440e

Merge branch 'master' into congbobo184_pending_ack_lazy_load

56a151c

Change handle request move lock outside

a210e83

eolivelli modified the milestones: 2.9.0, 2.10.0 Oct 6, 2021

congbo added 3 commits October 8, 2021 15:03

Merge branch 'master' into congbobo184_pending_ack_lazy_load

6083221

Merge branch 'master' into congbobo184_pending_ack_lazy_load

8d2e427

# Conflicts: # pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/TransactionMetricsTest.java

Merge branch 'master' into congbobo184_pending_ack_lazy_load

314e8f1

codelipenghui approved these changes Oct 9, 2021

View reviewed changes

congbobo184 merged commit 8b50af5 into apache:master Oct 9, 2021

codelipenghui removed the release/2.8.2 label Oct 11, 2021

codelipenghui modified the milestones: 2.10.0, 2.9.0 Nov 29, 2021

Conversation

congbobo184 commented Jun 25, 2021

Motivation

implement

Verifying this change

Uh oh!

eolivelli left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

hangc0276 commented Aug 5, 2021

Uh oh!

eolivelli left a comment

Choose a reason for hiding this comment

Uh oh!

hangc0276 commented Aug 6, 2021

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gaoran10 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

congbobo184 Sep 3, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

congbobo184 commented Sep 3, 2021

Uh oh!

Reviewers

congbobo184 Sep 3, 2021 •

edited

Loading