[Transaction] Transaction pending ack persistent #8881

congbobo184 · 2020-12-09T15:47:16Z

Motivation

in order to handle pending ack persistent.

implement

add the transaction pending ack store, it will handle the pending ack metadata store.
when the sub unload, we will replay the pendingAckHandle.
we use one manageLedger to store the pending ack metadata by one sub , and replay by this managedLedger open cursor.
when we commit or abort the transaction, we will append the marker to the pendingAckStore then we will modify state memory in pendingAckHandle
we also modify the in memory state when append fail, because we don't know the persistent state, when we replay it, it will produce the wrong operation. so we append fail, we should wait tc time out or client abort this transaction.
when we append pending ack log, we will compare the the log position store the biggest topic position is bigger than persistent topic markDeletePosition. if it is smaller, will delete the position.

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)

…tion_pendingack_persistent # Conflicts: # pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckHandle.java # pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java

eolivelli

this is a great work.
I believe this is a fundamental implementation in order to ship Transactions to the users

eolivelli · 2021-01-15T07:39:47Z

pulsar-broker/generate_protobuf.sh

+
+
+PROTOC=${PROTOC:-protoc}
+${PROTOC} --java_out=pulsar-broker/src/main/java pulsar-broker/src/main/proto/TransactionPendingAck.proto


why do we need this file ?

eolivelli · 2021-01-15T07:39:53Z

pulsar-broker/generate_protobuf_docker.sh

+docker pull $IMAGE
+
+WORKDIR=/workdir
+docker run -i \


why do we need this file ?

eolivelli · 2021-01-15T07:41:39Z

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

