TransactionCoordinator: prevent thread leaks during broker shutdown (#…

…813) I noticed this thread leak during the executions of the tests ``` "Thread-3" #163 prio=5 os_prio=0 tid=0x00007f0c34011800 nid=0x603a sleeping[0x00007f0c0dea4000] java.lang.Thread.State: TIMED_WAITING (sleeping) at java.lang.Thread.sleep(Native Method) at io.streamnative.pulsar.handlers.kop.coordinator.transaction.TransactionMarkerChannelManager.lambda$new$0(TransactionMarkerChannelManager.java:148) at io.streamnative.pulsar.handlers.kop.coordinator.transaction.TransactionMarkerChannelManager$$Lambda$624/1475886210.run(Unknown Source) at java.lang.Thread.run(Thread.java:748) ``` this patch fixes the problem Co-authored-by: Enrico Olivelli <[email protected]>
streamnative · Oct 20, 2021 · 43d1180 · 43d1180
1 parent 97c49eb
commit 43d1180
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 4 deletions.
diff --git a/...a/io/streamnative/pulsar/handlers/kop/coordinator/transaction/TransactionCoordinator.java b/...a/io/streamnative/pulsar/handlers/kop/coordinator/transaction/TransactionCoordinator.java
@@ -851,6 +851,7 @@ public void shutdown() {
         log.info("Shutting down transaction coordinator ...");
         producerIdManager.shutdown();
         txnManager.shutdown();
+        transactionMarkerChannelManager.close();
         // TODO shutdown txn
         log.info("Shutdown transaction coordinator complete.");
     }
@@ -871,5 +872,4 @@ public List<FetchResponse.AbortedTransaction> getAbortedIndexList(long fetchOffs
         }
         return abortedTransactions;
     }
-
 }
diff --git a/...amnative/pulsar/handlers/kop/coordinator/transaction/TransactionMarkerChannelManager.java b/...amnative/pulsar/handlers/kop/coordinator/transaction/TransactionMarkerChannelManager.java
@@ -70,6 +70,7 @@ public class TransactionMarkerChannelManager {
     private Map<InetSocketAddress, TxnMarkerQueue> markersQueuePerBroker = new ConcurrentHashMap<>();
     private TxnMarkerQueue markersQueueForUnknownBroker = new TxnMarkerQueue(null);
     private BlockingQueue<PendingCompleteTxn> txnLogAppendRetryQueue = new LinkedBlockingQueue<>();
+    private volatile boolean closed;
 
     @AllArgsConstructor
     private static class PendingCompleteTxn {
@@ -141,16 +142,18 @@ public TransactionMarkerChannelManager(KafkaServiceConfiguration kafkaConfig,
         bootstrap.channel(NioSocketChannel.class);
         bootstrap.handler(new TransactionMarkerChannelInitializer(kafkaConfig, enableTls));
 
-        new Thread(() -> {
-            while (true) {
+        Thread thread = new Thread(() -> {
+            while (!closed) {
                 drainQueuedTransactionMarkers();
                 try {
                     Thread.sleep(1);
                 } catch (InterruptedException e) {
                     e.printStackTrace();
                 }
             }
-        }).start();
+        }, "kop-transaction-channel-manager");
+        thread.setDaemon(true);
+        thread.start();
     }
 
     public CompletableFuture<TransactionMarkerChannelHandler> getChannel(InetSocketAddress socketAddress) {
@@ -410,4 +413,8 @@ private void drainQueuedTransactionMarkers() {
         }
     }
 
+    public void close() {
+        this.closed = true;
+    }
+
 }