Shutdown BigQuery, Kinesis, Redis and Memory connectors

core/trino-main/src/main/java/io/trino/connector/system/SystemConnector.java

@@ -19,12 +19,10 @@
 import io.trino.spi.connector.ConnectorSession;
 import io.trino.spi.connector.ConnectorSplitManager;
 import io.trino.spi.connector.ConnectorTransactionHandle;
-import io.trino.spi.connector.SystemTable;
 import io.trino.spi.transaction.IsolationLevel;
 import io.trino.transaction.InternalConnector;
 import io.trino.transaction.TransactionId;
 
-import java.util.Set;
 import java.util.function.Function;
 
 import static java.util.Objects.requireNonNull;
@@ -37,14 +35,6 @@ public class SystemConnector
     private final ConnectorPageSourceProvider pageSourceProvider;
     private final Function<TransactionId, ConnectorTransactionHandle> transactionHandleFunction;
 
-    public SystemConnector(
-            InternalNodeManager nodeManager,
-            Set<SystemTable> tables,
-            Function<TransactionId, ConnectorTransactionHandle> transactionHandleFunction)
-    {
-        this(nodeManager, new StaticSystemTablesProvider(tables), transactionHandleFunction);
-    }
-
     public SystemConnector(
             InternalNodeManager nodeManager,
             SystemTablesProvider tables,

plugin/trino-bigquery/src/main/java/io/trino/plugin/bigquery/BigQueryConnector.java

@@ -13,6 +13,7 @@
  */
 package io.trino.plugin.bigquery;
 
+import io.airlift.bootstrap.LifeCycleManager;
 import io.trino.plugin.base.classloader.ClassLoaderSafeConnectorMetadata;
 import io.trino.plugin.base.session.SessionPropertiesProvider;
 import io.trino.spi.connector.Connector;
@@ -37,6 +38,7 @@
 public class BigQueryConnector
         implements Connector
 {
+    private final LifeCycleManager lifeCycleManager;
     private final BigQueryTransactionManager transactionManager;
     private final BigQuerySplitManager splitManager;
     private final BigQueryPageSourceProvider pageSourceProvider;
@@ -46,13 +48,15 @@ public class BigQueryConnector
 
     @Inject
     public BigQueryConnector(
+            LifeCycleManager lifeCycleManager,
             BigQueryTransactionManager transactionManager,
             BigQuerySplitManager splitManager,
             BigQueryPageSourceProvider pageSourceProvider,
             BigQueryPageSinkProvider pageSinkProvider,
             Set<ConnectorTableFunction> connectorTableFunctions,
             Set<SessionPropertiesProvider> sessionPropertiesProviders)
     {
+        this.lifeCycleManager = requireNonNull(lifeCycleManager, "lifeCycleManager is null");
         this.transactionManager = requireNonNull(transactionManager, "transactionManager is null");
         this.splitManager = requireNonNull(splitManager, "splitManager is null");
         this.pageSourceProvider = requireNonNull(pageSourceProvider, "pageSourceProvider is null");
@@ -116,4 +120,10 @@ public List<PropertyMetadata<?>> getSessionProperties()
     {
         return sessionProperties;
     }
+
+    @Override
+    public void shutdown()
+    {
+        lifeCycleManager.stop();
+    }
 }

plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisConnector.java

@@ -15,6 +15,7 @@
 
 import com.google.common.collect.ImmutableList;
 import com.google.inject.Inject;
+import io.airlift.bootstrap.LifeCycleManager;
 import io.trino.spi.connector.Connector;
 import io.trino.spi.connector.ConnectorMetadata;
 import io.trino.spi.connector.ConnectorRecordSetProvider;
@@ -33,6 +34,7 @@
 public class KinesisConnector
         implements Connector
 {
+    private final LifeCycleManager lifeCycleManager;
     private final KinesisMetadata metadata;
     private final KinesisSplitManager splitManager;
     private final KinesisRecordSetProvider recordSetProvider;
@@ -41,11 +43,13 @@ public class KinesisConnector
 
     @Inject
     public KinesisConnector(
+            LifeCycleManager lifeCycleManager,
             KinesisMetadata metadata,
             KinesisSplitManager splitManager,
             KinesisRecordSetProvider recordSetProvider,
             KinesisSessionProperties properties)
     {
+        this.lifeCycleManager = requireNonNull(lifeCycleManager, "lifeCycleManager is null");
         this.metadata = requireNonNull(metadata, "metadata is null");
         this.splitManager = requireNonNull(splitManager, "splitManager is null");
         this.recordSetProvider = requireNonNull(recordSetProvider, "recordSetProvider is null");
@@ -82,4 +86,10 @@ public List<PropertyMetadata<?>> getSessionProperties()
     {
         return propertyList;
     }
+
+    @Override
+    public void shutdown()
+    {
+        lifeCycleManager.stop();
+    }
 }

plugin/trino-memory/src/main/java/io/trino/plugin/memory/MemoryConnector.java

@@ -13,6 +13,7 @@
  */
 package io.trino.plugin.memory;
 
+import io.airlift.bootstrap.LifeCycleManager;
 import io.trino.spi.connector.Connector;
 import io.trino.spi.connector.ConnectorMetadata;
 import io.trino.spi.connector.ConnectorPageSinkProvider;
@@ -24,21 +25,26 @@
 
 import javax.inject.Inject;
 
+import static java.util.Objects.requireNonNull;
+
 public class MemoryConnector
         implements Connector
 {
+    private final LifeCycleManager lifeCycleManager;
     private final MemoryMetadata metadata;
     private final MemorySplitManager splitManager;
     private final MemoryPageSourceProvider pageSourceProvider;
     private final MemoryPageSinkProvider pageSinkProvider;
 
     @Inject
     public MemoryConnector(
+            LifeCycleManager lifeCycleManager,
             MemoryMetadata metadata,
             MemorySplitManager splitManager,
             MemoryPageSourceProvider pageSourceProvider,
             MemoryPageSinkProvider pageSinkProvider)
     {
+        this.lifeCycleManager = requireNonNull(lifeCycleManager, "lifeCycleManager is null");
         this.metadata = metadata;
         this.splitManager = splitManager;
         this.pageSourceProvider = pageSourceProvider;
@@ -74,4 +80,10 @@ public ConnectorPageSinkProvider getPageSinkProvider()
     {
         return pageSinkProvider;
     }
+
+    @Override
+    public void shutdown()
+    {
+        lifeCycleManager.stop();
+    }
 }

plugin/trino-redis/src/main/java/io/trino/plugin/redis/RedisConnector.java

@@ -13,6 +13,7 @@
  */
 package io.trino.plugin.redis;
 
+import io.airlift.bootstrap.LifeCycleManager;
 import io.trino.spi.connector.Connector;
 import io.trino.spi.connector.ConnectorMetadata;
 import io.trino.spi.connector.ConnectorRecordSetProvider;
@@ -33,17 +34,20 @@
 public class RedisConnector
         implements Connector
 {
+    private final LifeCycleManager lifeCycleManager;
     private final RedisMetadata metadata;
 
     private final RedisSplitManager splitManager;
     private final RedisRecordSetProvider recordSetProvider;
 
     @Inject
     public RedisConnector(
+            LifeCycleManager lifeCycleManager,
             RedisMetadata metadata,
             RedisSplitManager splitManager,
             RedisRecordSetProvider recordSetProvider)
     {
+        this.lifeCycleManager = requireNonNull(lifeCycleManager, "lifeCycleManager is null");
         this.metadata = requireNonNull(metadata, "metadata is null");
         this.splitManager = requireNonNull(splitManager, "splitManager is null");
         this.recordSetProvider = requireNonNull(recordSetProvider, "recordSetProvider is null");
@@ -73,4 +77,10 @@ public ConnectorRecordSetProvider getRecordSetProvider()
     {
         return recordSetProvider;
     }
+
+    @Override
+    public void shutdown()
+    {
+        lifeCycleManager.stop();
+    }
 }