[Transaction] Transaction pending ack persistent

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java

@@ -392,6 +392,13 @@ void markDelete(Position position, Map<String, Long> properties)
      */
     Position getMarkDeletedPosition();
 
+    /**
+     * Get the persistent newest mark deleted position on this cursor.
+     *
+     * @return the persistent mark deleted position
+     */
+    Position getPersistentMarkDeletedPosition();
+
     /**
      * Rewind the cursor to the mark deleted position to replay all the already read but not yet mark deleted messages.
      *

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java

@@ -116,6 +116,9 @@ public class ManagedCursorImpl implements ManagedCursor {
 
     protected volatile PositionImpl markDeletePosition;
 
+    // this position is have persistent mark delete position
+    protected volatile PositionImpl persistentMarkDeletePosition;
+
     protected static final AtomicReferenceFieldUpdater<ManagedCursorImpl, PositionImpl> READ_POSITION_UPDATER =
             AtomicReferenceFieldUpdater.newUpdater(ManagedCursorImpl.class, PositionImpl.class, "readPosition");
     protected volatile PositionImpl readPosition;
@@ -500,6 +503,7 @@ private void recoveredCursor(PositionImpl position, Map<String, Long> properties
 
         messagesConsumedCounter = -getNumberOfEntries(Range.openClosed(position, ledger.getLastPosition()));
         markDeletePosition = position;
+        persistentMarkDeletePosition = position;
         readPosition = ledger.getNextValidPosition(position);
         lastMarkDeleteEntry = new MarkDeleteEntry(markDeletePosition, properties, null, null);
         // assign cursor-ledger so, it can be deleted when new ledger will be switched
@@ -576,7 +580,8 @@ public void asyncReadEntries(int numberOfEntriesToRead, long maxSizeBytes, ReadE
                                  Object ctx, PositionImpl maxPosition) {
         checkArgument(numberOfEntriesToRead > 0);
         if (isClosed()) {
-            callback.readEntriesFailed(new ManagedLedgerException("Cursor was already closed"), ctx);
+            callback.readEntriesFailed(new ManagedLedgerException
+                    .CursorAlreadyClosedException("Cursor was already closed"), ctx);
             return;
         }
 
@@ -628,7 +633,8 @@ public void asyncGetNthEntry(int n, IndividualDeletedEntries deletedEntries, Rea
             Object ctx) {
         checkArgument(n > 0);
         if (isClosed()) {
-            callback.readEntryFailed(new ManagedLedgerException("Cursor was already closed"), ctx);
+            callback.readEntryFailed(new ManagedLedgerException
+                    .CursorAlreadyClosedException("Cursor was already closed"), ctx);
             return;
         }
 
@@ -781,7 +787,7 @@ public void asyncReadEntriesOrWait(int maxEntries, long maxSizeBytes, ReadEntrie
         }
     }
 
-    private boolean isClosed() {
+    public boolean isClosed() {
         return state == State.Closed || state == State.Closing;
     }
 
@@ -1597,7 +1603,8 @@ public void asyncMarkDelete(final Position position, Map<String, Long> propertie
         checkArgument(position instanceof PositionImpl);
 
         if (isClosed()) {
-            callback.markDeleteFailed(new ManagedLedgerException("Cursor was already closed"), ctx);
+            callback.markDeleteFailed(new ManagedLedgerException
+                    .CursorAlreadyClosedException("Cursor was already closed"), ctx);
             return;
         }
 
@@ -1673,7 +1680,8 @@ protected void internalAsyncMarkDelete(final PositionImpl newPosition, Map<Strin
             // The state might have changed while we were waiting on the queue mutex
             switch (STATE_UPDATER.get(this)) {
             case Closed:
-                callback.markDeleteFailed(new ManagedLedgerException("Cursor was already closed"), ctx);
+                callback.markDeleteFailed(new ManagedLedgerException
+                        .CursorAlreadyClosedException("Cursor was already closed"), ctx);
                 return;
 
             case NoLedger:
@@ -1729,6 +1737,11 @@ public void operationComplete() {
                         subMap.values().forEach(BitSetRecyclable::recycle);
                         subMap.clear();
                     }
+                    if (persistentMarkDeletePosition == null
+                            || mdEntry.newPosition.compareTo(persistentMarkDeletePosition) > 0) {
+                        persistentMarkDeletePosition = mdEntry.newPosition;
+                    }
+
                 } finally {
                     lock.writeLock().unlock();
                 }
@@ -1835,7 +1848,8 @@ public void deleteFailed(ManagedLedgerException exception, Object ctx) {
     @Override
     public void asyncDelete(Iterable<Position> positions, AsyncCallbacks.DeleteCallback callback, Object ctx) {
         if (isClosed()) {
-            callback.deleteFailed(new ManagedLedgerException("Cursor was already closed"), ctx);
+            callback.deleteFailed(new ManagedLedgerException
+                    .CursorAlreadyClosedException("Cursor was already closed"), ctx);
             return;
         }
 
@@ -2064,6 +2078,11 @@ public Position getMarkDeletedPosition() {
         return markDeletePosition;
     }
 
+    @Override
+    public Position getPersistentMarkDeletedPosition() {
+        return this.persistentMarkDeletePosition;
+    }
+
     @Override
     public void rewind() {
         lock.writeLock().lock();

managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorConcurrencyTest.java

@@ -161,7 +161,7 @@ public void run() {
                         Thread.sleep(1);
                     }
                 } catch (ManagedLedgerException e) {
-                    if (!e.getMessage().equals("Cursor was already closed")) {
+                    if (!(e instanceof ManagedLedgerException.CursorAlreadyClosedException)) {
                         gotException.set(true);
                     }
 

managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainerTest.java

@@ -130,6 +130,11 @@ public Position getMarkDeletedPosition() {
             return position;
         }
 
+        @Override
+        public Position getPersistentMarkDeletedPosition() {
+            return position;
+        }
+
         @Override
         public String getName() {
             return name;

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

@@ -221,13 +221,6 @@ public class ServiceConfiguration implements PulsarConfiguration {
     )
     private int numCacheExecutorThreadPoolSize = 10;
 
-    @FieldContext(
-            category = CATEGORY_SERVER,
-            doc = "Number of threads to use for pulsar broker service."
-                    + " The executor in thread pool will do transaction recover"
-    )
-    private int numTransactionExecutorThreadPoolSize = Runtime.getRuntime().availableProcessors();
-
     @FieldContext(category = CATEGORY_SERVER, doc = "Max concurrent web requests")
     private int maxConcurrentHttpRequests = 1024;
 
@@ -2039,6 +2032,20 @@ public class ServiceConfiguration implements PulsarConfiguration {
     private String transactionBufferProviderClassName =
             "org.apache.pulsar.broker.transaction.buffer.impl.TopicTransactionBufferProvider";
 
+    @FieldContext(
+            category = CATEGORY_TRANSACTION,
+            doc = "Class name for transaction pending ack store provider"
+    )
+    private String transactionPendingAckStoreProviderClassName =
+            "org.apache.pulsar.broker.transaction.pendingack.impl.MLPendingAckStoreProvider";
+
+    @FieldContext(
+            category = CATEGORY_TRANSACTION,
+            doc = "Number of threads to use for pulsar transaction replay PendingAckStore or TransactionBuffer."
+                    + "Default is 5"
+    )
+    private int numTransactionReplayThreadPoolSize = Runtime.getRuntime().availableProcessors();
+
     @FieldContext(
             category = CATEGORY_TRANSACTION,
             doc = "Transaction buffer take snapshot transaction count"

pulsar-broker/pom.xml

@@ -477,7 +477,10 @@
         <configuration>
           <protocArtifact>com.google.protobuf:protoc:${protoc3.version}:exe:${os.detected.classifier}</protocArtifact>
           <checkStaleness>true</checkStaleness>
-          <excludes>**/ResourceUsage.proto</excludes>
+          <excludes>
+            <exclude>**/ResourceUsage.proto</exclude>
+            <exclude>**/TransactionPendingAck.proto</exclude>
+          </excludes>
         </configuration>
         <executions>
           <execution>
