diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java index 5670579372590..341154f3b7111 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java +++ b/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 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; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java index ba6cbee355775..3e893e8f26aee 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java +++ b/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 { protected final ConcurrentHashMap> clients; protected final NamespaceEventsSystemTopicFactory namespaceEventsSystemTopicFactory; + protected final PulsarClientImpl pulsarClient; protected final Class schemaType; protected final EventType systemTopicType; private final ConcurrentHashMap> refCountedWriterMap; - @Getter - private final TableView 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> 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 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> createReader(TopicName topicName) { @@ -173,4 +177,14 @@ public void close() throws Exception { refCountedWriterMap.clear(); } + public TableView getTableView(ScheduledExecutorService scheduledExecutor) { + TableView tableView = tableViewThreadLocal.get(); + if (tableView == null) { + tableView = new TableView<>(this::createReader, + pulsarClient.getConfiguration().getOperationTimeoutMs(), scheduledExecutor); + tableViewThreadLocal.set(tableView); + } + return tableView; + } + } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java index c737da2ed0e8b..5d360a5822df2 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java +++ b/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 recoverFromSnapshot() { final var future = new CompletableFuture(); 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(), diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java index 38b902dc52089..38ae39583e578 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java +++ b/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 takeAbortedTxnsSnapshot(Position maxReadPosition) public CompletableFuture recoverFromSnapshot() { final var pulsar = topic.getBrokerService().getPulsar(); final var future = new CompletableFuture(); - 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; } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java index b127c632009ba..7f64d51fcbe85 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java +++ b/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 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 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) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java index c3f969bf5f77f..9a8ce266b1fc9 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java +++ b/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");