PHOENIX-7638 Creating a large number of views leads to OS thread exhaustion (#2273)

Author: jpisaac

phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java

@@ -4745,9 +4745,9 @@ public MutationState addColumn(PTable table, List<ColumnDef> origColumnDefs,
           /**
            * To check if TTL is defined at any of the child below we are checking it at
            * {@link org.apache.phoenix.coprocessor.MetaDataEndpointImpl#mutateColumn(List, ColumnMutator, int, PTable, PTable, boolean)}
-           * level where in function {@link org.apache.phoenix.coprocessor.MetaDataEndpointImpl#
-           * validateIfMutationAllowedOnParent(PTable, List, PTableType, long, byte[], byte[],
-           * byte[], List, int)} we are already traversing through allDescendantViews.
+           * level where in function
+           * {@link org.apache.phoenix.coprocessor.MetaDataEndpointImpl# validateIfMutationAllowedOnParent(PTable, List, PTableType, long, byte[], byte[], byte[], List, int)}
+           * we are already traversing through allDescendantViews.
            */
         }
 

phoenix-core-client/src/main/java/org/apache/phoenix/util/TupleUtil.java

@@ -33,7 +33,6 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
-
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.CellUtil;
 import org.apache.hadoop.hbase.DoNotRetryIOException;
@@ -237,9 +236,9 @@ public static int write(Tuple result, DataOutput out) throws IOException {
    * @throws SQLException If any SQL operation fails.
    */
   @SuppressWarnings(value = "OBL_UNSATISFIED_OBLIGATION_EXCEPTION_EDGE",
-          justification = "Tge statement object needs to be kept open for the returned RS to be "
-          + "valid, however this is acceptable as not callingPhoenixStatement.close() "
-          + "causes no resource leak")
+      justification = "Tge statement object needs to be kept open for the returned RS to be "
+        + "valid, however this is acceptable as not callingPhoenixStatement.close() "
+        + "causes no resource leak")
   public static ResultSet getResultSet(Tuple toProject, TableName tableName, Connection conn,
     boolean withPrefetch) throws SQLException {
     if (tableName == null) {

phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/PhoenixRegionServerEndpoint.java

@@ -41,6 +41,7 @@
 import org.apache.phoenix.query.QueryServicesOptions;
 import org.apache.phoenix.util.ClientUtil;
 import org.apache.phoenix.util.SchemaUtil;
+import org.apache.phoenix.util.ServerUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -66,10 +67,12 @@ public void start(CoprocessorEnvironment env) throws IOException {
 
   @Override
   public void stop(CoprocessorEnvironment env) throws IOException {
+    RegionServerCoprocessor.super.stop(env);
     if (uncoveredIndexThreadPool != null) {
       uncoveredIndexThreadPool
         .stop("PhoenixRegionServerEndpoint is stopping. Shutting down uncovered index threadpool.");
     }
+    ServerUtil.ConnectionFactory.shutdown();
   }
 
   @Override

phoenix-core-server/src/main/java/org/apache/phoenix/util/ServerUtil.java

@@ -45,22 +45,16 @@
 import org.apache.hadoop.hbase.regionserver.Region.RowLock;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.phoenix.hbase.index.util.IndexManagementUtil;
-import org.apache.phoenix.hbase.index.util.VersionUtil;
 import org.apache.phoenix.hbase.index.write.IndexWriterUtils;
 import org.apache.phoenix.query.QueryServices;
 import org.apache.phoenix.query.QueryServicesOptions;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class ServerUtil {
-  private static final int COPROCESSOR_SCAN_WORKS = VersionUtil.encodeVersion("0.98.6");
   private static final Logger LOGGER = LoggerFactory.getLogger(ServerUtil.class);
   private static final String FORMAT_FOR_TIMESTAMP = ",serverTimestamp=%d,";
 
-  private static boolean coprocessorScanWorks(RegionCoprocessorEnvironment env) {
-    return (VersionUtil.encodeVersion(env.getHBaseVersion()) >= COPROCESSOR_SCAN_WORKS);
-  }
-
   public static boolean hasCoprocessor(RegionCoprocessorEnvironment env,
     String CoprocessorClassName) {
     Collection<CoprocessorDescriptor> coprocessors =
@@ -99,17 +93,11 @@ private static Table getTableFromSingletonPool(RegionCoprocessorEnvironment env,
 
   public static Table getHTableForCoprocessorScan(RegionCoprocessorEnvironment env,
     Table writerTable) throws IOException {
-    if (coprocessorScanWorks(env)) {
-      return writerTable;
-    }
     return getTableFromSingletonPool(env, writerTable.getName());
   }
 
   public static Table getHTableForCoprocessorScan(RegionCoprocessorEnvironment env,
     TableName tableName) throws IOException {
-    if (coprocessorScanWorks(env)) {
-      return env.getConnection().getTable(tableName);
-    }
     return getTableFromSingletonPool(env, tableName);
   }
 
@@ -222,6 +210,7 @@ public static Configuration getTypeSpecificConfiguration(ConnectionType connecti
 
     public static void shutdown() {
       synchronized (ConnectionFactory.class) {
+        LOGGER.info("Closing ServerUtil.ConnectionFactory connections");
         for (Connection connection : connections.values()) {
           try {
             connection.close();

phoenix-core/src/it/java/org/apache/phoenix/end2end/MetadataServerConnectionsIT.java

@@ -18,24 +18,45 @@
 package org.apache.phoenix.end2end;
 
 import static org.apache.phoenix.query.QueryServices.DISABLE_VIEW_SUBTREE_VALIDATION;
+import static org.apache.phoenix.query.QueryServicesTestImpl.DEFAULT_HCONNECTION_POOL_CORE_SIZE;
+import static org.apache.phoenix.query.QueryServicesTestImpl.DEFAULT_HCONNECTION_POOL_MAX_SIZE;
+import static org.apache.phoenix.util.PhoenixRuntime.TENANT_ID_ATTRIB;
+import static org.junit.Assert.assertEquals;
 
 import com.google.protobuf.RpcCallback;
 import com.google.protobuf.RpcController;
+import java.io.IOException;
+import java.lang.reflect.Field;
 import java.sql.Connection;
 import java.sql.DriverManager;
 import java.sql.ResultSet;
 import java.sql.Statement;
+import java.util.Arrays;
+import java.util.List;
 import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.hbase.CoprocessorEnvironment;
 import org.apache.hadoop.hbase.DoNotRetryIOException;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.ConnectionImplementation;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.coprocessor.CoprocessorException;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
 import org.apache.phoenix.coprocessor.MetaDataEndpointImpl;
 import org.apache.phoenix.coprocessor.generated.MetaDataProtos;
 import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
 import org.apache.phoenix.protobuf.ProtobufUtil;
 import org.apache.phoenix.query.BaseTest;
 import org.apache.phoenix.query.QueryServices;
 import org.apache.phoenix.util.ClientUtil;
+import org.apache.phoenix.util.MetaDataUtil;
 import org.apache.phoenix.util.ReadOnlyProps;
 import org.apache.phoenix.util.SchemaUtil;
+import org.apache.phoenix.util.ServerUtil;
 import org.apache.phoenix.util.TestUtil;
 import org.junit.Assert;
 import org.junit.BeforeClass;
@@ -64,6 +85,105 @@ public static synchronized void doSetup() throws Exception {
   }
 
   public static class TestMetaDataEndpointImpl extends MetaDataEndpointImpl {
+    private RegionCoprocessorEnvironment env;
+
+    public static void setTestCreateView(boolean testCreateView) {
+      TestMetaDataEndpointImpl.testCreateView = testCreateView;
+    }
+
+    private static volatile boolean testCreateView = false;
+
+    @Override
+    public void start(CoprocessorEnvironment env) throws IOException {
+      super.start(env);
+      if (env instanceof RegionCoprocessorEnvironment) {
+        this.env = (RegionCoprocessorEnvironment) env;
+      } else {
+        throw new CoprocessorException("Must be loaded on a table region!");
+      }
+    }
+
+    @Override
+    public void createTable(RpcController controller, MetaDataProtos.CreateTableRequest request,
+      RpcCallback<MetaDataProtos.MetaDataResponse> done) {
+      // Invoke the actual create table routine
+      super.createTable(controller, request, done);
+
+      byte[][] rowKeyMetaData = new byte[3][];
+      byte[] schemaName = null;
+      byte[] tableName = null;
+      String fullTableName = null;
+
+      // Get the singleton connection for testing
+      org.apache.hadoop.hbase.client.Connection conn = ServerUtil.ConnectionFactory
+        .getConnection(ServerUtil.ConnectionType.DEFAULT_SERVER_CONNECTION, env);
+      try {
+        // Get the current table creation details
+        List<Mutation> tableMetadata = ProtobufUtil.getMutations(request);
+        MetaDataUtil.getTenantIdAndSchemaAndTableName(tableMetadata, rowKeyMetaData);
+        schemaName = rowKeyMetaData[PhoenixDatabaseMetaData.SCHEMA_NAME_INDEX];
+        tableName = rowKeyMetaData[PhoenixDatabaseMetaData.TABLE_NAME_INDEX];
+        fullTableName = SchemaUtil.getTableName(schemaName, tableName);
+
+        ThreadPoolExecutor ctpe = null;
+        ThreadPoolExecutor htpe = null;
+
+        // Get the thread pool executor from the connection.
+        if (conn instanceof ConnectionImplementation) {
+          ConnectionImplementation connImpl = ((ConnectionImplementation) conn);
+          Field props = null;
+          props = ConnectionImplementation.class.getDeclaredField("batchPool");
+          props.setAccessible(true);
+          ctpe = (ThreadPoolExecutor) props.get(connImpl);
+          LOGGER.debug("ConnectionImplementation Thread pool info :" + ctpe.toString());
+
+        }
+
+        // Get the thread pool executor from the HTable.
+        Table hTable =
+          ServerUtil.getHTableForCoprocessorScan(env, TableName.valueOf(fullTableName));
+        if (hTable instanceof HTable) {
+          HTable testTable = (HTable) hTable;
+          Field props = testTable.getClass().getDeclaredField("pool");
+          props.setAccessible(true);
+          htpe = ((ThreadPoolExecutor) props.get(hTable));
+          LOGGER.debug("HTable Thread pool info :" + htpe.toString());
+          // Assert the HTable thread pool config match the Connection pool configs.
+          // Since we are not overriding any defaults, it should match the defaults.
+          assertEquals(htpe.getMaximumPoolSize(), DEFAULT_HCONNECTION_POOL_MAX_SIZE);
+          assertEquals(htpe.getCorePoolSize(), DEFAULT_HCONNECTION_POOL_CORE_SIZE);
+          LOGGER.debug("HTable threadpool info {}, {}, {}, {}", htpe.getCorePoolSize(),
+            htpe.getMaximumPoolSize(), htpe.getQueue().remainingCapacity(),
+            htpe.getKeepAliveTime(TimeUnit.SECONDS));
+
+          int count = Thread.activeCount();
+          Thread[] th = new Thread[count];
+          // returns the number of threads put into the array
+          Thread.enumerate(th);
+          long hTablePoolCount =
+            Arrays.stream(th).filter(s -> s.getName().equals("htable-pool-0")).count();
+          // Assert no default HTable threadpools are created.
+          assertEquals(0, hTablePoolCount);
+          LOGGER.debug("htable-pool-0 threads {}", hTablePoolCount);
+        }
+        // Assert that the threadpool from Connection and HTable are the same.
+        assertEquals(ctpe, htpe);
+      } catch (RuntimeException | NoSuchFieldException | IllegalAccessException | IOException t) {
+        // handle cases that an IOE is wrapped inside a RuntimeException
+        // like HTableInterface#createHTableInterface
+        MetaDataProtos.MetaDataResponse.Builder builder =
+          MetaDataProtos.MetaDataResponse.newBuilder();
+
+        LOGGER.error("This is unexpected");
+        ProtobufUtil.setControllerException(controller,
+          ClientUtil.createIOException(SchemaUtil
+            .getPhysicalTableName(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES, false)
+            .toString(), new DoNotRetryIOException("Not allowed")));
+        done.run(builder.build());
+
+      }
+
+    }
 
     @Override
     public void getVersion(RpcController controller, MetaDataProtos.GetVersionRequest request,
@@ -81,6 +201,67 @@ public void getVersion(RpcController controller, MetaDataProtos.GetVersionReques
     }
   }
 
+  @Test
+  public void testViewCreationAndServerConnections() throws Throwable {
+    final String tableName = generateUniqueName();
+    final String view01 = "v01_" + tableName;
+    final String view02 = "v02_" + tableName;
+    final String index_view01 = "idx_v01_" + tableName;
+    final String index_view02 = "idx_v02_" + tableName;
+    final String index_view03 = "idx_v03_" + tableName;
+    final String index_view04 = "idx_v04_" + tableName;
+    final int NUM_VIEWS = 50;
+
+    TestMetaDataEndpointImpl.setTestCreateView(true);
+    try (Connection conn = DriverManager.getConnection(getUrl())) {
+      TestUtil.removeCoprocessor(conn, "SYSTEM.CATALOG", MetaDataEndpointImpl.class);
+      TestUtil.addCoprocessor(conn, "SYSTEM.CATALOG", TestMetaDataEndpointImpl.class);
+
+      final Statement stmt = conn.createStatement();
+
+      stmt.execute("CREATE TABLE " + tableName
+        + " (COL1 CHAR(10) NOT NULL, COL2 CHAR(5) NOT NULL, COL3 VARCHAR,"
+        + " COL4 VARCHAR CONSTRAINT pk PRIMARY KEY(COL1, COL2))"
+        + " UPDATE_CACHE_FREQUENCY=ALWAYS, MULTI_TENANT=true");
+      conn.commit();
+
+      for (int i = 0; i < NUM_VIEWS; i++) {
+        Properties props = new Properties();
+        String viewTenantId = String.format("00T%012d", i);
+        props.setProperty(TENANT_ID_ATTRIB, viewTenantId);
+        // Create multilevel tenant views
+        try (Connection tConn = DriverManager.getConnection(getUrl(), props)) {
+          final Statement viewStmt = tConn.createStatement();
+          viewStmt
+            .execute("CREATE VIEW " + view01 + " (VCOL1 CHAR(8), COL5 VARCHAR) AS SELECT * FROM "
+              + tableName + " WHERE COL2 = 'col2'");
+
+          viewStmt.execute("CREATE VIEW " + view02 + " (VCOL2 CHAR(10), COL6 VARCHAR)"
+            + " AS SELECT * FROM " + view01 + " WHERE VCOL1 = 'vcol1'");
+          tConn.commit();
+
+          // Create multilevel tenant indexes
+          final Statement indexStmt = tConn.createStatement();
+          indexStmt.execute("CREATE INDEX " + index_view01 + " ON " + view01 + " (COL5) INCLUDE "
+            + "(COL1, COL2, COL3)");
+          indexStmt.execute("CREATE INDEX " + index_view02 + " ON " + view02 + " (COL6) INCLUDE "
+            + "(COL1, COL2, COL3)");
+          indexStmt.execute(
+            "CREATE INDEX " + index_view03 + " ON " + view01 + " (COL5) INCLUDE " + "(COL2, COL1)");
+          indexStmt.execute(
+            "CREATE INDEX " + index_view04 + " ON " + view02 + " (COL6) INCLUDE " + "(COL2, COL1)");
+
+          tConn.commit();
+
+        }
+
+      }
+
+      TestUtil.removeCoprocessor(conn, "SYSTEM.CATALOG", TestMetaDataEndpointImpl.class);
+      TestUtil.addCoprocessor(conn, "SYSTEM.CATALOG", MetaDataEndpointImpl.class);
+    }
+  }
+
   @Test
   public void testConnectionFromMetadataServer() throws Throwable {
     final String tableName = generateUniqueName();

phoenix-core/src/it/java/org/apache/phoenix/end2end/UpsertValuesIT.java

@@ -125,8 +125,8 @@ public void testPlainUpsertWithReturning() throws Exception {
     assertEquals("a", rs.getString(1));
     assertEquals("b", rs.getString(2));
     assertFalse(rs.next());
-    stmt = conn.prepareStatement("UPSERT INTO " + tableName
-            + " (inst,host,\"DATE\") VALUES(?,'b',CURRENT_DATE())");
+    stmt = conn.prepareStatement(
+      "UPSERT INTO " + tableName + " (inst,host,\"DATE\") VALUES(?,'b',CURRENT_DATE())");
     stmt.setString(1, "a");
     stmt.execute();
     rs = stmt.getResultSet();