[Transaction] Transaction buffer snapshot implementation. #9490

congbobo184 · 2021-02-05T08:46:01Z

Motivation

Transaction buffer snapshot to recover ongoing and abort transaction.

implement

snapshot metadata:

public class AbortTxnMetadata {
    long txnIdMostBits;
    long txnIdLeastBits;
    long ledgerId;
    long entryId;
}

public class TransactionBufferSnapshot {
    private String topicName;
    private long maxReadPositionLedgerId;
    private long maxReadPositionEntryId;
    private List<AbortTxnMetadata> aborts;
}

now only sore the max read position and aborts transaction, follow-up will store ongoing txns.
implement this by a namespace event topic. __transaction_buffer_snapshot
add transactionBufferSnapshotMaxTransactionCount and transactionBufferSnapshotMinTimeInMills to control taking snapshot
add SystemTopicBaseTxnBufferSnapshotService to create take snapshot Writer and Reader and cache the same namespace client.

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: (yes)
The rest endpoints: (no)
The admin cli options: (no)
Anything that affects deployment: (no)

eolivelli

great work

eolivelli · 2021-02-05T14:13:40Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

+                    log.debug("[{}]Transaction buffer not recover complete!", topic);
+                }
+                publishContext.completed(new BrokerServiceException
+                        .ServiceUnitNotReadyException("[{" + topic + "}]Transaction buffer not "


what about: "Transaction buffer recovery is still running" ?

eolivelli · 2021-02-05T14:19:12Z

...r/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java

        this.topic = topic;
+        this.changeToInitializingState();


what about moving this initialisation procedure in a separate method ?

executing non trivial code in the constructor is problematic because in case of problems you are not able to dispose correctly the object, because the caller does not receive a reference to the new object but we are still leaking references to the object.

thanks. i will move it to recover method

eolivelli · 2021-02-05T14:21:33Z

...r/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java

@@ -168,7 +277,53 @@ public void addFailed(ManagedLedgerException exception, Object ctx) {
        }
    }

+    private synchronized void takeSnapshotBuyChangeTimes() {


typo: "buy" -> "by"

congbobo184 · 2021-02-09T06:05:30Z

/pulsarbot run-failure-checks

gaoran10

Great work! left some comments.

gaoran10 · 2021-02-09T17:05:55Z

...oker/src/main/java/org/apache/pulsar/broker/systopic/TransactionBufferSystemTopicClient.java

+    protected CompletableFuture<Reader<TransactionBufferSnapshot>> newReaderAsyncInternal() {
+        return client.newReader(Schema.AVRO(TransactionBufferSnapshot.class))
+                .topic(topicName.toString())
+                .startMessageId(MessageId.latest)


Maybe the start message-id is MessageId.earliest?

gaoran10 · 2021-02-09T17:30:01Z

...r/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java

+                            + "messageId : {}", topic.getName(), messageId);
+                }
+            }).exceptionally(e -> {
+                if (log.isDebugEnabled()) {


Maybe we could make a warn-level log? Because if many snapshot operations are failed, the recovery time will be too long, the warn-level log could help users find out why the recovery cost such time.

yes, great suggestion！

gaoran10 · 2021-02-09T17:35:29Z

...r/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java

+        if (!checkIfReady()) {
+            if (log.isDebugEnabled()) {
+                log.debug("[{}]Transaction buffer not recover complete!", topic.getName());
+            }
+            return FutureUtil.failedFuture(
+                    new ServiceUnitNotReadyException("[{" + topic.getName()
+                            + "}]Transaction buffer not recover complete!"));
+        }
+


How about making these codes as a method? I found three points use this check.

gaoran10 · 2021-02-09T17:58:01Z

...r/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java

+            this.recover = recover;
+        }
+        void fillQueue() {
+            if (entryQueue.size() < entryQueue.capacity() && outstandingReadsRequests.get() == 0) {


Maybe we could wait for the entry queue size to a threshold value then read more entries to fill the entry queue.

we should make sure there is only one read operation outside, so we should not only use threshold value

gaoran10 · 2021-02-09T18:01:49Z

...r/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java

+                            }
+                        } else {
+                            if (exceptionNumber.get() > MAX_EXCEPTION_NUMBER) {
+                                log.error("[{}]Transaction buffer recover fail when "


Why the entry could be null?

this is poll method, when read operation not complete, the entry will be null.

gaoran10 · 2021-02-09T18:03:40Z

...r/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java

+        }
+    }
+
+    static class TopicTransactionBufferRecover implements Runnable {


Could you add a simple description for the recovery flow?

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

.../org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferRecoverCallBack.java

+    /**
+     * Topic transaction buffer recover complete.
+     */
+    void replayComplete();


How about changing the method name to recoverComplete?

gaoran10 · 2021-02-09T18:11:06Z

.../org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferRecoverCallBack.java

+    void handleSnapshot(TransactionBufferSnapshot snapshot);
+
+    /**
+     * Handle transaction entry.


Suggested change

* Handle transaction entry.

* Handle transaction entry beyond the snapshot.

congbobo184 · 2021-02-22T08:58:44Z

/pulsarbot run-failure-checks

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

conf/broker.conf

@@ -1203,6 +1203,13 @@ brokerServicePurgeInactiveFrequencyInSeconds=60
 transactionCoordinatorEnabled=false
 transactionMetadataStoreProviderClassName=org.apache.pulsar.transaction.coordinator.impl.MLTransactionMetadataStoreProvider

+# Transaction buffer take snapshot interval number
+transactionBufferTakeSnapshotIntervalNumber=1000


The number is the transactions count or entry count? And it's better to use max as the prefix?

codelipenghui · 2021-02-22T14:11:04Z

conf/broker.conf

+
+# Transaction buffer take snapshot interval time
+# Unit : millisecond
+transactionBufferTakeSnapshotIntervalTime=5000


Suggested change

transactionBufferTakeSnapshotIntervalTime=5000

transactionBufferSnapshotMinTimeInMills=5000

codelipenghui · 2021-02-22T14:16:21Z

...r/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java

+                log.debug("[{}]Transaction buffer not recover complete!", topic.getName());
+            }
+            return FutureUtil.failedFuture(
+                    new ServiceUnitNotReadyException("[{" + topic.getName()


If return ServiceUntiNotReadException, the client will redo the lookup. It should be a transaction exception

codelipenghui · 2021-02-22T14:18:35Z

...r/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java

@@ -168,7 +280,53 @@ public void addFailed(ManagedLedgerException exception, Object ctx) {
        }
    }

+    private synchronized void takeSnapshotByChangeTimes() {
+        if (changeMaxReadPositionAndAddAbortTimes.get() >= takeSnapshotIntervalNumber) {
+            takeSnapshot();


takeSnapshot is an async method. we need to avoid concurrent snapshots.

codelipenghui · 2021-02-22T14:21:42Z

...r/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java

+        private final AtomicLong exceptionNumber = new AtomicLong();
+
+        // TODO: MAX_EXCEPTION_NUMBER can config
+        private static final int MAX_EXCEPTION_NUMBER = 500;


What the expected behavior if the exceptions < MAX_EXCEPTION_NUMBER? In this case, if the transaction buffer recovered, does the transaction buffer will lose some transaction state?

it won't lose transaction state, because when read fail the read position will not change. it always read the correct position by the cursor.

congbobo184 · 2021-02-23T14:52:55Z

/pulsarbot run-failure-checks

congbobo184 · 2021-03-01T10:48:55Z

/pulsarbot run-failure-checks

codelipenghui · 2021-03-01T14:20:24Z

pulsar-common/src/main/proto/PulsarApi.proto

@@ -224,6 +224,7 @@ enum ServerError {
                         // use this error to indicate that this producer is now permanently
                         // fenced. Applications are now supposed to close it and create a
                         // new producer
+    TransactionBufferNotRecover = 26; // When transaction buffer not recover completely


Shall we need to expose this error to the client-side? It's better to handle it in the broker internal.

codelipenghui · 2021-03-03T00:41:37Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

+    protected final CompletableFuture<Void> transactionCompletableFuture;
    protected final TransactionBuffer transactionBuffer;


We don't need to use 2 fields for this case? We can just use CompletableFuture?

congbobo184 · 2021-03-07T05:24:30Z

/pulsarbot run-failure-checks

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

congbobo184 · 2021-03-08T09:46:23Z

/pulsarbot run-failure-checks

congbobo184 · 2021-03-09T02:17:44Z

/pulsarbot run-failure-checks

congbobo184 · 2021-03-09T06:54:20Z

/pulsarbot run-failure-checks

eolivelli

Great work.

Overall looks good to me

But there is a typo in the configuration bean, I believe this should be fixed.

eolivelli · 2021-03-09T14:34:53Z

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

+            category = CATEGORY_TRANSACTION,
+            doc = "Transaction buffer take snapshot min interval time"
+    )
+    private int transactionBufferSnapshotMinTimeInMills = 5000;


typo:
transactionBufferSnapshotMinTimeInMills -> transactionBufferSnapshotMinTimeInMillis

eolivelli · 2021-03-09T14:36:03Z

...-broker/src/main/java/org/apache/pulsar/broker/service/TransactionBufferSnapshotService.java

+    /**
+     * Close transaction buffer snapshot service.
+     */
+    void close() throws Exception;


do we really need to throw a generic Exception ?
how can this error be handled downstream ?

this method only close pulsarService use, so I think we don't need to think too much .

eolivelli

+1

congbobo184 · 2021-03-10T00:28:34Z

/pulsarbot run-faliure-checks

congbobo184 · 2021-03-10T00:34:36Z

/pulsarbot run-failure-checks

[Transaction] Transaction buffer snapshot implementation.

874e7a6

zymap requested review from codelipenghui, sijie and merlimat February 5, 2021 09:11

zymap assigned congbobo184 Feb 5, 2021

zymap added the area/transaction label Feb 5, 2021

zymap added this to the 2.8.0 milestone Feb 5, 2021

eolivelli reviewed Feb 5, 2021

View reviewed changes

congbo added 2 commits February 7, 2021 19:35

Merge branch 'master' into congbobo184_transaction_buffer_snapshot

0e96de3

Change proto to avro

e8c2399

congbo added 2 commits February 9, 2021 18:25

fix the test nullpoint

ea517a5

Fix the transaction buffer provider

688f27c

gaoran10 reviewed Feb 9, 2021

View reviewed changes

congbo added 2 commits February 14, 2021 23:21

fix some comment

cbb1339

Merge branch 'master' into congbobo184_transaction_buffer_snapshot

bb8bfca

codelipenghui reviewed Feb 22, 2021

View reviewed changes

fix some comment

ab97093

congbobo184 closed this Feb 25, 2021

congbobo184 reopened this Feb 25, 2021

congbo added 2 commits February 25, 2021 18:05

Merge branch 'master' into congbobo184_transaction_buffer_snapshot

2860c65

Add the error in cpp-client

a7504fb

Fix transaction buffer handle exception

3dc8464

codelipenghui reviewed Mar 1, 2021

View reviewed changes

congbo added 2 commits March 1, 2021 22:21

Fix the codestyle

bc53492

Delete transaction buffer not recover exception

d434987

codelipenghui reviewed Mar 3, 2021

View reviewed changes

congbo added 2 commits March 6, 2021 13:43

add the handle the future logic

879a0ef

fix the code style

25290d7

congbo added 3 commits March 7, 2021 18:34

Merge branch 'master' into congbobo184_transaction_buffer_snapshot

cf58bf5

Merge branch 'master' into congbobo184_transaction_buffer_snapshot

d4f123e

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

fix the transaction test

03da520

Merge branch 'master' into congbobo184_transaction_buffer_snapshot

77db43d

codelipenghui approved these changes Mar 9, 2021

View reviewed changes

eolivelli requested changes Mar 9, 2021

View reviewed changes

change the config name

fabe392

eolivelli approved these changes Mar 9, 2021

View reviewed changes

codelipenghui merged commit 488d384 into apache:master Mar 10, 2021

codelipenghui mentioned this pull request Mar 10, 2021

Transaction buffer snapshot streamnative/pulsar-archived#2025

Closed

thetumbled mentioned this pull request Apr 8, 2024

[cleanup][broker] remove useless code comment #22459

Merged

15 tasks

	* Handle transaction entry.
	* Handle transaction entry beyond the snapshot.

	transactionBufferTakeSnapshotIntervalTime=5000
	transactionBufferSnapshotMinTimeInMills=5000

		protected final CompletableFuture<Void> transactionCompletableFuture;
		protected final TransactionBuffer transactionBuffer;

[Transaction] Transaction buffer snapshot implementation. #9490

[Transaction] Transaction buffer snapshot implementation. #9490

Conversation

congbobo184 commented Feb 5, 2021 • 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

congbobo184 commented Feb 9, 2021

gaoran10 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

congbobo184 Feb 14, 2021 • 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

congbobo184 commented Feb 22, 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

Choose a reason for hiding this comment

congbobo184 commented Feb 23, 2021

congbobo184 commented Mar 1, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

congbobo184 commented Mar 7, 2021

congbobo184 commented Mar 8, 2021

congbobo184 commented Mar 9, 2021

congbobo184 commented Mar 9, 2021

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

eolivelli left a comment

Choose a reason for hiding this comment

congbobo184 commented Mar 10, 2021

congbobo184 commented Mar 10, 2021

congbobo184 commented Feb 5, 2021 •

edited

Loading

congbobo184 Feb 14, 2021 •

edited

Loading