@@ -495,7 +498,10 @@
         <artifactId>lightproto-maven-plugin</artifactId>
         <version>${lightproto-maven-plugin.version}</version>
         <configuration>
-          <sources>${project.basedir}/src/main/proto/ResourceUsage.proto</sources>
+          <sources>
+            <source>${project.basedir}/src/main/proto/TransactionPendingAck.proto</source>
+            <source>${project.basedir}/src/main/proto/ResourceUsage.proto</source>
+          </sources>
           <targetSourcesSubDir>generated-sources/lightproto/java</targetSourcesSubDir>
           <targetTestSourcesSubDir>generated-sources/lightproto/java</targetTestSourcesSubDir>
         </configuration>

pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java

@@ -106,6 +106,7 @@
 import org.apache.pulsar.broker.storage.ManagedLedgerStorage;
 import org.apache.pulsar.broker.transaction.buffer.TransactionBufferProvider;
 import org.apache.pulsar.broker.transaction.buffer.impl.TransactionBufferClientImpl;
+import org.apache.pulsar.broker.transaction.pendingack.TransactionPendingAckStoreProvider;
 import org.apache.pulsar.broker.validator.MultipleListenerValidator;
 import org.apache.pulsar.broker.web.WebService;
 import org.apache.pulsar.client.admin.PulsarAdmin;
