ChinmaySKulkarni commented on a change in pull request #771:
URL: https://github.com/apache/phoenix/pull/771#discussion_r416065566
##########
File path:
phoenix-core/src/test/java/org/apache/phoenix/compile/WhereOptimizerTest.java
##########
@@ -2440,11 +2439,37 @@ public void testEqualityAndGreaterThanRVC() throws
SQLException {
String query = "SELECT * FROM T WHERE A = 'C' and (A,B,C) >
('C','B','X') and C='C'";
QueryPlan queryPlan = TestUtil.getOptimizeQueryPlan(conn, query);
Scan scan = queryPlan.getContext().getScan();
+ //
+ // Note: The optimal scan boundary for the above query is ['CCC' -
*), however, I don't see an easy way to fix this currently so prioritizing.
Opened JIRA
assertArrayEquals(ByteUtil.concat(PChar.INSTANCE.toBytes("C"),
PChar.INSTANCE.toBytes("C"), PChar.INSTANCE.toBytes("C")), scan.getStartRow());
assertArrayEquals(PChar.INSTANCE.toBytes("D"), scan.getStopRow());
}
}
-
+
+ @Test
+ public void testEqualityAndGreaterThanRVC2() throws SQLException {
Review comment:
Looks like some RVC stuff got in here. Revert?
##########
File path:
phoenix-core/src/it/java/org/apache/phoenix/query/MaxConcurrentConnectionsIT.java
##########
@@ -0,0 +1,152 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.phoenix.query;
+
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.phoenix.end2end.BaseUniqueNamesOwnClusterIT;
+import org.apache.phoenix.jdbc.PhoenixDriver;
+import org.apache.phoenix.util.DelayedRegionServer;
+import org.apache.phoenix.util.PhoenixRuntime;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+
+import static
org.apache.phoenix.exception.SQLExceptionCode.NEW_CONNECTION_THROTTLED;
+import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_OPEN_INTERNAL_PHOENIX_CONNECTIONS;
+import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_OPEN_PHOENIX_CONNECTIONS;
+import static
org.apache.phoenix.query.QueryServices.CLIENT_CONNECTION_MAX_ALLOWED_CONNECTIONS;
+import static
org.apache.phoenix.query.QueryServices.INTERNAL_CONNECTION_MAX_ALLOWED_CONNECTIONS;
+import static org.apache.phoenix.query.QueryServices.RENEW_LEASE_ENABLED;
+import static org.apache.phoenix.util.PhoenixRuntime.JDBC_PROTOCOL_SEPARATOR;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class MaxConcurrentConnectionsIT extends BaseUniqueNamesOwnClusterIT {
+
+ private static HBaseTestingUtility hbaseTestUtil;
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ hbaseTestUtil = new HBaseTestingUtility();
+
hbaseTestUtil.startMiniCluster(1,1,null,null,DelayedRegionServer.class);
Review comment:
This API is deprecated. Use startMiniCluster with
`StartMiniClusterOption` instead.
##########
File path:
phoenix-core/src/it/java/org/apache/phoenix/end2end/QueryTimeoutIT.java
##########
@@ -139,4 +165,9 @@ public void testQueryTimeout() throws Exception {
}
conn.close();
}
+
+
+
+
+
Review comment:
Why do we need the changes in this file?
##########
File path:
phoenix-core/src/it/java/org/apache/phoenix/query/MaxConcurrentConnectionsIT.java
##########
@@ -0,0 +1,152 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.phoenix.query;
+
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.phoenix.end2end.BaseUniqueNamesOwnClusterIT;
+import org.apache.phoenix.jdbc.PhoenixDriver;
+import org.apache.phoenix.util.DelayedRegionServer;
+import org.apache.phoenix.util.PhoenixRuntime;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+
+import static
org.apache.phoenix.exception.SQLExceptionCode.NEW_CONNECTION_THROTTLED;
+import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_OPEN_INTERNAL_PHOENIX_CONNECTIONS;
+import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_OPEN_PHOENIX_CONNECTIONS;
+import static
org.apache.phoenix.query.QueryServices.CLIENT_CONNECTION_MAX_ALLOWED_CONNECTIONS;
+import static
org.apache.phoenix.query.QueryServices.INTERNAL_CONNECTION_MAX_ALLOWED_CONNECTIONS;
+import static org.apache.phoenix.query.QueryServices.RENEW_LEASE_ENABLED;
+import static org.apache.phoenix.util.PhoenixRuntime.JDBC_PROTOCOL_SEPARATOR;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class MaxConcurrentConnectionsIT extends BaseUniqueNamesOwnClusterIT {
+
+ private static HBaseTestingUtility hbaseTestUtil;
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ hbaseTestUtil = new HBaseTestingUtility();
+
hbaseTestUtil.startMiniCluster(1,1,null,null,DelayedRegionServer.class);
+ // establish url and quorum. Need to use PhoenixDriver and not
PhoenixTestDriver
+ String zkQuorum = "localhost:" +
hbaseTestUtil.getZkCluster().getClientPort();
+ url = PhoenixRuntime.JDBC_PROTOCOL + JDBC_PROTOCOL_SEPARATOR +
zkQuorum +
+ JDBC_PROTOCOL_SEPARATOR + "uniqueConn=A";
+ DriverManager.registerDriver(PhoenixDriver.INSTANCE);
+ }
+
+ private String getUniqueUrl() {
+ return url + generateUniqueName();
+ }
+
+ //Have to shutdown our special delayed region server
+ @AfterClass
+ public static void tearDown() throws Exception {
+ hbaseTestUtil.shutdownMiniCluster();
+ }
+
+ @Test
+ public void testCannotCreateOverMaximumConnections() throws Exception {
+ int maxConnections = 3;
+ Properties props = new Properties();
+
props.setProperty(CLIENT_CONNECTION_MAX_ALLOWED_CONNECTIONS,String.valueOf(maxConnections));
+ props.setProperty(RENEW_LEASE_ENABLED,String.valueOf(false));
+ assertEquals(0,GLOBAL_OPEN_PHOENIX_CONNECTIONS.getMetric().getValue());
+
+ List<Connection> connections = new ArrayList<>();
+ String url = getUniqueUrl();
+
+ try{
+ for(int i = 0; i < maxConnections; i++) {
+ connections.add(DriverManager.getConnection(url, props));
+ }
+
+ try {
+ DriverManager.getConnection(url, props);
+ fail();
+ } catch(SQLException e) {
+
assertEquals(NEW_CONNECTION_THROTTLED.getErrorCode(),e.getErrorCode());
+
assertEquals(NEW_CONNECTION_THROTTLED.getSQLState(),e.getSQLState());
+ }
+
+ } finally {
+ for(Connection conn : connections) {
+ conn.close();
+ }
+ }
+ }
+
+ /**
+ * This tests the delete path which creates a internal phoenix connection
per region
+ * @throws Exception
+ */
+ @Test
+ public void testDeleteRuntimeFailureClosesConnections() throws Exception {
+ String tableName = generateUniqueName();
+
+ //table with lots of regions
+ String ddl = "create table " + tableName + " (i integer not null
primary key, j integer) SALT_BUCKETS=256 ";
+
+ Properties props = new Properties();
+ props.setProperty(QueryServices.ENABLE_SERVER_SIDE_DELETE_MUTATIONS,
+ String.valueOf(false));
+
props.setProperty(CLIENT_CONNECTION_MAX_ALLOWED_CONNECTIONS,String.valueOf(10));
+
props.setProperty(INTERNAL_CONNECTION_MAX_ALLOWED_CONNECTIONS,String.valueOf(10));
+
+ String deleteStmt = "DELETE FROM " + tableName + " WHERE 20 = j";
+
+ assertEquals(0,GLOBAL_OPEN_PHOENIX_CONNECTIONS.getMetric().getValue());
+
assertEquals(0,GLOBAL_OPEN_INTERNAL_PHOENIX_CONNECTIONS.getMetric().getValue());
+ Connection conn = null;
+ try{
+ conn = DriverManager.getConnection(getUniqueUrl(), props);
+ try(Statement statement = conn.createStatement()) {
+ statement.execute(ddl);
+ }
+ //Enable delay for the delete
+ DelayedRegionServer.setDelayEnabled(true);
+ try(Statement statement = conn.createStatement()) {
+ statement.execute(deleteStmt);
+ }
+ fail();
+ } catch (SQLException e) {
+
assertEquals(NEW_CONNECTION_THROTTLED.getErrorCode(),e.getErrorCode());
Review comment:
So here, the client opens just 1 connection and the delete path creates
1 cloned conn per region and so this leads to client throttling? I thought
internal conns wouldn't get counted in the client conn throttling?
##########
File path:
phoenix-core/src/it/java/org/apache/phoenix/query/MaxConcurrentConnectionsIT.java
##########
@@ -0,0 +1,152 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.phoenix.query;
+
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.phoenix.end2end.BaseUniqueNamesOwnClusterIT;
+import org.apache.phoenix.jdbc.PhoenixDriver;
+import org.apache.phoenix.util.DelayedRegionServer;
+import org.apache.phoenix.util.PhoenixRuntime;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+
+import static
org.apache.phoenix.exception.SQLExceptionCode.NEW_CONNECTION_THROTTLED;
+import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_OPEN_INTERNAL_PHOENIX_CONNECTIONS;
+import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_OPEN_PHOENIX_CONNECTIONS;
+import static
org.apache.phoenix.query.QueryServices.CLIENT_CONNECTION_MAX_ALLOWED_CONNECTIONS;
+import static
org.apache.phoenix.query.QueryServices.INTERNAL_CONNECTION_MAX_ALLOWED_CONNECTIONS;
+import static org.apache.phoenix.query.QueryServices.RENEW_LEASE_ENABLED;
+import static org.apache.phoenix.util.PhoenixRuntime.JDBC_PROTOCOL_SEPARATOR;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class MaxConcurrentConnectionsIT extends BaseUniqueNamesOwnClusterIT {
+
+ private static HBaseTestingUtility hbaseTestUtil;
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ hbaseTestUtil = new HBaseTestingUtility();
+
hbaseTestUtil.startMiniCluster(1,1,null,null,DelayedRegionServer.class);
+ // establish url and quorum. Need to use PhoenixDriver and not
PhoenixTestDriver
+ String zkQuorum = "localhost:" +
hbaseTestUtil.getZkCluster().getClientPort();
+ url = PhoenixRuntime.JDBC_PROTOCOL + JDBC_PROTOCOL_SEPARATOR +
zkQuorum +
+ JDBC_PROTOCOL_SEPARATOR + "uniqueConn=A";
+ DriverManager.registerDriver(PhoenixDriver.INSTANCE);
+ }
+
+ private String getUniqueUrl() {
+ return url + generateUniqueName();
+ }
+
+ //Have to shutdown our special delayed region server
+ @AfterClass
+ public static void tearDown() throws Exception {
+ hbaseTestUtil.shutdownMiniCluster();
+ }
+
+ @Test
+ public void testCannotCreateOverMaximumConnections() throws Exception {
+ int maxConnections = 3;
+ Properties props = new Properties();
+
props.setProperty(CLIENT_CONNECTION_MAX_ALLOWED_CONNECTIONS,String.valueOf(maxConnections));
+ props.setProperty(RENEW_LEASE_ENABLED,String.valueOf(false));
+ assertEquals(0,GLOBAL_OPEN_PHOENIX_CONNECTIONS.getMetric().getValue());
+
+ List<Connection> connections = new ArrayList<>();
+ String url = getUniqueUrl();
+
+ try{
+ for(int i = 0; i < maxConnections; i++) {
+ connections.add(DriverManager.getConnection(url, props));
+ }
+
+ try {
+ DriverManager.getConnection(url, props);
+ fail();
+ } catch(SQLException e) {
+
assertEquals(NEW_CONNECTION_THROTTLED.getErrorCode(),e.getErrorCode());
+
assertEquals(NEW_CONNECTION_THROTTLED.getSQLState(),e.getSQLState());
+ }
+
+ } finally {
+ for(Connection conn : connections) {
+ conn.close();
+ }
+ }
+ }
+
+ /**
+ * This tests the delete path which creates a internal phoenix connection
per region
+ * @throws Exception
+ */
+ @Test
+ public void testDeleteRuntimeFailureClosesConnections() throws Exception {
+ String tableName = generateUniqueName();
+
+ //table with lots of regions
+ String ddl = "create table " + tableName + " (i integer not null
primary key, j integer) SALT_BUCKETS=256 ";
+
+ Properties props = new Properties();
+ props.setProperty(QueryServices.ENABLE_SERVER_SIDE_DELETE_MUTATIONS,
Review comment:
What happens if server-side mutations are enabled? If enabled, since we
no longer would need to `rs.next` and read all the data from the server +
generate delete mutations for them, is there any difference in behavior w.r.t.
whether we create cloned connections?
##########
File path:
phoenix-core/src/it/java/org/apache/phoenix/query/MaxConcurrentConnectionsIT.java
##########
@@ -0,0 +1,152 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.phoenix.query;
+
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.phoenix.end2end.BaseUniqueNamesOwnClusterIT;
+import org.apache.phoenix.jdbc.PhoenixDriver;
+import org.apache.phoenix.util.DelayedRegionServer;
+import org.apache.phoenix.util.PhoenixRuntime;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+
+import static
org.apache.phoenix.exception.SQLExceptionCode.NEW_CONNECTION_THROTTLED;
+import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_OPEN_INTERNAL_PHOENIX_CONNECTIONS;
+import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_OPEN_PHOENIX_CONNECTIONS;
+import static
org.apache.phoenix.query.QueryServices.CLIENT_CONNECTION_MAX_ALLOWED_CONNECTIONS;
+import static
org.apache.phoenix.query.QueryServices.INTERNAL_CONNECTION_MAX_ALLOWED_CONNECTIONS;
+import static org.apache.phoenix.query.QueryServices.RENEW_LEASE_ENABLED;
+import static org.apache.phoenix.util.PhoenixRuntime.JDBC_PROTOCOL_SEPARATOR;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class MaxConcurrentConnectionsIT extends BaseUniqueNamesOwnClusterIT {
+
+ private static HBaseTestingUtility hbaseTestUtil;
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ hbaseTestUtil = new HBaseTestingUtility();
+
hbaseTestUtil.startMiniCluster(1,1,null,null,DelayedRegionServer.class);
+ // establish url and quorum. Need to use PhoenixDriver and not
PhoenixTestDriver
+ String zkQuorum = "localhost:" +
hbaseTestUtil.getZkCluster().getClientPort();
+ url = PhoenixRuntime.JDBC_PROTOCOL + JDBC_PROTOCOL_SEPARATOR +
zkQuorum +
+ JDBC_PROTOCOL_SEPARATOR + "uniqueConn=A";
+ DriverManager.registerDriver(PhoenixDriver.INSTANCE);
+ }
+
+ private String getUniqueUrl() {
+ return url + generateUniqueName();
+ }
+
+ //Have to shutdown our special delayed region server
+ @AfterClass
+ public static void tearDown() throws Exception {
+ hbaseTestUtil.shutdownMiniCluster();
+ }
+
+ @Test
+ public void testCannotCreateOverMaximumConnections() throws Exception {
+ int maxConnections = 3;
+ Properties props = new Properties();
+
props.setProperty(CLIENT_CONNECTION_MAX_ALLOWED_CONNECTIONS,String.valueOf(maxConnections));
+ props.setProperty(RENEW_LEASE_ENABLED,String.valueOf(false));
+ assertEquals(0,GLOBAL_OPEN_PHOENIX_CONNECTIONS.getMetric().getValue());
+
+ List<Connection> connections = new ArrayList<>();
+ String url = getUniqueUrl();
+
+ try{
+ for(int i = 0; i < maxConnections; i++) {
+ connections.add(DriverManager.getConnection(url, props));
+ }
+
+ try {
+ DriverManager.getConnection(url, props);
+ fail();
+ } catch(SQLException e) {
+
assertEquals(NEW_CONNECTION_THROTTLED.getErrorCode(),e.getErrorCode());
+
assertEquals(NEW_CONNECTION_THROTTLED.getSQLState(),e.getSQLState());
+ }
+
+ } finally {
+ for(Connection conn : connections) {
+ conn.close();
+ }
+ }
+ }
+
+ /**
+ * This tests the delete path which creates a internal phoenix connection
per region
+ * @throws Exception
+ */
+ @Test
+ public void testDeleteRuntimeFailureClosesConnections() throws Exception {
+ String tableName = generateUniqueName();
+
+ //table with lots of regions
+ String ddl = "create table " + tableName + " (i integer not null
primary key, j integer) SALT_BUCKETS=256 ";
+
+ Properties props = new Properties();
+ props.setProperty(QueryServices.ENABLE_SERVER_SIDE_DELETE_MUTATIONS,
+ String.valueOf(false));
+
props.setProperty(CLIENT_CONNECTION_MAX_ALLOWED_CONNECTIONS,String.valueOf(10));
+
props.setProperty(INTERNAL_CONNECTION_MAX_ALLOWED_CONNECTIONS,String.valueOf(10));
+
+ String deleteStmt = "DELETE FROM " + tableName + " WHERE 20 = j";
+
+ assertEquals(0,GLOBAL_OPEN_PHOENIX_CONNECTIONS.getMetric().getValue());
+
assertEquals(0,GLOBAL_OPEN_INTERNAL_PHOENIX_CONNECTIONS.getMetric().getValue());
+ Connection conn = null;
+ try{
+ conn = DriverManager.getConnection(getUniqueUrl(), props);
+ try(Statement statement = conn.createStatement()) {
+ statement.execute(ddl);
+ }
+ //Enable delay for the delete
+ DelayedRegionServer.setDelayEnabled(true);
+ try(Statement statement = conn.createStatement()) {
+ statement.execute(deleteStmt);
+ }
+ fail();
+ } catch (SQLException e) {
+
assertEquals(NEW_CONNECTION_THROTTLED.getErrorCode(),e.getErrorCode());
+
assertEquals(NEW_CONNECTION_THROTTLED.getSQLState(),e.getSQLState());
+ } finally {
+ DelayedRegionServer.setDelayEnabled(false);
+ if(conn != null) {
+ conn.close();
+ }
+ long connections =
GLOBAL_OPEN_PHOENIX_CONNECTIONS.getMetric().getValue();
+ assertTrue(String.format("Found %d connections still
open.",connections),2 >= connections);
+ connections =
GLOBAL_OPEN_INTERNAL_PHOENIX_CONNECTIONS.getMetric().getValue();
Review comment:
I'm not sure I understand this condition. Aren't we expecting 10 conns?
##########
File path:
phoenix-core/src/it/java/org/apache/phoenix/query/MaxConcurrentConnectionsIT.java
##########
@@ -0,0 +1,152 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.phoenix.query;
+
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.phoenix.end2end.BaseUniqueNamesOwnClusterIT;
+import org.apache.phoenix.jdbc.PhoenixDriver;
+import org.apache.phoenix.util.DelayedRegionServer;
+import org.apache.phoenix.util.PhoenixRuntime;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+
+import static
org.apache.phoenix.exception.SQLExceptionCode.NEW_CONNECTION_THROTTLED;
+import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_OPEN_INTERNAL_PHOENIX_CONNECTIONS;
+import static
org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_OPEN_PHOENIX_CONNECTIONS;
+import static
org.apache.phoenix.query.QueryServices.CLIENT_CONNECTION_MAX_ALLOWED_CONNECTIONS;
+import static
org.apache.phoenix.query.QueryServices.INTERNAL_CONNECTION_MAX_ALLOWED_CONNECTIONS;
+import static org.apache.phoenix.query.QueryServices.RENEW_LEASE_ENABLED;
+import static org.apache.phoenix.util.PhoenixRuntime.JDBC_PROTOCOL_SEPARATOR;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class MaxConcurrentConnectionsIT extends BaseUniqueNamesOwnClusterIT {
+
+ private static HBaseTestingUtility hbaseTestUtil;
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ hbaseTestUtil = new HBaseTestingUtility();
+
hbaseTestUtil.startMiniCluster(1,1,null,null,DelayedRegionServer.class);
+ // establish url and quorum. Need to use PhoenixDriver and not
PhoenixTestDriver
+ String zkQuorum = "localhost:" +
hbaseTestUtil.getZkCluster().getClientPort();
+ url = PhoenixRuntime.JDBC_PROTOCOL + JDBC_PROTOCOL_SEPARATOR +
zkQuorum +
+ JDBC_PROTOCOL_SEPARATOR + "uniqueConn=A";
+ DriverManager.registerDriver(PhoenixDriver.INSTANCE);
+ }
+
+ private String getUniqueUrl() {
+ return url + generateUniqueName();
+ }
+
+ //Have to shutdown our special delayed region server
+ @AfterClass
+ public static void tearDown() throws Exception {
+ hbaseTestUtil.shutdownMiniCluster();
+ }
+
+ @Test
+ public void testCannotCreateOverMaximumConnections() throws Exception {
Review comment:
It looks like `PhoenixMetricsIT` already has tests for checking the
working on client-connection throttling. Do we need this extra test?
##########
File path:
phoenix-core/src/it/java/org/apache/phoenix/util/DelayedRegionServer.java
##########
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.phoenix.util;
+
+import com.google.protobuf.RpcController;
+import com.google.protobuf.ServiceException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.CoordinatedStateManager;
+import org.apache.hadoop.hbase.MiniHBaseCluster;
+import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
+import org.apache.hadoop.hbase.regionserver.HRegionServer;
+import org.apache.hadoop.hbase.regionserver.RSRpcServices;
+
+import java.io.IOException;
+
+public class DelayedRegionServer extends
MiniHBaseCluster.MiniHBaseClusterRegionServer {
+ static boolean doDelay = false;
+ // Activate the delays after table creation to test get/scan/put
+ private static int DELAY_GET = 0;
+ private static int DELAY_SCAN = 30000;
+ private static int DELAY_MUTATE = 0;
+
+ public static void setDelayEnabled(boolean delay) {
+ doDelay = delay;
+ }
+
+ public static void setDelayGet(int delayGet) {
+ DELAY_GET = delayGet;
+ }
+
+ public static void setDelayScan(int delayScan) {
+ DELAY_SCAN = delayScan;
+ }
+
+ public static void setDelayMutate(int delayMutate) {
+ DELAY_MUTATE = delayMutate;
+ }
+
+ public DelayedRegionServer(Configuration conf, CoordinatedStateManager cp)
+ throws IOException, InterruptedException {
+ super(conf, cp);
+ }
+
+ @Override protected RSRpcServices createRpcServices() throws IOException {
+ return new DelayedRSRpcServices(this);
+ }
+
+ /**
+ * This class injects delay for Rpc calls and after executes super methods
is delay is set.
+ */
+ public static class DelayedRSRpcServices extends RSRpcServices {
+
+ DelayedRSRpcServices(HRegionServer rs) throws IOException {
+ super(rs);
+ }
+
+ @Override public ClientProtos.GetResponse get(RpcController controller,
+ ClientProtos.GetRequest request) throws ServiceException {
+ try {
+ if (doDelay) {
+ Thread.sleep(DELAY_GET);
+ }
+ } catch (InterruptedException e) {
+ LOG.error("Sleep interrupted during get operation", e);
Review comment:
Shouldn't this class use its own logger?
##########
File path:
phoenix-core/src/it/java/org/apache/phoenix/util/DelayedRegionServer.java
##########
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.phoenix.util;
+
+import com.google.protobuf.RpcController;
+import com.google.protobuf.ServiceException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.CoordinatedStateManager;
+import org.apache.hadoop.hbase.MiniHBaseCluster;
+import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
+import org.apache.hadoop.hbase.regionserver.HRegionServer;
+import org.apache.hadoop.hbase.regionserver.RSRpcServices;
+
+import java.io.IOException;
+
+public class DelayedRegionServer extends
MiniHBaseCluster.MiniHBaseClusterRegionServer {
Review comment:
nit: Add class-level comment
##########
File path:
phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
##########
@@ -4957,12 +4964,22 @@ private void
returnAllSequences(ConcurrentMap<SequenceKey,Sequence> sequenceMap)
public void addConnection(PhoenixConnection connection) throws
SQLException {
if (returnSequenceValues || shouldThrottleNumConnections) {
synchronized (connectionCountLock) {
- if (shouldThrottleNumConnections && connectionCount + 1 >
maxConnectionsAllowed){
- GLOBAL_PHOENIX_CONNECTIONS_THROTTLED_COUNTER.increment();
- throw new
SQLExceptionInfo.Builder(SQLExceptionCode.NEW_CONNECTION_THROTTLED).
- build().buildException();
+
+ if(shouldThrottleNumConnections) {
Review comment:
The value of `shouldThrottleNumConnections` should now be set based on
this internal conn config and the max num conns from a client right?
##########
File path:
phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
##########
@@ -4957,12 +4964,22 @@ private void
returnAllSequences(ConcurrentMap<SequenceKey,Sequence> sequenceMap)
public void addConnection(PhoenixConnection connection) throws
SQLException {
if (returnSequenceValues || shouldThrottleNumConnections) {
synchronized (connectionCountLock) {
- if (shouldThrottleNumConnections && connectionCount + 1 >
maxConnectionsAllowed){
- GLOBAL_PHOENIX_CONNECTIONS_THROTTLED_COUNTER.increment();
- throw new
SQLExceptionInfo.Builder(SQLExceptionCode.NEW_CONNECTION_THROTTLED).
- build().buildException();
+
+ if(shouldThrottleNumConnections) {
+ int futureConnections = 1 + (
connection.isInternalConnection() ? internalConnectionCount : connectionCount);
+ int allowedConnections = connection.isInternalConnection()
? maxInternalConnectionsAllowed : maxConnectionsAllowed;
+ if(allowedConnections != 0 && futureConnections >
allowedConnections) {
+
GLOBAL_PHOENIX_CONNECTIONS_THROTTLED_COUNTER.increment();
+ throw new
SQLExceptionInfo.Builder(SQLExceptionCode.NEW_CONNECTION_THROTTLED).
Review comment:
Should we throw a different error for cases where internal conns exceed
the max config vs. when external do?
##########
File path:
phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
##########
@@ -4977,25 +4994,27 @@ public void removeConnection(PhoenixConnection
connection) throws SQLException {
if (returnSequenceValues) {
ConcurrentMap<SequenceKey,Sequence> formerSequenceMap = null;
synchronized (connectionCountLock) {
- if (--connectionCount <= 0) {
- if (!this.sequenceMap.isEmpty()) {
- formerSequenceMap = this.sequenceMap;
- this.sequenceMap = Maps.newConcurrentMap();
+ if(!connection.isInternalConnection()) {
Review comment:
We used to decrement `connectionCount` here before. Just commenting to
make sure we are still decrementing correctly.
##########
File path:
phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
##########
@@ -4957,12 +4964,22 @@ private void
returnAllSequences(ConcurrentMap<SequenceKey,Sequence> sequenceMap)
public void addConnection(PhoenixConnection connection) throws
SQLException {
if (returnSequenceValues || shouldThrottleNumConnections) {
synchronized (connectionCountLock) {
- if (shouldThrottleNumConnections && connectionCount + 1 >
maxConnectionsAllowed){
- GLOBAL_PHOENIX_CONNECTIONS_THROTTLED_COUNTER.increment();
- throw new
SQLExceptionInfo.Builder(SQLExceptionCode.NEW_CONNECTION_THROTTLED).
- build().buildException();
+
+ if(shouldThrottleNumConnections) {
+ int futureConnections = 1 + (
connection.isInternalConnection() ? internalConnectionCount : connectionCount);
+ int allowedConnections = connection.isInternalConnection()
? maxInternalConnectionsAllowed : maxConnectionsAllowed;
+ if(allowedConnections != 0 && futureConnections >
allowedConnections) {
+
GLOBAL_PHOENIX_CONNECTIONS_THROTTLED_COUNTER.increment();
Review comment:
Better to add a comment somewhere around here about the expected working
of counting internal phoenix conns and how/if this works against client conns,
etc.
##########
File path:
phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
##########
@@ -4977,25 +4994,27 @@ public void removeConnection(PhoenixConnection
connection) throws SQLException {
if (returnSequenceValues) {
ConcurrentMap<SequenceKey,Sequence> formerSequenceMap = null;
synchronized (connectionCountLock) {
- if (--connectionCount <= 0) {
- if (!this.sequenceMap.isEmpty()) {
- formerSequenceMap = this.sequenceMap;
- this.sequenceMap = Maps.newConcurrentMap();
+ if(!connection.isInternalConnection()) {
+ if (connectionCount + internalConnectionCount - 1 <= 0) {
Review comment:
hmm, can you explain this condition? Why does internalConnectionCount
come into the picture here?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
