[feat][txn] Implement the AbortedTxnProcessor for TransactionBuffer

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

@@ -163,7 +163,8 @@ public int filterEntriesForConsumer(Optional<MessageMetadata[]> optMetadataArray
                     entry.release();
                     continue;
                 } else if (((PersistentTopic) subscription.getTopic())
-                        .isTxnAborted(new TxnID(msgMetadata.getTxnidMostBits(), msgMetadata.getTxnidLeastBits()))) {
+                        .isTxnAborted(new TxnID(msgMetadata.getTxnidMostBits(), msgMetadata.getTxnidLeastBits()),
+                                (PositionImpl) entry.getPosition())) {
                     individualAcknowledgeMessageIfNeeded(entry.getPosition(), Collections.emptyMap());
                     entries.set(i, null);
                     entry.release();

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

@@ -3337,8 +3337,8 @@ public PositionImpl getMaxReadPosition() {
         return this.transactionBuffer.getMaxReadPosition();
     }
 
-    public boolean isTxnAborted(TxnID txnID) {
-        return this.transactionBuffer.isTxnAborted(txnID);
+    public boolean isTxnAborted(TxnID txnID, PositionImpl readPosition) {
+        return this.transactionBuffer.isTxnAborted(txnID, readPosition);
     }
 
     public TransactionInBufferStats getTransactionInBufferStats(TxnID txnID) {

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

@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+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;
+
+
+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
+     */
+    void putAbortedTxnAndPosition(TxnID txnID, PositionImpl position);
+
+    /**
+     * Clean up invalid aborted transactions.
+     */
+    void trimExpiredAbortedTxns();
+
+    /**
+     * 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);
+
+    /**
+     * Recover transaction buffer by transaction buffer snapshot.
+     * @return a Position (startReadCursorPosition) determiner where to start to recover in the original topic.
+     */
+
+    CompletableFuture<PositionImpl> recoverFromSnapshot();
+
+    /**
+     * Delete the transaction buffer aborted transaction snapshot.
+     * @return a completableFuture.
+     */
+    CompletableFuture<Void> deleteAbortedTxnSnapshot();
+
+    /**
+     * Take aborted transactions snapshot.
+     * @return a completableFuture.
+     */
+    CompletableFuture<Void> takeAbortedTxnsSnapshot(PositionImpl maxReadPosition);
+
+    /**
+     * Get the lastSnapshotTimestamps.
+     * @return the lastSnapshotTimestamps.
+     */
+    long getLastSnapshotTimestamps();
+
+    CompletableFuture<Void> closeAsync();
+
+}

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

@@ -141,9 +141,10 @@ public interface TransactionBuffer {
     /**
      * Close the buffer asynchronously.
      * @param txnID {@link TxnID} txnId.
+     * @param readPosition the persitent position of the txn message.
      * @return the txnId is aborted.
      */
-    boolean isTxnAborted(TxnID txnID);
+    boolean isTxnAborted(TxnID txnID, PositionImpl readPosition);
 
     /**
      * Sync max read position for normal publish.

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

@@ -360,7 +360,7 @@ public CompletableFuture<Void> closeAsync() {
     }
 
     @Override
-    public boolean isTxnAborted(TxnID txnID) {
+    public boolean isTxnAborted(TxnID txnID, PositionImpl readPosition) {
         return false;
     }
 

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

@@ -0,0 +1,182 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.buffer.impl;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+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;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.broker.systopic.SystemTopicClient;
+import org.apache.pulsar.broker.transaction.buffer.AbortedTxnProcessor;
+import org.apache.pulsar.broker.transaction.buffer.metadata.AbortTxnMetadata;
+import org.apache.pulsar.broker.transaction.buffer.metadata.TransactionBufferSnapshot;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.util.FutureUtil;
+
+@Slf4j
+public class SingleSnapshotAbortedTxnProcessorImpl implements AbortedTxnProcessor {
+    private final PersistentTopic topic;
+    private final CompletableFuture<SystemTopicClient.Writer<TransactionBufferSnapshot>> takeSnapshotWriter;
+    /**
+     * Aborts, map for jude message is aborted, linked for remove abort txn in memory when this
+     * position have been deleted.
+     */
+    private final LinkedMap<TxnID, PositionImpl> aborts = new LinkedMap<>();
+
+    private volatile long lastSnapshotTimestamps;
+
+    public SingleSnapshotAbortedTxnProcessorImpl(PersistentTopic topic) {
+        this.topic = topic;
+        this.takeSnapshotWriter = this.topic.getBrokerService().getPulsar()
+                .getTransactionBufferSnapshotServiceFactory()
+                .getTxnBufferSnapshotService().createWriter(TopicName.get(topic.getName()));
+    }
+
+    @Override
+    public void putAbortedTxnAndPosition(TxnID abortedTxnId, PositionImpl position) {
+        aborts.put(abortedTxnId, position);
+    }
+
+    //In this implementation we clear the invalid aborted txn ID one by one.
+    @Override
+    public void trimExpiredAbortedTxns() {
+        while (!aborts.isEmpty() && !((ManagedLedgerImpl) topic.getManagedLedger())
+                .ledgerExists(aborts.get(aborts.firstKey()).getLedgerId())) {
+            if (log.isDebugEnabled()) {
+                log.debug("[{}] Topic transaction buffer clear aborted transaction, TxnId : {}, Position : {}",
+                        topic.getName(), aborts.firstKey(), aborts.get(aborts.firstKey()));
+            }
+            aborts.remove(aborts.firstKey());
+        }
+    }
+
+    @Override
+    public boolean checkAbortedTransaction(TxnID txnID, Position readPosition) {
+        return aborts.containsKey(txnID);
+    }
+
+
+    @Override
+    public CompletableFuture<PositionImpl> recoverFromSnapshot() {
+        return topic.getBrokerService().getPulsar().getTransactionBufferSnapshotServiceFactory()
+                .getTxnBufferSnapshotService()
+                .createReader(TopicName.get(topic.getName())).thenComposeAsync(reader -> {
+                    PositionImpl startReadCursorPosition = null;
+                    try {
+                        while (reader.hasMoreEvents()) {
+                            Message<TransactionBufferSnapshot> message = reader.readNext();
+                            if (topic.getName().equals(message.getKey())) {
+                                TransactionBufferSnapshot transactionBufferSnapshot = message.getValue();
+                                if (transactionBufferSnapshot != null) {
+                                    handleSnapshot(transactionBufferSnapshot);
+                                    startReadCursorPosition = PositionImpl.get(
+                                            transactionBufferSnapshot.getMaxReadPositionLedgerId(),
+                                            transactionBufferSnapshot.getMaxReadPositionEntryId());
+                                }
+                            }
+                        }
+                        closeReader(reader);
+                        return CompletableFuture.completedFuture(startReadCursorPosition);
+                    } catch (Exception ex) {
+                        log.error("[{}] Transaction buffer recover fail when read "
+                                + "transactionBufferSnapshot!", topic.getName(), ex);
+                        closeReader(reader);
+                        return FutureUtil.failedFuture(ex);
+                    }
+
+                },  topic.getBrokerService().getPulsar().getTransactionExecutorProvider()
+                        .getExecutor(this));
+    }
+
+    @Override
+    public CompletableFuture<Void> deleteAbortedTxnSnapshot() {
+        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);
+        });
+    }
+
+    @Override
+    public CompletableFuture<Void> takeAbortedTxnsSnapshot(PositionImpl maxReadPosition) {
+        return takeSnapshotWriter.thenCompose(writer -> {
+            TransactionBufferSnapshot snapshot = new TransactionBufferSnapshot();
+            snapshot.setTopicName(topic.getName());
+            snapshot.setMaxReadPositionLedgerId(maxReadPosition.getLedgerId());
+            snapshot.setMaxReadPositionEntryId(maxReadPosition.getEntryId());
+            List<AbortTxnMetadata> list = new ArrayList<>();
+            aborts.forEach((k, v) -> {
+                AbortTxnMetadata abortTxnMetadata = new AbortTxnMetadata();
+                abortTxnMetadata.setTxnIdMostBits(k.getMostSigBits());
+                abortTxnMetadata.setTxnIdLeastBits(k.getLeastSigBits());
+                abortTxnMetadata.setLedgerId(v.getLedgerId());
+                abortTxnMetadata.setEntryId(v.getEntryId());
+                list.add(abortTxnMetadata);
+            });
+            snapshot.setAborts(list);
+            return writer.writeAsync(snapshot.getTopicName(), snapshot).thenAccept(messageId -> {
+                this.lastSnapshotTimestamps = System.currentTimeMillis();
+                if (log.isDebugEnabled()) {
+                    log.debug("[{}]Transaction buffer take snapshot success! "
+                            + "messageId : {}", topic.getName(), messageId);
+                }
+            }).exceptionally(e -> {
+                log.warn("[{}]Transaction buffer take snapshot fail! ", topic.getName(), e.getCause());
+                return null;
+            });
+        });
+    }
+
+    @Override
+    public long getLastSnapshotTimestamps() {
+        return this.lastSnapshotTimestamps;
+    }
+
+    @Override
+    public CompletableFuture<Void> closeAsync() {
+        return takeSnapshotWriter.thenCompose(SystemTopicClient.Writer::closeAsync);
+    }
+
+    private void closeReader(SystemTopicClient.Reader<TransactionBufferSnapshot> reader) {
+        reader.closeAsync().exceptionally(e -> {
+            log.error("[{}]Transaction buffer reader close error!", topic.getName(), e);
+            return null;
+        });
+    }
+
+    private void handleSnapshot(TransactionBufferSnapshot snapshot) {
+        if (snapshot.getAborts() != null) {
+            snapshot.getAborts().forEach(abortTxnMetadata ->
+                    aborts.put(new TxnID(abortTxnMetadata.getTxnIdMostBits(),
+                                    abortTxnMetadata.getTxnIdLeastBits()),
+                            PositionImpl.get(abortTxnMetadata.getLedgerId(),
+                                    abortTxnMetadata.getEntryId())));
+        }
+    }
+
+}