You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by xi...@apache.org on 2023/02/20 05:25:58 UTC
[pulsar-site] branch xiangying/txn/example-cummulativeACK created (now b1498a5598d)
This is an automated email from the ASF dual-hosted git repository.
xiangying pushed a change to branch xiangying/txn/example-cummulativeACK
in repository https://gitbox.apache.org/repos/asf/pulsar-site.git
at b1498a5598d [improve][doc]Add exception examples for transaction API
This branch includes the following new commits:
new b1498a5598d [improve][doc]Add exception examples for transaction API
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[pulsar-site] 01/01: [improve][doc]Add exception examples for transaction API
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xiangying pushed a commit to branch xiangying/txn/example-cummulativeACK
in repository https://gitbox.apache.org/repos/asf/pulsar-site.git
commit b1498a5598dc215716fc2f420b0340fefc38dc16
Author: xiangying <19...@qq.com>
AuthorDate: Mon Feb 20 13:24:54 2023 +0800
[improve][doc]Add exception examples for transaction API
---
docs/txn-use.md | 113 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 file changed, 113 insertions(+)
diff --git a/docs/txn-use.md b/docs/txn-use.md
index 59ccdf0f497..1486b609e05 100644
--- a/docs/txn-use.md
+++ b/docs/txn-use.md
@@ -92,3 +92,116 @@ Consumer<byte[]> consumer = pulsarClient
.subscribe();
```
+[2] Example of using transactions to ack messages individually
+```java
+// resource prepare
+String sourceTopicName = "persistent://" + NAMESPACE1 + "/sourceTopic";
+String sinkTopicName = "persistent://" + NAMESPACE1 + "/sinkTopic";
+String subName = "shared-subscription";
+String producerName = "txn-message-producer";
+try {
+ @Cleanup
+ Producer<String> sinkProducer = pulsarClient.newProducer(Schema.STRING)
+ .topic(sinkTopicName)
+ .producerName(producerName)
+ .create();
+ @Cleanup
+ Consumer<String> sourceConsumer = pulsarClient.newConsumer(Schema.STRING)
+ .topic(sourceTopicName)
+ .subscriptionName(subName)
+ .subscriptionType(SubscriptionType.Shared)
+ .subscribe();
+ Message<String> message = null;
+ Transaction transaction = null;
+ while (true) {
+ try {
+ message = sourceConsumer.receive();
+ //Open a transaction to handle the received message
+ transaction = pulsarClient.newTransaction()
+ .withTransactionTimeout(5, TimeUnit.SECONDS)
+ .build()
+ .get();
+ //Do some things there
+ //Send message to another topic
+ sinkProducer.newMessage(transaction)
+ .value("handle message " + message.getValue())
+ .send();
+ //Ack the message that has been consumed
+ sourceConsumer.acknowledgeAsync(message.getMessageId(), transaction).get();
+ //Commit the transaction
+ transaction.commit().get();
+ } catch (ExecutionException e) {
+ Throwable exception = e.getCause();
+ if (!(exception instanceof PulsarClientException.TransactionConflictException)) {
+ //The message may not be handled, so we need to redeliver it
+ sourceConsumer.negativeAcknowledge(message);
+ }
+ if (!(exception instanceof TransactionCoordinatorClientException.TransactionNotFoundException) && transaction !=null) {
+ //Abort the transaction if there is an exception and the transaction is not end.
+ transaction.abort().get();
+ }
+ }
+ }
+} catch (Exception e) {
+ log.error("Catch Exception", e);
+}
+```
+
+[3] Example of using transactions to ack batch messages cumulatively
+```java
+// resource prepare
+String sourceTopicName = "persistent://" + NAMESPACE1 + "/sourceTopic";
+String sinkTopicName = "persistent://" + NAMESPACE1 + "/sinkTopic";
+String subName = "failover-subscription";
+String producerName = "txn-message-producer";
+try {
+ @Cleanup
+ Producer<String> sinkProducer = pulsarClient.newProducer(Schema.STRING)
+ .topic(sinkTopicName)
+ .producerName(producerName)
+ .create();
+ @Cleanup
+ Consumer<String> sourceConsumer = pulsarClient.newConsumer(Schema.STRING)
+ .topic(sourceTopicName)
+ .subscriptionName(subName)
+ .subscriptionType(SubscriptionType.Failover)
+ .subscribe();
+ Messages<String> messages = null;
+ Transaction transaction = null;
+ while (true) {
+ try {
+ messages = sourceConsumer.batchReceive();
+ if (messages.size() > 0) {
+ //Open a transaction to handle the received message
+ transaction = pulsarClient.newTransaction()
+ .withTransactionTimeout(5, TimeUnit.SECONDS)
+ .build()
+ .get();
+ Message<String> finalMessage = null;
+ for (Message<String> message : messages) {
+ //Do some things there
+ //Send message to another topic
+ sinkProducer.newMessage(transaction)
+ .value("handle message " + message.getValue())
+ .send();
+ finalMessage = message;
+ }
+ //Ack the messages that has been consumed
+ sourceConsumer.acknowledgeCumulativeAsync(finalMessage.getMessageId(), transaction).get();
+ //Commit the transaction
+ transaction.commit().get();
+ }
+ } catch (ExecutionException e) {
+ Throwable exception = e.getCause();
+ if (!(exception instanceof TransactionCoordinatorClientException.TransactionNotFoundException) && transaction !=null) {
+ //Abort the transaction if there is an exception and the transaction is not end.
+ transaction.abort().get();
+ }
+ //Redeliver messages after the transaction is aborted
+ sourceConsumer.redeliverUnacknowledgedMessages();
+ }
+ }
+} catch (Exception e) {
+ log.error("Catch Exception", e);
+}
+```
\ No newline at end of file