[improve][broker]Reduce the lock range of SimpleCache to enhance performance

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

@@ -308,7 +308,7 @@ public class PulsarService implements AutoCloseable, ShutdownService {
 
     private TransactionPendingAckStoreProvider transactionPendingAckStoreProvider;
     private final ExecutorProvider transactionExecutorProvider;
-    private final ExecutorProvider transactionSnapshotRecoverExecutorProvider;
+    private final OrderedScheduler transactionSnapshotRecoverExecutorProvider;
     private final MonotonicClock monotonicClock;
     private String brokerId;
     private final CompletableFuture<Void> readyForIncomingRequestsFuture = new CompletableFuture<>();
@@ -380,8 +380,10 @@ public PulsarService(ServiceConfiguration config,
         if (config.isTransactionCoordinatorEnabled()) {
             this.transactionExecutorProvider = new ExecutorProvider(this.getConfiguration()
                     .getNumTransactionReplayThreadPoolSize(), "pulsar-transaction-executor");
-            this.transactionSnapshotRecoverExecutorProvider = new ExecutorProvider(this.getConfiguration()
-                    .getNumTransactionReplayThreadPoolSize(), "pulsar-transaction-snapshot-recover");
+            this.transactionSnapshotRecoverExecutorProvider = OrderedScheduler.newSchedulerBuilder()
+                    .numThreads(this.getConfiguration().getNumTransactionReplayThreadPoolSize())
+                    .name("pulsar-transaction-snapshot-recover")
+                    .build();
         } else {
             this.transactionExecutorProvider = null;
             this.transactionSnapshotRecoverExecutorProvider = null;

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

@@ -21,8 +21,8 @@
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.atomic.AtomicLong;
-import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.broker.PulsarServerException;
 import org.apache.pulsar.broker.PulsarService;
@@ -41,13 +41,19 @@ public class SystemTopicTxnBufferSnapshotService<T> {
 
     protected final ConcurrentHashMap<NamespaceName, SystemTopicClient<T>> clients;
     protected final NamespaceEventsSystemTopicFactory namespaceEventsSystemTopicFactory;
+    protected final PulsarClientImpl pulsarClient;
 
     protected final Class<T> schemaType;
     protected final EventType systemTopicType;
 
     private final ConcurrentHashMap<NamespaceName, ReferenceCountedWriter<T>> refCountedWriterMap;
-    @Getter
-    private final TableView<T> tableView;
+
+    /** SystemTopicTxnBufferSnapshotService is created only three, see also
+     *  {@link TransactionBufferSnapshotServiceFactory}. At the same time, each object can only
+     * be fixed threads access, see also {@link PulsarService#transactionSnapshotRecoverExecutorProvider}.
+     * So the un-static ThreadLocal is safe.
+     */
+    private final ThreadLocal<TableView<T>> tableViewThreadLocal = new ThreadLocal<>();
 
     // The class ReferenceCountedWriter will maintain the reference count,
     // when the reference count decrement to 0, it will be removed from writerFutureMap, the writer will be closed.
@@ -103,14 +109,12 @@ public synchronized void release() {
 
     public SystemTopicTxnBufferSnapshotService(PulsarService pulsar, EventType systemTopicType,
                                                Class<T> schemaType) throws PulsarServerException {
-        final var client = (PulsarClientImpl) pulsar.getClient();
-        this.namespaceEventsSystemTopicFactory = new NamespaceEventsSystemTopicFactory(client);
+        this.pulsarClient = (PulsarClientImpl) pulsar.getClient();
+        this.namespaceEventsSystemTopicFactory = new NamespaceEventsSystemTopicFactory(pulsarClient);
         this.systemTopicType = systemTopicType;
         this.schemaType = schemaType;
         this.clients = new ConcurrentHashMap<>();
         this.refCountedWriterMap = new ConcurrentHashMap<>();
-        this.tableView = new TableView<>(this::createReader,
-                client.getConfiguration().getOperationTimeoutMs(), pulsar.getExecutor());
     }
 
     public CompletableFuture<SystemTopicClient.Reader<T>> createReader(TopicName topicName) {
@@ -173,4 +177,14 @@ public void close() throws Exception {
         refCountedWriterMap.clear();
     }
 
+    public TableView<T> getTableView(ScheduledExecutorService scheduledExecutor) {
+        TableView<T> tableView = tableViewThreadLocal.get();
+        if (tableView == null) {
+            tableView = new TableView<>(this::createReader,
+                    pulsarClient.getConfiguration().getOperationTimeoutMs(), scheduledExecutor);
+            tableViewThreadLocal.set(tableView);
+        }
+        return tableView;
+    }
+
 }

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

@@ -21,6 +21,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.bookkeeper.mledger.Position;
 import org.apache.bookkeeper.mledger.PositionFactory;
@@ -91,10 +92,13 @@ public boolean checkAbortedTransaction(TxnID txnID) {
     public CompletableFuture<Position> recoverFromSnapshot() {
         final var future = new CompletableFuture<Position>();
         final var pulsar = topic.getBrokerService().getPulsar();
-        pulsar.getTransactionSnapshotRecoverExecutorProvider().getExecutor(this).execute(() -> {
+        final String namespace = TopicName.get(topic.getName()).getNamespace();
+        final ScheduledExecutorService scheduledExecutor = pulsar.getTransactionSnapshotRecoverExecutorProvider()
+                .chooseThread(namespace);
+        scheduledExecutor.execute(() -> {
             try {
                 final var snapshot = pulsar.getTransactionBufferSnapshotServiceFactory().getTxnBufferSnapshotService()
-                        .getTableView().readLatest(topic.getName());
+                        .getTableView(scheduledExecutor).readLatest(topic.getName());
                 if (snapshot != null) {
                     handleSnapshot(snapshot);
                     final var startReadCursorPosition = PositionFactory.create(snapshot.getMaxReadPositionLedgerId(),

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

@@ -28,6 +28,7 @@
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ConcurrentLinkedDeque;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
@@ -231,10 +232,14 @@ public CompletableFuture<Void> takeAbortedTxnsSnapshot(Position maxReadPosition)
     public CompletableFuture<Position> recoverFromSnapshot() {
         final var pulsar = topic.getBrokerService().getPulsar();
         final var future = new CompletableFuture<Position>();
-        pulsar.getTransactionSnapshotRecoverExecutorProvider().getExecutor(this).execute(() -> {
+        final String namespace = TopicName.get(topic.getName()).getNamespace();
+        final ScheduledExecutorService scheduledExecutor = pulsar.getTransactionSnapshotRecoverExecutorProvider()
+                .chooseThread(namespace);
+        scheduledExecutor.execute(() -> {
             try {
                 final var indexes = pulsar.getTransactionBufferSnapshotServiceFactory()
-                        .getTxnBufferSnapshotIndexService().getTableView().readLatest(topic.getName());
+                        .getTxnBufferSnapshotIndexService().getTableView(scheduledExecutor)
+                        .readLatest(topic.getName());
                 if (indexes == null) {
                     // Try recovering from the old format snapshot
                     future.complete(recoverOldSnapshot());
@@ -342,6 +347,10 @@ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
 
     // This method will be deprecated and removed in version 4.x.0
     private Position recoverOldSnapshot() throws Exception {
+        final String namespace = TopicName.get(topic.getName()).getNamespace();
+        final ScheduledExecutorService scheduledExecutor = topic.getBrokerService().getPulsar()
+                .getTransactionSnapshotRecoverExecutorProvider()
+                .chooseThread(namespace);
         final var pulsar = topic.getBrokerService().getPulsar();
         final var topicName = TopicName.get(topic.getName());
         final var topics = wait(pulsar.getPulsarResources().getTopicResources().listPersistentTopicsAsync(
@@ -351,7 +360,7 @@ private Position recoverOldSnapshot() throws Exception {
             return null;
         }
         final var snapshot = pulsar.getTransactionBufferSnapshotServiceFactory().getTxnBufferSnapshotService()
-                .getTableView().readLatest(topic.getName());
+                .getTableView(scheduledExecutor).readLatest(topic.getName());
         if (snapshot == null) {
             return null;
         }

pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java

@@ -44,6 +44,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import lombok.Cleanup;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.common.util.OrderedScheduler;
 import org.apache.bookkeeper.mledger.AsyncCallbacks;
 import org.apache.bookkeeper.mledger.Entry;
 import org.apache.bookkeeper.mledger.ManagedLedgerException;
@@ -592,6 +593,10 @@ public SystemTopicClient.Reader<TransactionBufferSnapshot> getReader(String topi
 
     @Test(timeOut = 30000)
     public void testTransactionBufferRecoverThrowException() throws Exception {
+        OrderedScheduler scheduler = OrderedScheduler.newSchedulerBuilder()
+                .numThreads(1)
+                .name("pulsar-transaction-snapshot-recover")
+                .build();
         String topic = NAMESPACE1 + "/testTransactionBufferRecoverThrowPulsarClientException";
         @Cleanup
         Producer<byte[]> producer = pulsarClient
@@ -620,7 +625,8 @@ public void testTransactionBufferRecoverThrowException() throws Exception {
         doReturn(CompletableFuture.completedFuture(reader))
                 .when(systemTopicTxnBufferSnapshotService).createReader(any());
         doReturn(refCounterWriter).when(systemTopicTxnBufferSnapshotService).getReferenceWriter(any());
-        doReturn(new MockTableView(pulsarServiceList.get(0))).when(systemTopicTxnBufferSnapshotService).getTableView();
+        doReturn(new MockTableView(pulsarServiceList.get(0))).when(systemTopicTxnBufferSnapshotService)
+                .getTableView(scheduler);
         TransactionBufferSnapshotServiceFactory transactionBufferSnapshotServiceFactory =
                 mock(TransactionBufferSnapshotServiceFactory.class);
         doReturn(systemTopicTxnBufferSnapshotService)

pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java

@@ -1155,9 +1155,11 @@ public void testCancelTxnTimeout() throws Exception{
 
     @Test
     public void testNotChangeMaxReadPositionCountWhenCheckIfNoSnapshot() throws Exception {
+        final String topic = NAMESPACE1 + "/changeMaxReadPositionCount" + UUID.randomUUID();
+        pulsarClient.newProducer().topic(topic).create().close();
         PersistentTopic persistentTopic = (PersistentTopic) getPulsarServiceList().get(0)
                 .getBrokerService()
-                .getTopic(NAMESPACE1 + "/changeMaxReadPositionCount" + UUID.randomUUID(), true)
+                .getTopic(topic, true)
                 .get().get();
         TransactionBuffer buffer = persistentTopic.getTransactionBuffer();
         Field processorField = TopicTransactionBuffer.class.getDeclaredField("snapshotAbortedTxnProcessor");