New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[Transaction] TransactionBuffer Refactor #8347

Merged

codelipenghui merged 14 commits into apache:master from gaoran10:txn-tb-refactor

Oct 30, 2020

Contributor

gaoran10 commented Oct 22, 2020 •

edited

Loading

Fix https://github.com/streamnative/pulsar/issues/1575
Fix issue #8378

Motivation

Use the above approach instead of the sidecar approach.

Modifications

Produce transaction messages to the topic partition.
The commit marker needs to record the related message-id list of its transaction.
When the dispatcher read a transaction marker, get the messages of the transaction by message-id list in the marker and send them to the consumer.
TransactionBuffer doesn't maintain any index data.

Verifying this change

This change added tests and can be verified as follows:

Added unit tests for produce transaction messages and end transaction and read the messages

org.apache.pulsar.broker.transaction.TransactionProduceTest
org.apache.pulsar.broker.transaction.TransactionConsumeTest

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: (yes)
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)

Documentation

Does this pull request introduce a new feature? (yes / no)

codelipenghui reviewed

View reviewed changes

pulsar-common/src/main/proto/PulsarMarkers.proto Outdated

@@ @@ -80,10 +80,12 @@ message ClusterMessageId { @@
               message MessageIdData {
               	required uint64 ledger_id = 1;
               	required uint64 entry_id  = 2;
+                  optional int32 partition = 3 [default = -1];

Contributor

codelipenghui Oct 24, 2020

Why we need partition information in the marker? A marker can only belong to a single partition right?

Contributor Author

gaoran10 Oct 26, 2020

Yes, I'll fix this.

pulsar-common/src/main/proto/PulsarMarkers.proto Outdated

Comment on lines 89 to 90

		optional MessageIdData message_id = 1;
		repeated MessageIdData messageIdList = 2;

Contributor

codelipenghui Oct 24, 2020

I think we can keep repeated MessageIdData message_ids here. Since we don't need to consider the compatibility guarantee for the first transaction release.

Contributor Author

gaoran10 Oct 26, 2020

Ok

...va/org/apache/pulsar/broker/transaction/buffer/impl/PersistentTransactionBufferProvider.java Outdated

@@ @@ -34,6 +34,7 @@ @@
               /**
                * Persistent transaction buffer provider.
                */
+              @Deprecated

Contributor

codelipenghui Oct 24, 2020

It's some of the old codes are not useful, we can delete them directly. we don't want to tell users there some deprecated components in the first transaction release.

Please check all.

Contributor Author

gaoran10 Oct 26, 2020

Ok

codelipenghui assigned gaoran10

codelipenghui added this to the 2.7.0 milestone

codelipenghui requested changes

View reviewed changes

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

+              @Slf4j
+              public class TopicTransactionBuffer implements TransactionBuffer {
+                  private PersistentTopic topic;

Contributor

codelipenghui Oct 26, 2020

Suggested change

      
                private PersistentTopic topic;
          
                private final PersistentTopic topic;

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

Comment on lines +50 to +52

+                  public CompletableFuture<TransactionMeta> getTransactionMeta(TxnID txnID) {
+                      return null;
+                  }

Contributor

codelipenghui Oct 26, 2020

Shall we need this method?

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

+                      CompletableFuture<Position> completableFuture = new CompletableFuture<>();
+                      topic.publishMessage(buffer, (e, ledgerId, entryId) -> {
+                          if (e != null) {
+                              log.error("Failed to appendBufferToTxn for txn {}", txnId, e);

Contributor

codelipenghui Oct 26, 2020

Suggested change

      
                            log.error("Failed to appendBufferToTxn for txn {}", txnId, e);
          
                            log.error("Failed to append buffer to txn {}", txnId, e);

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

+                  }
+                  private List<PulsarMarkers.MessageIdData> getMessageIdDataList(List<MessageIdData> sendMessageIdList) {
+                      List<PulsarMarkers.MessageIdData> messageIdDataList = new ArrayList<>();

Contributor

codelipenghui Oct 26, 2020

Suggested change

      
                    List<PulsarMarkers.MessageIdData> messageIdDataList = new ArrayList<>();
          
                    List<PulsarMarkers.MessageIdData> messageIdDataList = new ArrayList<>(sendMessageIdList.size());

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

+                                          .setLedgerId(msgIdData.getLedgerId())
+                                          .setEntryId(msgIdData.getEntryId()).build());
+                      }
+                      return messageIdDataList;

Contributor

codelipenghui Oct 26, 2020

Recycle the MessageIdData list before return?

pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Markers.java Outdated

                       try {
                           return Commands.serializeMetadataAndPayload(ChecksumType.Crc32c, msgMetadata, payload);
                       } finally {
                           payload.release();
                           msgMetadata.recycle();
                           msgMetadataBuilder.recycle();
+                          if (commitMarkerBuilder != null) {

Contributor

codelipenghui Oct 26, 2020

The commitMarkerBuilder always not null.

Contributor Author

gaoran10 Oct 26, 2020

I'll fix this.

pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Markers.java

+                      PulsarMarkers.TxnCommitMarker.Builder commitMarkerBuilder = PulsarMarkers.TxnCommitMarker.newBuilder();
+                      messageIdDataList.ifPresent(commitMarkerBuilder::addAllMessageId);
+                      PulsarMarkers.TxnCommitMarker commitMarker = commitMarkerBuilder.build();

Contributor

codelipenghui Oct 26, 2020

The commitMarker also should be recycled.

Contributor Author

gaoran10 Oct 26, 2020

Ok

pulsar-client/src/main/java/org/apache/pulsar/client/impl/TransactionMetaStoreHandler.java

                           return callback;
                       }
                       long requestId = client.newRequestId();
-                      ByteBuf cmd = Commands.newEndTxn(requestId, txnID.getLeastSigBits(), txnID.getMostSigBits(), PulsarApi.TxnAction.COMMIT);
+                      List<PulsarApi.MessageIdData> messageIdDataList = new ArrayList<>();

Contributor

codelipenghui Oct 26, 2020

Should be recycled after no longer use

Contributor Author

gaoran10 Oct 26, 2020

Ok

pulsar-client/src/main/java/org/apache/pulsar/client/impl/TransactionMetaStoreHandler.java

                           return callback;
                       }
                       long requestId = client.newRequestId();
-                      ByteBuf cmd = Commands.newEndTxn(requestId, txnID.getLeastSigBits(), txnID.getMostSigBits(), PulsarApi.TxnAction.ABORT);
+                      List<PulsarApi.MessageIdData> messageIdDataList = new ArrayList<>();

Contributor

codelipenghui Oct 26, 2020

Should be recycled after no longer use

Contributor Author

gaoran10 Oct 26, 2020

Ok

...main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionBufferHandlerImpl.java

                       CompletableFuture<TxnID> cb = new CompletableFuture<>();
                       if (!canSendRequest(cb)) {
                           return cb;
                       }
                       long requestId = requestIdGenerator.getAndIncrement();
-                      ByteBuf cmd = Commands.newEndTxnOnPartition(requestId, txnIdLeastBits, txnIdMostBits, topic, action);
+                      List<PulsarApi.MessageIdData> messageIdDataList = new ArrayList<>();

Contributor

codelipenghui Oct 26, 2020

Should be recycled after no longer use

Contributor Author

gaoran10 Oct 26, 2020

Ok

congbobo184 mentioned this pull request

[Transaction] ignore the txnMessageAckTest #8378

Closed

sijie mentioned this pull request

ISSUE-8378: [Transaction] ignore the txnMessageAckTest streamnative/pulsar-archived#1610

Closed

gaoran10 added 10 commits

October 29, 2020 01:30


          1. produce transaction message refactor.

54578e4

2. transaction commit refactor.


          add transaction message reader

14400cb


          1. topic transaction buffer abort operation.

f93730c

2. add new test for topic transaction buffer.


          add tests for produce transaction messages and end transaction.

afedd94

fix

f7a68ee


          code clean

f31a9bb

fix

edb1162

fix

fbb4e90


          change protocol field name, messageIdData recycle.

da9b53b


          transaction messages read refactor

37e56ec

gaoran10 force-pushed the txn-tb-refactor branch from 667a25d to 37e56ec Compare

October 28, 2020 17:33


          transaction abort marker clean

3f60bf9

codelipenghui reviewed

View reviewed changes

...c/main/java/org/apache/pulsar/broker/service/persistent/TransactionMessageDeduplication.java Outdated

+              import org.apache.bookkeeper.mledger.ManagedLedger;
+              import org.apache.pulsar.broker.PulsarService;
+              public class TransactionMessageDeduplication extends MessageDeduplication {

Contributor

codelipenghui Oct 27, 2020

Looks not related to this PR.

Contributor Author

gaoran10 Oct 29, 2020

I'll remove this class.

...va/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java Outdated

@@ @@ -73,9 +77,11 @@ @@
                   private final ServiceConfiguration serviceConfig;
                   private volatile ScheduledFuture<?> readOnActiveConsumerTask = null;
+                  LongPairSet messagesToRedeliver = new ConcurrentSortedLongPairSet(128, 2);

Contributor

codelipenghui Oct 29, 2020

Only instantiate when the transaction is enabled.

Contributor Author

gaoran10 Oct 29, 2020

I'll fix this.

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Dispatcher.java Outdated

@@ @@ -115,4 +115,9 @@ default void cursorIsReset() { @@
                   default void acknowledgementWasProcessed() {
                       // No-op
                   }
+                  default void addMessageToRedelivery(long ledgerId, long entryId) {

Contributor

codelipenghui Oct 29, 2020

Suggested change

      
                default void addMessageToRedelivery(long ledgerId, long entryId) {
          
                default void addMessageToReplay(long ledgerId, long entryId) {

Contributor

codelipenghui Oct 29, 2020

I think use replay is more reasonable here. The redeliver looks like the consumer receiver the messages.

Contributor Author

gaoran10 Oct 29, 2020

ok

...va/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java Outdated

-                              transactionReader.read(messagesToRead, consumer, this);
+                          if (havePendingReplayRead) {
+                              if (log.isDebugEnabled()) {
+                                  log.debug("have pending replay read");

Contributor

codelipenghui Oct 29, 2020

It's better to print the topic name ,subscription name in the log. You can refer to other log patterns in the PersistentDispatcherSingleActiveConsumer.java

Contributor Author

gaoran10 Oct 29, 2020

Ok

...oker/src/main/java/org/apache/pulsar/broker/service/persistent/TransactionMessageReader.java Outdated

+                  private final Dispatcher dispatcher;
+                  private final Executor executor;
+                  private final ConcurrentLongPairSet pendingReadPosition;

Contributor

codelipenghui Oct 29, 2020

We already have messageToRedeliver in the Dispatcher, why need to add a pendingReadPosition?

Contributor Author

gaoran10 Oct 29, 2020

I'll remove the class TransactionMessageReader.

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java Outdated

                       int totalMessages = 0;
                       long totalBytes = 0;
                       int totalChunkedMessages = 0;
+                      boolean afterTxnCommitMarker = false;

Contributor

codelipenghui Oct 29, 2020

Suggested change

      
                    boolean afterTxnCommitMarker = false;
          
                    boolean isAfterTxnCommitMarker = false;

gaoran10 added 3 commits

October 29, 2020 13:16

fix

a9abf62


          fix test

3a9e1e5


          fix test

c3d6e2a

codelipenghui approved these changes

View reviewed changes

codelipenghui merged commit 9173174 into apache:master

codelipenghui added component/transaction area/transaction and removed component/transaction labels

gaoran10 mentioned this pull request

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

Closed

50 tasks

huangdx0726 pushed a commit to huangdx0726/pulsar that referenced this pull request


          [Transaction] TransactionBuffer Refactor (apache#8347)

3a77ad7

Fix https://github.com/streamnative/pulsar/issues/1575
Fix issue apache#8378 

### Motivation

![image](https://user-images.githubusercontent.com/15029908/96908159-d38d3f80-14ce-11eb-9e52-ee066434d960.png)

Use the above approach instead of the sidecar approach.

### Modifications

1. Produce transaction messages to the topic partition.
2. The commit marker needs to record the related message-id list of its transaction.
3. When the dispatcher read a transaction marker, get the messages of the transaction by message-id list in the marker and send them to the consumer.
2. TransactionBuffer doesn't maintain any index data.

flowchartsman pushed a commit to flowchartsman/pulsar that referenced this pull request


          [Transaction] TransactionBuffer Refactor (apache#8347)

9323d16

Fix https://github.com/streamnative/pulsar/issues/1575
Fix issue apache#8378 

### Motivation

![image](https://user-images.githubusercontent.com/15029908/96908159-d38d3f80-14ce-11eb-9e52-ee066434d960.png)

Use the above approach instead of the sidecar approach.

### Modifications

1. Produce transaction messages to the topic partition.
2. The commit marker needs to record the related message-id list of its transaction.
3. When the dispatcher read a transaction marker, get the messages of the transaction by message-id list in the marker and send them to the consumer.
2. TransactionBuffer doesn't maintain any index data.

gaoran10 mentioned this pull request

[Transaction] Support send transaction messages synchronously #8631

Merged

codelipenghui pushed a commit that referenced this pull request


          [Transaction] Support send transaction messages synchronously (#8631)

a1cf30a

### Motivation

Currently, sending transaction messages in a sync way is disabled, because the design of the `TransactionBuffer` had some changes(refer to #8347), sending transaction messages in a sync way is allowed.

### Modifications

Remove the transaction check in the `send` method of the class `TypedMessageBuilderImpl`.

This was referenced Nov 21, 2020

PIP-31: Transactional Streaming #2664

Closed

Transactional Streaming V1 #8657

Closed

cbornet mentioned this pull request

Transaction support cbornet/pulsar-grpc#1

Open

cytnju commented Feb 21, 2022 •

edited

Loading

Would you please tell me why pulsar finally chose to replace sidecar's appraoch with marker's appraoch, which seems different from the conclusion in the previous PIP-31 ? @gaoran10

Contributor

eolivelli commented Feb 21, 2022

Would you please tell me why pulsar finally chose to replace sidecar's appraoch with marker's appraoch, which seems different from the conclusion in the previous PIP-31 ? @gaoran10

@cytnju please ask your question on dev@pulsar.apache.org
This is not the right place for discussions, a few people follow the notifications especially for merged prs

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

area/transaction