[feat][txn] implement the SnapshotSegmentAbortedTxnProcessor

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

@@ -72,7 +72,7 @@ protected CompletableFuture<SystemTopicClient<T>> getTransactionBufferSystemTopi
         }
         return CompletableFuture.completedFuture(clients.computeIfAbsent(systemTopicName,
                 (v) -> namespaceEventsSystemTopicFactory
-                        .createTransactionBufferSystemTopicClient(topicName.getNamespaceObject(),
+                        .createTransactionBufferSystemTopicClient(systemTopicName,
                                 this, schemaType)));
     }
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/systopic/NamespaceEventsSystemTopicFactory.java

@@ -44,12 +44,10 @@ public TopicPoliciesSystemTopicClient createTopicPoliciesSystemTopicClient(Names
     }
 
     public <T> TransactionBufferSnapshotBaseSystemTopicClient<T> createTransactionBufferSystemTopicClient(
-            NamespaceName namespaceName, SystemTopicTxnBufferSnapshotService<T>
+            TopicName systemTopicName, SystemTopicTxnBufferSnapshotService<T>
             systemTopicTxnBufferSnapshotService, Class<T> schemaType) {
-        TopicName topicName = TopicName.get(TopicDomain.persistent.value(), namespaceName,
-                SystemTopicNames.TRANSACTION_BUFFER_SNAPSHOT);
-        log.info("Create transaction buffer snapshot client, topicName : {}", topicName.toString());
-        return new TransactionBufferSnapshotBaseSystemTopicClient(client, topicName,
+        log.info("Create transaction buffer snapshot client, topicName : {}", systemTopicName.toString());
+        return new TransactionBufferSnapshotBaseSystemTopicClient(client, systemTopicName,
                 systemTopicTxnBufferSnapshotService, schemaType);
     }
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/systopic/TransactionBufferSnapshotBaseSystemTopicClient.java

@@ -188,6 +188,7 @@ public SystemTopicClient<T> getSystemTopic() {
     protected CompletableFuture<Writer<T>> newWriterAsyncInternal() {
         return client.newProducer(Schema.AVRO(schemaType))
                 .topic(topicName.toString())
+                .enableBatching(false)
                 .createAsync().thenApply(producer -> {
                     if (log.isDebugEnabled()) {
                         log.debug("[{}] A new {} writer is created", topicName, schemaType.getName());

pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/AbortedTxnProcessor.java

@@ -19,7 +19,6 @@
 package org.apache.pulsar.broker.transaction.buffer;
 
 import java.util.concurrent.CompletableFuture;
-import org.apache.bookkeeper.mledger.Position;
 import org.apache.bookkeeper.mledger.impl.PositionImpl;
 import org.apache.pulsar.client.api.transaction.TxnID;
 
@@ -30,9 +29,9 @@ public interface AbortedTxnProcessor {
      * After the transaction buffer writes a transaction aborted marker to the topic,
      * the transaction buffer will put the aborted txnID and the aborted marker position to AbortedTxnProcessor.
      * @param txnID aborted transaction ID.
-     * @param position the position of the abort txnID
+     * @param abortedMarkerPersistentPosition the position of the abort txn marker.
      */
-    void putAbortedTxnAndPosition(TxnID txnID, PositionImpl position);
+    void putAbortedTxnAndPosition(TxnID txnID, PositionImpl abortedMarkerPersistentPosition);
 
     /**
      * Clean up invalid aborted transactions.
@@ -42,10 +41,9 @@ public interface AbortedTxnProcessor {
     /**
      * Check whether the transaction ID is an aborted transaction ID.
      * @param txnID the transaction ID that needs to be checked.
-     * @param readPosition the read position of the transaction message, can be used to find the segment.
      * @return a boolean, whether the transaction ID is an aborted transaction ID.
      */
-    boolean checkAbortedTransaction(TxnID txnID, Position readPosition);
+    boolean checkAbortedTransaction(TxnID txnID);
 
     /**
      * Recover transaction buffer by transaction buffer snapshot.
@@ -58,7 +56,7 @@ public interface AbortedTxnProcessor {
      * Delete the transaction buffer aborted transaction snapshot.
      * @return a completableFuture.
      */
-    CompletableFuture<Void> deleteAbortedTxnSnapshot();
+    CompletableFuture<Void> clearAbortedTxnSnapshot();
 
     /**
      * Take aborted transactions snapshot.

pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java

@@ -24,7 +24,6 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.bookkeeper.mledger.Position;
 import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
 import org.apache.bookkeeper.mledger.impl.PositionImpl;
 import org.apache.commons.collections4.map.LinkedMap;
@@ -57,11 +56,16 @@ public SingleSnapshotAbortedTxnProcessorImpl(PersistentTopic topic) {
         this.takeSnapshotWriter = this.topic.getBrokerService().getPulsar()
                 .getTransactionBufferSnapshotServiceFactory()
                 .getTxnBufferSnapshotService().createWriter(TopicName.get(topic.getName()));
+        this.takeSnapshotWriter.exceptionally((ex) -> {
+                    log.error("{} Failed to create snapshot writer", topic.getName());
+                    topic.close();
+                    return null;
+                });
     }
 
     @Override
-    public void putAbortedTxnAndPosition(TxnID abortedTxnId, PositionImpl position) {
-        aborts.put(abortedTxnId, position);
+    public void putAbortedTxnAndPosition(TxnID abortedTxnId, PositionImpl abortedMarkerPersistentPosition) {
+        aborts.put(abortedTxnId, abortedMarkerPersistentPosition);
     }
 
     //In this implementation we clear the invalid aborted txn ID one by one.
@@ -78,7 +82,7 @@ public void trimExpiredAbortedTxns() {
     }
 
     @Override
-    public boolean checkAbortedTransaction(TxnID txnID, Position readPosition) {
+    public boolean checkAbortedTransaction(TxnID txnID) {
         return aborts.containsKey(txnID);
     }
 
@@ -127,14 +131,12 @@ public CompletableFuture<PositionImpl> recoverFromSnapshot() {
     }
 
     @Override
-    public CompletableFuture<Void> deleteAbortedTxnSnapshot() {
+    public CompletableFuture<Void> clearAbortedTxnSnapshot() {
         return this.takeSnapshotWriter.thenCompose(writer -> {
             TransactionBufferSnapshot snapshot = new TransactionBufferSnapshot();
             snapshot.setTopicName(topic.getName());
             return writer.deleteAsync(snapshot.getTopicName(), snapshot);
-        }).thenRun(() -> {
-            log.info("[{}] Successes to delete the aborted transaction snapshot", this.topic);
-        });
+        }).thenRun(() -> log.info("[{}] Successes to delete the aborted transaction snapshot", this.topic));
     }
 
     @Override