Fix FS exchange manager thread leak

Author: findepi

plugin/trino-exchange-filesystem/src/main/java/io/trino/plugin/exchange/filesystem/FileSystemExchangeManager.java

@@ -15,6 +15,7 @@
 
 import com.google.common.collect.ImmutableList;
 import com.google.inject.Inject;
+import io.airlift.bootstrap.LifeCycleManager;
 import io.opentelemetry.api.trace.Tracer;
 import io.trino.spi.TrinoException;
 import io.trino.spi.exchange.Exchange;
@@ -23,6 +24,7 @@
 import io.trino.spi.exchange.ExchangeSink;
 import io.trino.spi.exchange.ExchangeSinkInstanceHandle;
 import io.trino.spi.exchange.ExchangeSource;
+import jakarta.annotation.PreDestroy;
 
 import java.net.URI;
 import java.util.List;
@@ -40,6 +42,7 @@ public class FileSystemExchangeManager
 {
     public static final String PATH_SEPARATOR = "/";
 
+    private final LifeCycleManager lifeCycleManager;
     private final FileSystemExchangeStorage exchangeStorage;
     private final FileSystemExchangeStats stats;
     private final Tracer tracer;
@@ -57,11 +60,13 @@ public class FileSystemExchangeManager
 
     @Inject
     public FileSystemExchangeManager(
+            LifeCycleManager lifeCycleManager,
             FileSystemExchangeStorage exchangeStorage,
             FileSystemExchangeStats stats,
             FileSystemExchangeConfig fileSystemExchangeConfig,
             Tracer tracer)
     {
+        this.lifeCycleManager = requireNonNull(lifeCycleManager, "lifeCycleManager is null");
         this.exchangeStorage = requireNonNull(exchangeStorage, "exchangeStorage is null");
         this.stats = requireNonNull(stats, "stats is null");
         this.baseDirectories = ImmutableList.copyOf(requireNonNull(fileSystemExchangeConfig.getBaseDirectories(), "baseDirectories is null"));
@@ -78,6 +83,12 @@ public FileSystemExchangeManager(
         this.executor = newCachedThreadPool(daemonThreadsNamed("exchange-source-handles-creation-%s"));
     }
 
+    @PreDestroy
+    public void destroy()
+    {
+        executor.shutdownNow();
+    }
+
     @Override
     public Exchange createExchange(ExchangeContext context, int outputPartitionCount, boolean preserveOrderWithinPartition)
     {
@@ -132,4 +143,10 @@ public boolean supportsConcurrentReadAndWrite()
     {
         return false;
     }
+
+    @Override
+    public void shutdown()
+    {
+        lifeCycleManager.stop();
+    }
 }