@@ -223,7 +224,6 @@ public class PulsarService implements AutoCloseable {
     private TransactionMetadataStoreService transactionMetadataStoreService;
     private TransactionBufferProvider transactionBufferProvider;
     private TransactionBufferClient transactionBufferClient;
-    private ScheduledExecutorService transactionExecutor;
     private HashedWheelTimer transactionTimer;
 
     private BrokerInterceptor brokerInterceptor;
@@ -240,6 +240,9 @@ public class PulsarService implements AutoCloseable {
     private MetadataStoreExtended configurationMetadataStore;
     private PulsarResources pulsarResources;
 
+    private TransactionPendingAckStoreProvider transactionPendingAckStoreProvider;
+    private final ScheduledExecutorService transactionReplayExecutor;
+
     public enum State {
         Init, Started, Closing, Closed
     }
@@ -287,6 +290,14 @@ public PulsarService(ServiceConfiguration config,
                 new DefaultThreadFactory("pulsar"));
         this.cacheExecutor = Executors.newScheduledThreadPool(config.getNumCacheExecutorThreadPoolSize(),
                 new DefaultThreadFactory("zk-cache-callback"));
+
+        if (config.isTransactionCoordinatorEnabled()) {
+            this.transactionReplayExecutor = Executors.newScheduledThreadPool(
+                    config.getNumTransactionReplayThreadPoolSize(),
+                    new DefaultThreadFactory("transaction-replay"));
+        } else {
+            this.transactionReplayExecutor = null;
+        }
         }
 
     public MetadataStoreExtended createConfigurationMetadataStore() throws MetadataStoreException {
@@ -445,8 +456,6 @@ public CompletableFuture<Void> closeAsync() {
                 transactionBufferClient.close();
             }
 
-            executorServicesShutdown.shutdown(transactionExecutor);
-
             if (coordinationService != null) {
                 coordinationService.close();
             }
@@ -458,6 +467,10 @@ public CompletableFuture<Void> closeAsync() {
                 configurationMetadataStore.close();
             }
 
+            if (transactionReplayExecutor != null) {
+                transactionReplayExecutor.shutdown();
+            }
+
             // add timeout handling for closing executors
             asyncCloseFutures.add(executorServicesShutdown.handle());
 
@@ -729,9 +742,6 @@ public Boolean get() {
 
             // Register pulsar system namespaces and start transaction meta store service
             if (config.isTransactionCoordinatorEnabled()) {
-                this.transactionExecutor = Executors.newScheduledThreadPool(
-                        config.getNumTransactionExecutorThreadPoolSize(),
-                        new DefaultThreadFactory("pulsar-transaction"));
                 this.transactionBufferSnapshotService = new SystemTopicBaseTxnBufferSnapshotService(getClient());
                 this.transactionTimer =
                         new HashedWheelTimer(new DefaultThreadFactory("pulsar-transaction-timer"));
@@ -744,6 +754,8 @@ public Boolean get() {
 
                 transactionBufferProvider = TransactionBufferProvider
                         .newProvider(config.getTransactionBufferProviderClassName());
+                transactionPendingAckStoreProvider = TransactionPendingAckStoreProvider
+                        .newProvider(config.getTransactionPendingAckStoreProviderClassName());
             }
 
             this.metricsGenerator = new MetricsGenerator(this);
@@ -1170,6 +1182,10 @@ public ScheduledExecutorService getCacheExecutor() {
         return cacheExecutor;
     }
 
+    public ScheduledExecutorService getTransactionReplayExecutor() {
+        return transactionReplayExecutor;
+    }
+
     public ScheduledExecutorService getLoadManagerExecutor() {
         return loadManagerExecutor;
     }

pulsar-broker/src/main/java/org/apache/pulsar/broker/TransactionMetadataStoreService.java

@@ -399,17 +399,13 @@ private CompletableFuture<Void> endTxnInTransactionBuffer(TxnID txnID, int txnAc
     }
 
     private static boolean isRetryableException(Throwable e) {
-        if (e instanceof TransactionMetadataStoreStateException
+        return e instanceof TransactionMetadataStoreStateException
                 || e instanceof RequestTimeoutException
                 || e instanceof ManagedLedgerException
                 || e instanceof BrokerPersistenceException
                 || e instanceof LookupException
                 || e instanceof ReachMaxPendingOpsException
-                || e instanceof ConnectException) {
-            return true;
-        } else {
-            return false;
-        }
+                || e instanceof ConnectException;
     }
 
     private CompletableFuture<Void> endTxnInTransactionMetadataStore(TxnID txnID, int txnAction) {

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

@@ -253,19 +253,18 @@ protected int getNumberOfSameAddressConsumers(final String clientAddress,
         return count;
     }
 
-    protected void addConsumerToSubscription(Subscription subscription, Consumer consumer)
-            throws BrokerServiceException {
+    protected CompletableFuture<Void> addConsumerToSubscription(Subscription subscription, Consumer consumer) {
         if (isConsumersExceededOnTopic()) {
             log.warn("[{}] Attempting to add consumer to topic which reached max consumers limit", topic);
-            throw new ConsumerBusyException("Topic reached max consumers limit");
+            return FutureUtil.failedFuture(new ConsumerBusyException("Topic reached max consumers limit"));
         }
 
         if (isSameAddressConsumersExceededOnTopic(consumer)) {
             log.warn("[{}] Attempting to add consumer to topic which reached max same address consumers limit", topic);
-            throw new ConsumerBusyException("Topic reached max same address consumers limit");
+            return FutureUtil.failedFuture(new ConsumerBusyException("Topic reached max same address consumers limit"));
         }
 
-        subscription.addConsumer(consumer);
+        return subscription.addConsumer(consumer);
     }
 
     @Override

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

@@ -128,7 +128,7 @@ public Consumer(Subscription subscription, SubType subType, String topicName, lo
                     int priorityLevel, String consumerName,
                     int maxUnackedMessages, TransportCnx cnx, String appId,
                     Map<String, String> metadata, boolean readCompacted, InitialPosition subscriptionInitialPosition,
-                    KeySharedMeta keySharedMeta) throws BrokerServiceException {
+                    KeySharedMeta keySharedMeta) {
 
         this.subscription = subscription;
         this.subType = subType;

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

@@ -1912,7 +1912,7 @@ protected void handleEndTxnOnPartition(CommandEndTxnOnPartition command) {
         if (topicFuture != null) {
             topicFuture.whenComplete((optionalTopic, t) -> {
                 if (!optionalTopic.isPresent()) {
-                    log.error("handleEndTxnOnPartition faile ! The topic {} does not exist in broker, "
+                    log.error("handleEndTxnOnPartition fail ! The topic {} does not exist in broker, "
                             + "txnId: [{}], txnAction: [{}]", topic, txnID, TxnAction.valueOf(txnAction));
                     ctx.writeAndFlush(Commands.newEndTxnOnPartitionResponse(
                             requestId, ServerError.ServiceNotReady,

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

@@ -37,7 +37,7 @@ public interface Subscription {
 
     String getName();
 
-    void addConsumer(Consumer consumer) throws BrokerServiceException;
+    CompletableFuture<Void> addConsumer(Consumer consumer);
 
     default void removeConsumer(Consumer consumer) throws BrokerServiceException {
         removeConsumer(consumer, false);