Instead scheduler executor to io.netty.timer

apache · Jul 26, 2022 · 346dd5d · 346dd5d
1 parent f48e17e
commit 346dd5d
Show file tree

Hide file tree

Showing 4 changed files with 20 additions and 14 deletions.
diff --git a/...src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStore.java b/...src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStore.java
@@ -22,6 +22,7 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ComparisonChain;
 import io.netty.buffer.ByteBuf;
+import io.netty.util.Timer;
 import io.netty.util.concurrent.FastThreadLocal;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -30,7 +31,6 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentSkipListMap;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Predicate;
 import java.util.stream.Stream;
@@ -117,7 +117,7 @@ public class MLPendingAckStore implements PendingAckStore {
     public MLPendingAckStore(ManagedLedger managedLedger, ManagedCursor cursor,
                              ManagedCursor subManagedCursor, long transactionPendingAckLogIndexMinLag,
                              TxnLogBufferedWriterConfig bufferedWriterConfig,
-                             ScheduledExecutorService scheduledExecutorService) {
+                             Timer timer) {
         this.managedLedger = managedLedger;
         this.cursor = cursor;
         this.currentLoadPosition = (PositionImpl) this.cursor.getMarkDeletedPosition();
@@ -128,7 +128,7 @@ public MLPendingAckStore(ManagedLedger managedLedger, ManagedCursor cursor,
         this.logIndexBackoff = new LogIndexLagBackoff(transactionPendingAckLogIndexMinLag, Long.MAX_VALUE, 1);
         this.maxIndexLag = logIndexBackoff.next(0);
         this.bufferedWriter = new TxnLogBufferedWriter(managedLedger, ((ManagedLedgerImpl) managedLedger).getExecutor(),
-                scheduledExecutorService, PendingAckLogSerializer.INSTANCE,
+                timer, PendingAckLogSerializer.INSTANCE,
                 bufferedWriterConfig.getBatchedWriteMaxRecords(), bufferedWriterConfig.getBatchedWriteMaxSize(),
                 bufferedWriterConfig.getBatchedWriteMaxDelayInMillis(), bufferedWriterConfig.isBatchEnabled());
         this.batchedPendingAckLogsWaitingForHandle = new ArrayList<>();

diff --git a/.../java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStoreProvider.java b/.../java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStoreProvider.java
@@ -18,8 +18,8 @@
  */
 package org.apache.pulsar.broker.transaction.pendingack.impl;
 
+import io.netty.util.Timer;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ScheduledExecutorService;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.bookkeeper.mledger.AsyncCallbacks;
 import org.apache.bookkeeper.mledger.ManagedCursor;
@@ -57,8 +57,8 @@ public CompletableFuture<PendingAckStore> newPendingAckStore(PersistentSubscript
         PersistentTopic originPersistentTopic = (PersistentTopic) subscription.getTopic();
         PulsarService pulsarService = originPersistentTopic.getBrokerService().getPulsar();
 
-        final ScheduledExecutorService transactionLogScheduledExecutor =
-                pulsarService.getBrokerService().getTransactionTimer();
+        final Timer brokerClientSharedTimer =
+                pulsarService.getBrokerClientSharedTimer();
         final ServiceConfiguration serviceConfiguration = pulsarService.getConfiguration();
         final TxnLogBufferedWriterConfig txnLogBufferedWriterConfig = new TxnLogBufferedWriterConfig();
         txnLogBufferedWriterConfig.setBatchEnabled(serviceConfiguration.isTransactionPendingAckBatchedWriteEnabled());
@@ -105,7 +105,7 @@ public void openCursorComplete(ManagedCursor cursor, Object ctx) {
                                                                         .getConfiguration()
                                                                         .getTransactionPendingAckLogIndexMinLag(),
                                                                 txnLogBufferedWriterConfig,
-                                                                transactionLogScheduledExecutor));
+                                                                brokerClientSharedTimer));
                                                         if (log.isDebugEnabled()) {
                                                             log.debug("{},{} open MLPendingAckStore cursor success",
                                                                     originPersistentTopic.getName(),

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
@@ -613,7 +613,8 @@ public void testEndTBRecoveringWhenManagerLedgerDisReadable() throws Exception{
     @Test
     public void testEndTPRecoveringWhenManagerLedgerDisReadable() throws Exception{
         TxnLogBufferedWriterConfig bufferedWriterConfig = new TxnLogBufferedWriterConfig();
-        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
+        HashedWheelTimer transactionTimer = new HashedWheelTimer(new DefaultThreadFactory("transaction-timer"),
+                1, TimeUnit.MILLISECONDS);
 
         String topic = NAMESPACE1 + "/testEndTPRecoveringWhenManagerLedgerDisReadable";
         admin.topics().createNonPartitionedTopic(topic);
@@ -647,7 +648,7 @@ public void testEndTPRecoveringWhenManagerLedgerDisReadable() throws Exception{
         TransactionPendingAckStoreProvider pendingAckStoreProvider = mock(TransactionPendingAckStoreProvider.class);
         doReturn(CompletableFuture.completedFuture(
                 new MLPendingAckStore(persistentTopic.getManagedLedger(), managedCursor, null,
-                        500, bufferedWriterConfig, scheduledExecutorService)))
+                        500, bufferedWriterConfig, transactionTimer)))
                 .when(pendingAckStoreProvider).newPendingAckStore(any());
         doReturn(CompletableFuture.completedFuture(true)).when(pendingAckStoreProvider).checkInitializedBefore(any());
 
@@ -680,6 +681,9 @@ public void testEndTPRecoveringWhenManagerLedgerDisReadable() throws Exception{
 
         Awaitility.await().untilAsserted(() ->
                 assertEquals(pendingAckHandle3.getStats(false).state, "Ready"));
+
+        // cleanup
+        transactionTimer.stop();
     }
 
     @Test

diff --git a/...src/test/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckMetadataTest.java b/...src/test/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckMetadataTest.java
@@ -18,8 +18,9 @@
  */
 package org.apache.pulsar.broker.transaction.pendingack;
 
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
+import io.netty.util.HashedWheelTimer;
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.concurrent.TimeUnit;
 import lombok.Cleanup;
 import org.apache.bookkeeper.mledger.AsyncCallbacks;
 import org.apache.bookkeeper.mledger.ManagedCursor;
@@ -52,7 +53,8 @@ public PendingAckMetadataTest() {
     @Test
     public void testPendingAckManageLedgerWriteFailState() throws Exception {
         TxnLogBufferedWriterConfig bufferedWriterConfig = new TxnLogBufferedWriterConfig();
-        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
+        HashedWheelTimer transactionTimer = new HashedWheelTimer(new DefaultThreadFactory("transaction-timer"),
+                1, TimeUnit.MILLISECONDS);
 
         ManagedLedgerFactoryConfig factoryConf = new ManagedLedgerFactoryConfig();
         factoryConf.setMaxCacheSize(0);
@@ -79,7 +81,7 @@ public void openLedgerFailed(ManagedLedgerException exception, Object ctx) {
         ManagedCursor subCursor = completableFuture.get().openCursor("test");
         MLPendingAckStore pendingAckStore =
                 new MLPendingAckStore(completableFuture.get(), cursor, subCursor, 500,
-                        bufferedWriterConfig, scheduledExecutorService);
+                        bufferedWriterConfig, transactionTimer);
 
         Field field = MLPendingAckStore.class.getDeclaredField("managedLedger");
         field.setAccessible(true);
@@ -102,7 +104,7 @@ public void openLedgerFailed(ManagedLedgerException exception, Object ctx) {
         completableFuture.get().close();
         cursor.close();
         subCursor.close();
-        scheduledExecutorService.shutdown();
+        transactionTimer.stop();
     }
 
 }