@@ -161,6 +162,12 @@ void setReplicated(boolean replicated) {

    @Override
    public synchronized void addConsumer(Consumer consumer) throws BrokerServiceException {
+        if (pendingAckHandle instanceof PendingAckHandleImpl) {
+            if (!((PendingAckHandleImpl) pendingAckHandle).checkIfReady()) {


can we add checkIdReady to PendingAckHandle ?
using instanceof is not a good practice, we should leverage polymorphism

...r/src/main/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckReplyCallBack.java

.../main/java/org/apache/pulsar/broker/transaction/pendingack/impl/InMemoryPendingAckStore.java

eolivelli · 2021-01-15T07:51:02Z

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

+                    try {
+                        Thread.sleep(1);
+                    } catch (InterruptedException e) {
+                        //no-op


this is not good,
the contract is to exit the current activity in case of InterrupedException or at least set Thread.interrupted flag

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

eolivelli · 2021-01-15T07:57:19Z

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

+                                        InitialPosition.Earliest, new AsyncCallbacks.OpenCursorCallback() {
+                                            @Override
+                                            public void openCursorComplete(ManagedCursor cursor, Object ctx) {
+                                                if (timer == null) {


what about creating a method in parent class MLPendingAckStoreProvider

void ensureTimer() { if (timer == null) { synchronized(this) { initialize(); } } }

IMHO code will be more readable

eolivelli · 2021-01-15T08:01:49Z

...in/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStoreTimerTask.java

+            this.timer.newTimeout(this, intervalTime, TimeUnit.MILLISECONDS);
+        } catch (Exception e) {
+            log.error("PendingAck timer task error!", e);
+            if ("Cursor was already closed".equals(e.getCause().getMessage())) {


this is very error prone,
can we detect an instance of a specific class of exception ?

…tion_pendingack_persistent # Conflicts: # pom.xml # pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java # pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TransactionMarkerDeleteTest.java # pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentSubscriptionTest.java # pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTestBase.java

…://github.com/congbobo184/pulsar into congbobo184_transaction_pendingack_persistent # Conflicts: # pom.xml # pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java # pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java # pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckHandle.java # pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckReplyCallBack.java # pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckStore.java # pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/TransactionPendingAckStoreProvider.java # pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/InMemoryPendingAckStore.java # pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckReplyCallBack.java # pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStore.java # pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStoreProvider.java # pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStoreTimerTask.java # pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleDisabled.java # pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java # pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TransactionMarkerDeleteTest.java # pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentSubscriptionTest.java # pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTestBase.java # pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckPersistentTest.java

…tion_pendingack_persistent

# Conflicts: # pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckHandle.java # pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java

codelipenghui · 2021-03-05T01:07:41Z

@gaoran10 Please help review this PR.

congbobo184 · 2021-03-17T01:13:29Z

/pulsarbot run-failure-checks

congbobo184 · 2021-03-17T09:14:58Z

/pulsarbot run-failure-checks

# Conflicts: # pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckInMemoryDeleteTest.java delete pending ack recover exception

# Conflicts: # pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java

# Conflicts: # pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java # pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java # pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java

congbobo184 · 2021-04-25T14:11:26Z

@codelipenghui @eolivelli please review again. thanks.

codelipenghui · 2021-05-11T01:32:29Z

...-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java

+                } catch (BrokerServiceException e) {
+                    if (e instanceof ConsumerBusyException) {
+                        log.warn("[{}][{}] Consumer {} {} already connected", topic, subscriptionName, consumerId,
+                                consumerName);
+                    } else if (e instanceof SubscriptionBusyException) {
+                        log.warn("[{}][{}] {}", topic, subscriptionName, e.getMessage());
+                    }
+
+                    decrementUsageCount();
+                    future.completeExceptionally(e);


Why do we need to catch the exception here? If the exception occurs, you already have .exceptionally to deal with this.

the close is Is a synchronous method, so we should catch the exception.

codelipenghui · 2021-05-11T01:34:17Z

...-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java

-        } catch (BrokerServiceException e) {
-            if (e instanceof ConsumerBusyException) {
+        }).exceptionally(e -> {
+            if (e.getCause() instanceof ConsumerBusyException) {


It's better to get the Cause first and use it in the followings.

codelipenghui · 2021-05-11T01:36:00Z

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

+
+    public static final String PENDING_ACK_STORE_SUFFIX = "__transaction_pendingack";
+
+    private static final String PENDING_ACK_STORE_CURSOR_NAME = "pendingack";


Suggested change

private static final String PENDING_ACK_STORE_CURSOR_NAME = "pendingack";

private static final String PENDING_ACK_STORE_CURSOR_NAME = "__pending_ack_state";

codelipenghui · 2021-05-12T12:00:00Z

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

+
+    private final ManagedCursor cursor;
+
+    public static final String PENDING_ACK_STORE_SUFFIX = "__transaction_pendingack";


Suggested change

public static final String PENDING_ACK_STORE_SUFFIX = "__transaction_pendingack";

public static final String PENDING_ACK_STORE_SUFFIX = "__transaction_pending_ack";

congbo added 3 commits December 9, 2020 19:08

Transaction pending ack persistent

1d86b4d

Fix some named

e4a11de

eolivelli requested changes Jan 15, 2021

View reviewed changes

congbo added 8 commits February 7, 2021 12:05

Transaction pending ack persistent

a17e959

Fix some named

45588c6

Merge remote-tracking branch 'apache/master' into congbobo184_transac…

b7a9296

…tion_pendingack_persistent

Change exception

de97591

Fix some test

ed268e1

Fix pending ack test.

14d8268

codelipenghui assigned congbobo184 Mar 2, 2021

codelipenghui added this to the 2.8.0 milestone Mar 2, 2021

congbo added 2 commits March 16, 2021 16:08

Fix some design

79d0e7b

Fix the test

3d38a40

congbo added 9 commits March 17, 2021 19:02

Delete the callback

5a12efc

Merge branch 'master' into congbobo184_transaction_pendingack_persistent

9785124

# Conflicts: # pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckInMemoryDeleteTest.java delete pending ack recover exception

Delete pending ack topic create pending ack

2347bc4

Merge branch 'master' into congbobo184_transaction_pendingack_persistent

8d29edb

# Conflicts: # pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java

Add some debug log

3f79b56

Fix some codestyle

8a2c078

Fix some test

b668c0b

Fix some catch exception.

3c2ae9b

congbobo184 requested review from codelipenghui and gaoran10 April 25, 2021 14:11

congbo added 3 commits May 10, 2021 08:56

Merge branch 'master' into congbobo184_transaction_pendingack_persistent

9553edc

Merge master and add some log

f9ac213

Merge branch 'master' into congbobo184_transaction_pendingack_persistent

23ed17c

eolivelli approved these changes May 12, 2021

View reviewed changes

Fix some comment

aa50b74

codelipenghui reviewed May 12, 2021

View reviewed changes

codelipenghui added the area/transaction label May 12, 2021

Fix some comment

9a58f9c

codelipenghui approved these changes May 12, 2021

View reviewed changes

codelipenghui merged commit 8f1f5ba into apache:master May 13, 2021

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Transaction] Transaction pending ack persistent #8881

[Transaction] Transaction pending ack persistent #8881

congbobo184 commented Dec 9, 2020 •

edited

Loading

eolivelli left a comment

eolivelli Jan 15, 2021

eolivelli Jan 15, 2021

eolivelli Jan 15, 2021

eolivelli Jan 15, 2021

eolivelli Jan 15, 2021

eolivelli Jan 15, 2021

codelipenghui commented Mar 5, 2021

congbobo184 commented Mar 17, 2021

congbobo184 commented Mar 17, 2021

congbobo184 commented Apr 25, 2021

codelipenghui May 11, 2021

congbobo184 May 12, 2021

codelipenghui May 11, 2021

codelipenghui May 11, 2021

codelipenghui May 12, 2021



		PROTOC=${PROTOC:-protoc}
		${PROTOC} --java_out=pulsar-broker/src/main/java pulsar-broker/src/main/proto/TransactionPendingAck.proto


		public static final String PENDING_ACK_STORE_SUFFIX = "__transaction_pendingack";

		private static final String PENDING_ACK_STORE_CURSOR_NAME = "pendingack";


		private final ManagedCursor cursor;

		public static final String PENDING_ACK_STORE_SUFFIX = "__transaction_pendingack";

[Transaction] Transaction pending ack persistent #8881

[Transaction] Transaction pending ack persistent #8881

Conversation

congbobo184 commented Dec 9, 2020 • edited Loading

Motivation

implement

Verifying this change

eolivelli 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

codelipenghui commented Mar 5, 2021

congbobo184 commented Mar 17, 2021

congbobo184 commented Mar 17, 2021

congbobo184 commented Apr 25, 2021

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

congbobo184 commented Dec 9, 2020 •

edited

Loading