This is an automated email from the ASF dual-hosted git repository.
tkhurana pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/phoenix.git
The following commit(s) were added to refs/heads/master by this push:
new 72865c53f4 PHOENIX-8023 SYSTEM.CATALOG close stalls on TTL lookup
self-scan (#2639)
72865c53f4 is described below
commit 72865c53f4db961177b0767174dc9ab30ca4c947
Author: tkhurana <[email protected]>
AuthorDate: Wed Oct 21 07:50:40 2026 -0700
PHOENIX-8023 SYSTEM.CATALOG close stalls on TTL lookup self-scan (#2639)
---
.../org/apache/phoenix/query/QueryServices.java | 5 +
.../apache/phoenix/query/QueryServicesOptions.java | 4 +
.../phoenix/coprocessor/MetaDataEndpointImpl.java | 194 ++++++++----
.../java/org/apache/phoenix/util/ServerUtil.java | 60 ++++
.../phoenix/end2end/SystemCatalogTTLLookupIT.java | 351 +++++++++++++++++++++
.../end2end/SystemCatalogTTLLookupSplitIT.java | 167 ++++++++++
.../org/apache/phoenix/util/ServerUtilTest.java | 117 +++++++
7 files changed, 828 insertions(+), 70 deletions(-)
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServices.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServices.java
index 085ac34a64..01640f5203 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServices.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServices.java
@@ -228,6 +228,11 @@ public interface QueryServices extends SQLCloseable {
public static final String METADATA_WRITE_RETRIES_NUMBER =
"phoenix.metadata.rpc.retries.number";
public static final String METADATA_WRITE_RETRY_PAUSE =
"phoenix.metadata.rpc.pause";
+ // Timeout (also bounding retries) for server side client scans of
SYSTEM.CATALOG issued while
+ // building a PTable
+ public static final String METADATA_SERVER_LOOKUP_TIMEOUT_MS =
+ "phoenix.metadata.server.lookup.timeout.ms";
+
// Config parameters for for configuring tracing
public static final String TRACING_FREQ_ATTRIB = "phoenix.trace.frequency";
public static final String TRACING_PAGE_SIZE_ATTRIB =
"phoenix.trace.read.pagesize";
diff --git
a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
index 4e3c29b6c3..4b20fb9da5 100644
---
a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
+++
b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
@@ -281,6 +281,10 @@ public class QueryServicesOptions {
public static final int DEFAULT_METADATA_WRITE_RETRIES_NUMBER = 20;
public static final int DEFAULT_METADATA_WRITE_RETRY_PAUSE = 100;
+ // Kept short because these reads run while holding a SYSTEM.CATALOG region
operation, at the
+ // cost of failing the lookup sooner against a slow region server
+ public static final int DEFAULT_METADATA_SERVER_LOOKUP_TIMEOUT_MS = 10000;
+
public static final int DEFAULT_TRACING_PAGE_SIZE = 100;
/**
* Configuration key to overwrite the tablename that should be used as the
target table
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
index 053976b586..966fce460b 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
@@ -154,6 +154,7 @@ import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
import org.apache.hadoop.hbase.coprocessor.CoprocessorException;
import org.apache.hadoop.hbase.coprocessor.CoreCoprocessor;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor;
@@ -1847,56 +1848,54 @@ public class MetaDataEndpointImpl extends
MetaDataProtocol implements RegionCopr
private TTLExpression getTTLFromHierarchy(byte[] viewKey, long
clientTimeStamp,
boolean checkForMappedView) throws IOException, SQLException {
Scan scan = MetaDataUtil.newTableRowsScan(viewKey, MIN_TABLE_TIMESTAMP,
clientTimeStamp);
- Table sysCat = ServerUtil.getHTableForCoprocessorScan(this.env,
- SchemaUtil.getPhysicalTableName(SYSTEM_CATALOG_NAME_BYTES,
env.getConfiguration()));
- ResultScanner scanner = sysCat.getScanner(scan);
- Result result = scanner.next();
-
byte[] tableKey = null;
- do {
-
- if (result == null) {
- return TTL_EXPRESSION_NOT_DEFINED;
- }
+ try (ResultScanner scanner = getSysCatScanner(viewKey, scan)) {
+ Result result = scanner.next();
+ do {
- // return TTL_NOT_DEFINED for Index on a Mapped View.
- if (checkForMappedView && checkIfViewIsMappedView(result)) {
- return TTL_EXPRESSION_NOT_DEFINED;
- }
+ if (result == null) {
+ return TTL_EXPRESSION_NOT_DEFINED;
+ }
- byte[] linkTypeBytes = result.getValue(TABLE_FAMILY_BYTES,
LINK_TYPE_BYTES);
- byte[][] rowKeyMetaData = new byte[5][];
- getVarChars(result.getRow(), 5, rowKeyMetaData);
- // Check if TTL is defined at the current given level
- if (result.getValue(TABLE_FAMILY_BYTES, TTL_BYTES) != null) {
- String ttlStr = (String) PVarchar.INSTANCE
- .toObject(result.getValue(DEFAULT_COLUMN_FAMILY_BYTES, TTL_BYTES));
- return TTLExpressionFactory.create(ttlStr);
- } else if (linkTypeBytes != null) {
- String parentSchema = SchemaUtil
-
.getSchemaNameFromFullName(rowKeyMetaData[PhoenixDatabaseMetaData.FAMILY_NAME_INDEX]);
- byte[] parentViewSchemaName =
- parentSchema != null ? parentSchema.getBytes(StandardCharsets.UTF_8)
: null;
- byte[] parentViewName = SchemaUtil
-
.getTableNameFromFullName(rowKeyMetaData[PhoenixDatabaseMetaData.FAMILY_NAME_INDEX])
- .getBytes(StandardCharsets.UTF_8);
- // Get TTL from up the hierarchy, Checking for Parent view link and
getting TTL from it.
- if (LinkType.fromSerializedValue(linkTypeBytes[0]) == PARENT_TABLE) {
- byte[] parentViewTenantId = result.getValue(TABLE_FAMILY_BYTES,
PARENT_TENANT_ID_BYTES);
- byte[] parentViewKey =
- SchemaUtil.getTableKey(parentViewTenantId, parentViewSchemaName,
parentViewName);
- return getTTLFromHierarchy(parentViewKey, clientTimeStamp, false);
+ // return TTL_NOT_DEFINED for Index on a Mapped View.
+ if (checkForMappedView && checkIfViewIsMappedView(result)) {
+ return TTL_EXPRESSION_NOT_DEFINED;
}
- // Store tableKey to use if we don't find TTL at current level and from
- // parent views above the hierarchy
- if (LinkType.fromSerializedValue(linkTypeBytes[0]) == PHYSICAL_TABLE) {
- tableKey = SchemaUtil.getTableKey(null, parentViewSchemaName,
parentViewName);
+ byte[] linkTypeBytes = result.getValue(TABLE_FAMILY_BYTES,
LINK_TYPE_BYTES);
+ byte[][] rowKeyMetaData = new byte[5][];
+ getVarChars(result.getRow(), 5, rowKeyMetaData);
+ // Check if TTL is defined at the current given level
+ if (result.getValue(TABLE_FAMILY_BYTES, TTL_BYTES) != null) {
+ String ttlStr = (String) PVarchar.INSTANCE
+ .toObject(result.getValue(DEFAULT_COLUMN_FAMILY_BYTES, TTL_BYTES));
+ return TTLExpressionFactory.create(ttlStr);
+ } else if (linkTypeBytes != null) {
+ String parentSchema = SchemaUtil
+
.getSchemaNameFromFullName(rowKeyMetaData[PhoenixDatabaseMetaData.FAMILY_NAME_INDEX]);
+ byte[] parentViewSchemaName =
+ parentSchema != null ?
parentSchema.getBytes(StandardCharsets.UTF_8) : null;
+ byte[] parentViewName = SchemaUtil
+
.getTableNameFromFullName(rowKeyMetaData[PhoenixDatabaseMetaData.FAMILY_NAME_INDEX])
+ .getBytes(StandardCharsets.UTF_8);
+ // Get TTL from up the hierarchy, Checking for Parent view link and
getting TTL from it.
+ if (LinkType.fromSerializedValue(linkTypeBytes[0]) == PARENT_TABLE) {
+ byte[] parentViewTenantId = result.getValue(TABLE_FAMILY_BYTES,
PARENT_TENANT_ID_BYTES);
+ byte[] parentViewKey =
+ SchemaUtil.getTableKey(parentViewTenantId, parentViewSchemaName,
parentViewName);
+ return getTTLFromHierarchy(parentViewKey, clientTimeStamp, false);
+ }
+
+ // Store tableKey to use if we don't find TTL at current level and
from
+ // parent views above the hierarchy
+ if (LinkType.fromSerializedValue(linkTypeBytes[0]) ==
PHYSICAL_TABLE) {
+ tableKey = SchemaUtil.getTableKey(null, parentViewSchemaName,
parentViewName);
+ }
}
- }
- result = scanner.next();
- } while (result != null);
+ result = scanner.next();
+ } while (result != null);
+ }
// Return TTL defined at Table level for the given hierarchy as we didn't
find TTL any of the
// views.
@@ -1925,39 +1924,94 @@ public class MetaDataEndpointImpl extends
MetaDataProtocol implements RegionCopr
private TTLExpression getTTLForParentTable(byte[] tableKey, long
clientTimeStamp,
String indexName, IndexType indexType) throws IOException {
Scan scan = MetaDataUtil.newTableRowsScan(tableKey, MIN_TABLE_TIMESTAMP,
clientTimeStamp);
- Table sysCat = ServerUtil.getHTableForCoprocessorScan(this.env,
- SchemaUtil.getPhysicalTableName(SYSTEM_CATALOG_NAME_BYTES,
env.getConfiguration()));
- ResultScanner scanner = sysCat.getScanner(scan);
- Result result = scanner.next();
- do {
- if (result == null) {
- return TTL_EXPRESSION_NOT_DEFINED;
- }
- if (result.getValue(TABLE_FAMILY_BYTES, TTL_BYTES) != null) {
- String ttlStr = (String) PVarchar.INSTANCE
- .toObject(result.getValue(DEFAULT_COLUMN_FAMILY_BYTES, TTL_BYTES));
- TTLExpression ttl = TTLExpressionFactory.create(ttlStr);
- if (UNCOVERED_GLOBAL.equals(indexType)) {
- if (CDCUtil.isCDCIndex(indexName)) {
- if (!ttl.equals(TTL_EXPRESSION_NOT_DEFINED)) {
- return TTL_EXPRESSION_FOREVER;
- }
- } else if (ttl instanceof ConditionalTTLExpression) {
- byte[] isStrictTTLBytes = result.getValue(TABLE_FAMILY_BYTES,
IS_STRICT_TTL_BYTES);
- boolean isStrictTTL = isStrictTTLBytes != null
- &&
Boolean.TRUE.equals(PBoolean.INSTANCE.toObject(isStrictTTLBytes));
- if (!isStrictTTL) {
- return TTL_EXPRESSION_NOT_DEFINED;
+ try (ResultScanner scanner = getSysCatScanner(tableKey, scan)) {
+ Result result = scanner.next();
+ do {
+ if (result == null) {
+ return TTL_EXPRESSION_NOT_DEFINED;
+ }
+ if (result.getValue(TABLE_FAMILY_BYTES, TTL_BYTES) != null) {
+ String ttlStr = (String) PVarchar.INSTANCE
+ .toObject(result.getValue(DEFAULT_COLUMN_FAMILY_BYTES, TTL_BYTES));
+ TTLExpression ttl = TTLExpressionFactory.create(ttlStr);
+ if (UNCOVERED_GLOBAL.equals(indexType)) {
+ if (CDCUtil.isCDCIndex(indexName)) {
+ if (!ttl.equals(TTL_EXPRESSION_NOT_DEFINED)) {
+ return TTL_EXPRESSION_FOREVER;
+ }
+ } else if (ttl instanceof ConditionalTTLExpression) {
+ byte[] isStrictTTLBytes = result.getValue(TABLE_FAMILY_BYTES,
IS_STRICT_TTL_BYTES);
+ boolean isStrictTTL = isStrictTTLBytes != null
+ &&
Boolean.TRUE.equals(PBoolean.INSTANCE.toObject(isStrictTTLBytes));
+ if (!isStrictTTL) {
+ return TTL_EXPRESSION_NOT_DEFINED;
+ }
}
}
+ return ttl;
}
- return ttl;
- }
- result = scanner.next();
- } while (result != null);
+ result = scanner.next();
+ } while (result != null);
+ }
return TTL_EXPRESSION_NOT_DEFINED;
}
+ /**
+ * Opens a scanner over SYSTEM.CATALOG rows for the given table key. Reads
the local region when
+ * it hosts the key, so that no client RPC is issued to a region whose
operation the caller may
+ * hold (a closing region would otherwise stall in client retries).
Otherwise falls back to a
+ * client scan with bounded retries and timeouts.
+ */
+ private ResultScanner getSysCatScanner(byte[] key, Scan scan) throws
IOException {
+ Region region = env.getRegion();
+ if (region.getRegionInfo().containsRow(key)) {
+ return new RegionResultScanner(region.getScanner(scan));
+ }
+ try (Table sysCat = ServerUtil.ConnectionFactory
+ .getConnection(ServerUtil.ConnectionType.METADATA_LOOKUP_CONNECTION,
env).getTable(
+ SchemaUtil.getPhysicalTableName(SYSTEM_CATALOG_NAME_BYTES,
env.getConfiguration()))) {
+ return sysCat.getScanner(scan);
+ }
+ }
+
+ /** Adapts a {@link RegionScanner} to a {@link ResultScanner}, one row per
{@link Result}. */
+ private static class RegionResultScanner implements ResultScanner {
+ private final RegionScanner scanner;
+ private boolean hasMore = true;
+
+ RegionResultScanner(RegionScanner scanner) {
+ this.scanner = scanner;
+ }
+
+ @Override
+ public Result next() throws IOException {
+ List<Cell> cells = new ArrayList<>();
+ while (hasMore && cells.isEmpty()) {
+ hasMore = scanner.next(cells);
+ }
+ return cells.isEmpty() ? null : Result.create(cells);
+ }
+
+ @Override
+ public void close() {
+ try {
+ scanner.close();
+ } catch (IOException e) {
+ LOGGER.warn("Failed to close SYSTEM.CATALOG region scanner", e);
+ }
+ }
+
+ @Override
+ public boolean renewLease() {
+ return false;
+ }
+
+ @Override
+ public ScanMetrics getScanMetrics() {
+ return null;
+ }
+ }
+
private Long getViewIndexId(Cell[] tableKeyValues, PDataType
viewIndexIdType) {
Cell viewIndexIdKv = tableKeyValues[VIEW_INDEX_ID_INDEX];
return viewIndexIdKv == null ? null : decodeViewIndexId(viewIndexIdKv,
viewIndexIdType);
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/util/ServerUtil.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/util/ServerUtil.java
index 4a409ac69e..6f5a3b1190 100644
--- a/phoenix-core-server/src/main/java/org/apache/phoenix/util/ServerUtil.java
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/util/ServerUtil.java
@@ -54,6 +54,13 @@ import org.slf4j.LoggerFactory;
public class ServerUtil {
private static final Logger LOGGER =
LoggerFactory.getLogger(ServerUtil.class);
private static final String FORMAT_FOR_TIMESTAMP = ",serverTimestamp=%d,";
+ // HBase client keys not available as constants in all supported HBase
versions
+ private static final String HBASE_CLIENT_META_READ_RPC_TIMEOUT_KEY =
+ "hbase.client.meta.read.rpc.timeout";
+ private static final String HBASE_CLIENT_META_SCANNER_TIMEOUT =
+ "hbase.client.meta.scanner.timeout.period";
+ private static final String HBASE_CLIENT_PAUSE_FOR_SERVER_OVERLOADED =
+ "hbase.client.pause.server.overloaded";
public static boolean hasCoprocessor(RegionCoprocessorEnvironment env,
String CoprocessorClassName) {
@@ -167,6 +174,7 @@ public class ServerUtil {
INDEX_WRITER_CONNECTION,
INDEX_WRITER_CONNECTION_WITH_CUSTOM_THREADS,
INDEX_WRITER_CONNECTION_WITH_CUSTOM_THREADS_NO_RETRIES,
+ METADATA_LOOKUP_CONNECTION,
DEFAULT_SERVER_CONNECTION;
}
@@ -203,6 +211,8 @@ public class ServerUtil {
return getIndexWriterConfigurationWithCustomThreads(conf);
case INDEX_WRITER_CONNECTION_WITH_CUSTOM_THREADS_NO_RETRIES:
return getNoRetriesIndexWriterConfigurationWithCustomThreads(conf);
+ case METADATA_LOOKUP_CONNECTION:
+ return getMetadataConfig(conf);
default:
return conf;
}
@@ -240,6 +250,56 @@ public class ServerUtil {
return compactionConfig;
}
+ public static Configuration getMetadataConfig(Configuration conf) {
+ Configuration metadataConfig = PropertiesUtil.cloneConfig(conf);
+ // fail fast, as the caller holds a SYSTEM.CATALOG region operation that
blocks its close
+ int timeoutMs =
conf.getInt(QueryServices.METADATA_SERVER_LOOKUP_TIMEOUT_MS,
+ QueryServicesOptions.DEFAULT_METADATA_SERVER_LOOKUP_TIMEOUT_MS);
+ if (timeoutMs <= 0) {
+ // HBase treats a non-positive timeout as no timeout
+ LOGGER.warn("Ignoring non-positive {}={}, using {}",
+ QueryServices.METADATA_SERVER_LOOKUP_TIMEOUT_MS, timeoutMs,
+ QueryServicesOptions.DEFAULT_METADATA_SERVER_LOOKUP_TIMEOUT_MS);
+ timeoutMs =
QueryServicesOptions.DEFAULT_METADATA_SERVER_LOOKUP_TIMEOUT_MS;
+ }
+ metadataConfig.setInt(HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD,
timeoutMs);
+ metadataConfig.setInt(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT,
timeoutMs);
+ metadataConfig.setInt(HConstants.HBASE_RPC_READ_TIMEOUT_KEY, timeoutMs);
+ // region location lookups use the meta timeouts, which default to much
longer ones
+ metadataConfig.setInt(HConstants.HBASE_CLIENT_META_OPERATION_TIMEOUT,
timeoutMs);
+ metadataConfig.setInt(HBASE_CLIENT_META_READ_RPC_TIMEOUT_KEY, timeoutMs);
+ metadataConfig.setInt(HBASE_CLIENT_META_SCANNER_TIMEOUT, timeoutMs);
+ // region location lookups retry without checking the timeout, so bound
the retries too,
+ // using the larger pause that is applied when the server is overloaded
+ long pauseMs =
+ conf.getLong(HConstants.HBASE_CLIENT_PAUSE,
HConstants.DEFAULT_HBASE_CLIENT_PAUSE);
+ long overloadedPauseMs =
conf.getLong(HBASE_CLIENT_PAUSE_FOR_SERVER_OVERLOADED,
+ conf.getLong(HConstants.HBASE_CLIENT_PAUSE_FOR_CQTBE, pauseMs));
+ int retries = getRetriesWithinTimeout(timeoutMs, Math.max(pauseMs,
overloadedPauseMs));
+ metadataConfig.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, retries);
+ LOGGER.info("SYSTEM.CATALOG lookup connection timeout={} ms, retries={}",
timeoutMs, retries);
+ return metadataConfig;
+ }
+
+ /**
+ * Returns the largest number of retries whose total nominal client backoff
(see
+ * {@link HConstants#RETRY_BACKOFF}, excluding jitter and RPC time) fits
within the given timeout.
+ */
+ static int getRetriesWithinTimeout(long timeoutMs, long pauseMs) {
+ long pause = Math.max(1, pauseMs);
+ int retries = 0;
+ long totalBackoff = 0;
+ while (true) {
+ long backoff =
+ pause * HConstants.RETRY_BACKOFF[Math.min(retries,
HConstants.RETRY_BACKOFF.length - 1)];
+ if (totalBackoff + backoff > timeoutMs) {
+ return retries;
+ }
+ totalBackoff += backoff;
+ retries++;
+ }
+ }
+
public static Configuration getIndexWriterConnection(Configuration conf) {
Configuration clonedConfig = PropertiesUtil.cloneConfig(conf);
/*
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/end2end/SystemCatalogTTLLookupIT.java
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/SystemCatalogTTLLookupIT.java
new file mode 100644
index 0000000000..0b5654d20f
--- /dev/null
+++
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/SystemCatalogTTLLookupIT.java
@@ -0,0 +1,351 @@
+/*
+ * 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.end2end;
+
+import static org.apache.phoenix.query.BaseTest.generateUniqueName;
+import static org.apache.phoenix.query.BaseTest.setUpConfigForMiniCluster;
+import static
org.apache.phoenix.schema.LiteralTTLExpression.TTL_EXPRESSION_NOT_DEFINED;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.util.List;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.coprocessor.ObserverContext;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.hadoop.hbase.coprocessor.RegionObserver;
+import org.apache.hadoop.hbase.regionserver.HRegion;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
+import org.apache.phoenix.jdbc.PhoenixTestDriver;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.util.PhoenixRuntime;
+import org.apache.phoenix.util.QueryUtil;
+import org.apache.phoenix.util.SchemaUtil;
+import org.apache.phoenix.util.TestUtil;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+/**
+ * Verifies that resolving TTL from the view/table/index hierarchy while
building a PTable reads
+ * SYSTEM.CATALOG from the local region instead of issuing a client scan to
it. Such a client scan
+ * targets the region whose operation the outer getTable holds, and stalls the
region's close until
+ * the client scanner timeout.
+ */
+@Category(NeedsOwnMiniClusterTest.class)
+public class SystemCatalogTTLLookupIT {
+ private static final TableName SYSTEM_CATALOG =
+ TableName.valueOf(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME);
+ private static final long MAX_CLOSE_MILLIS = 10000;
+ private static final long WAIT_MILLIS = 120000;
+
+ private static HBaseTestingUtility utility;
+ private static String url;
+
+ @BeforeClass
+ public static synchronized void doSetup() throws Exception {
+ Configuration conf = HBaseConfiguration.create();
+ setUpConfigForMiniCluster(conf);
+ // Restore the default client retries and scanner timeout (the test config
lowers the former and
+ // raises the latter) so that a client scan of a closing region retries
until the scanner
+ // timeout, as it does in production.
+ conf.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,
+ HConstants.DEFAULT_HBASE_CLIENT_RETRIES_NUMBER);
+ conf.setInt(HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD,
+ HConstants.DEFAULT_HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD);
+ utility = new HBaseTestingUtility(conf);
+ utility.startMiniCluster();
+ // Same test driver URL as the server side connections (e.g. for index
creation) build.
+ url = QueryUtil.getConnectionUrl(new Properties(),
utility.getConfiguration())
+ + PhoenixRuntime.PHOENIX_TEST_DRIVER_URL_PARAM;
+ DriverManager.registerDriver(new PhoenixTestDriver());
+ }
+
+ @AfterClass
+ public static synchronized void tearDown() throws Exception {
+ if (utility != null) {
+ utility.shutdownMiniCluster();
+ }
+ }
+
+ /**
+ * Counts client scans (RPC path only) of SYSTEM.CATALOG starting at any
watched row key, and
+ * optionally runs a hook on the first one.
+ */
+ public static class ScanCountingObserver implements RegionCoprocessor,
RegionObserver {
+ static final List<byte[]> WATCHED_KEYS = new CopyOnWriteArrayList<>();
+ static final AtomicInteger SCANS = new AtomicInteger();
+ static final AtomicReference<Runnable> ON_FIRST_SCAN = new
AtomicReference<>();
+
+ @Override
+ public Optional<RegionObserver> getRegionObserver() {
+ return Optional.of(this);
+ }
+
+ @Override
+ public void preScannerOpen(ObserverContext<RegionCoprocessorEnvironment>
c, Scan scan)
+ throws IOException {
+ for (byte[] key : WATCHED_KEYS) {
+ if (Bytes.equals(key, scan.getStartRow())) {
+ SCANS.incrementAndGet();
+ Runnable hook = ON_FIRST_SCAN.getAndSet(null);
+ if (hook != null) {
+ hook.run();
+ }
+ }
+ }
+ }
+ }
+
+ @Test
+ public void testTableWithIndexesDoesNotScanSystemCatalogViaRpc() throws
Exception {
+ String tableName = generateUniqueName();
+ String indexName1 = generateUniqueName();
+ String indexName2 = generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(url)) {
+ createTableWithTTL(conn, tableName);
+ conn.createStatement()
+ .execute("CREATE INDEX " + indexName1 + " ON " + tableName + " (V1)
INCLUDE (V2)");
+ conn.createStatement().execute("CREATE INDEX " + indexName2 + " ON " +
tableName + " (V2)");
+ }
+ watchTableKeys(tableName);
+
+ try (Connection conn = DriverManager.getConnection(url)) {
+ PhoenixConnection pconn = clearCachesAndResetCount(conn);
+ // Building the table also builds each of its indexes, which resolve TTL
from the table.
+ PTable table = pconn.getTableNoCache(tableName);
+
+ assertNotEquals(TTL_EXPRESSION_NOT_DEFINED, table.getTTLExpression());
+ assertEquals(2, table.getIndexes().size());
+ for (PTable index : table.getIndexes()) {
+ assertEquals(table.getTTLExpression(), index.getTTLExpression());
+ }
+ assertNoSystemCatalogClientScans();
+ }
+ }
+
+ @Test
+ public void testViewHierarchyDoesNotScanSystemCatalogViaRpc() throws
Exception {
+ String tableName = generateUniqueName();
+ String parentViewName = generateUniqueName();
+ String childViewName = generateUniqueName();
+ String viewIndexName = generateUniqueName();
+ createViewHierarchy(tableName, parentViewName, childViewName);
+ try (Connection conn = DriverManager.getConnection(url)) {
+ conn.createStatement()
+ .execute("CREATE INDEX " + viewIndexName + " ON " + childViewName + "
(V1)");
+ }
+ watchTableKeys(tableName, parentViewName, childViewName);
+
+ try (Connection conn = DriverManager.getConnection(url)) {
+ PhoenixConnection pconn = clearCachesAndResetCount(conn);
+ PTable table = pconn.getTableNoCache(tableName);
+ // Building the view also builds its index, which resolves TTL from the
view hierarchy.
+ PTable childView = pconn.getTableNoCache(childViewName);
+
+ assertNotEquals(TTL_EXPRESSION_NOT_DEFINED, table.getTTLExpression());
+ assertEquals(table.getTTLExpression(), childView.getTTLExpression());
+ assertEquals(1, childView.getIndexes().size());
+ assertEquals(table.getTTLExpression(),
childView.getIndexes().get(0).getTTLExpression());
+ assertNoSystemCatalogClientScans();
+ }
+ }
+
+ /**
+ * Closes SYSTEM.CATALOG while a getTable is resolving TTL. If the TTL
lookup issues a client
+ * scan, the close is started from that scan's preScannerOpen hook, so the
scan then finds the
+ * region closing and retries for the scanner timeout while the outer
getTable holds the region
+ * operation. Otherwise the region is closed once the getTable completes.
Either way the close
+ * must not stall.
+ */
+ @Test
+ public void testSystemCatalogCloseIsNotStalledByTTLLookup() throws Exception
{
+ String tableName = generateUniqueName();
+ String parentViewName = generateUniqueName();
+ String childViewName = generateUniqueName();
+ createViewHierarchy(tableName, parentViewName, childViewName);
+ watchTableKeys(tableName, parentViewName, childViewName);
+
+ HRegion sysCatRegion =
utility.getHBaseCluster().getRegions(SYSTEM_CATALOG).get(0);
+ byte[] regionName = sysCatRegion.getRegionInfo().getRegionName();
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ AtomicReference<Future<Long>> close = new AtomicReference<>();
+ AtomicBoolean closeStarted = new AtomicBoolean();
+ AtomicBoolean closingSeen = new AtomicBoolean();
+ try (Admin admin = utility.getAdmin()) {
+ Throwable failure = null;
+ try {
+ ScanCountingObserver.ON_FIRST_SCAN.set(() -> {
+ close.set(executor.submit(() -> timedUnassign(admin, regionName)));
+ closingSeen.set(waitUntilClosing(sysCatRegion));
+ closeStarted.set(true);
+ });
+ Future<?> lookup = executor.submit(() -> {
+ try (Connection conn = DriverManager.getConnection(url)) {
+ clearCachesAndResetCount(conn).getTableNoCache(childViewName);
+ }
+ return null;
+ });
+ // Wait for either the hook to start the close, or the lookup to
complete without it.
+ long deadline = System.currentTimeMillis() + WAIT_MILLIS;
+ while (!closeStarted.get() && !lookup.isDone()) {
+ assertTrue("TTL lookup did not complete", System.currentTimeMillis()
< deadline);
+ Thread.sleep(10);
+ }
+ ScanCountingObserver.ON_FIRST_SCAN.set(null);
+ if (closeStarted.get()) {
+ assertTrue("SYSTEM.CATALOG did not start closing",
closingSeen.get());
+ } else {
+ lookup.get(WAIT_MILLIS, TimeUnit.MILLISECONDS);
+ close.set(executor.submit(() -> timedUnassign(admin, regionName)));
+ }
+ long closeMillis = close.get().get(WAIT_MILLIS, TimeUnit.MILLISECONDS);
+ assertTrue("SYSTEM.CATALOG close took " + closeMillis + " ms",
+ closeMillis < MAX_CLOSE_MILLIS);
+ } catch (Throwable t) {
+ failure = t;
+ throw t;
+ } finally {
+ ScanCountingObserver.ON_FIRST_SCAN.set(null);
+ executor.shutdownNow();
+ executor.awaitTermination(WAIT_MILLIS, TimeUnit.MILLISECONDS);
+ restoreSystemCatalog(admin, regionName, close.get(), failure);
+ }
+ }
+ }
+
+ /**
+ * Brings SYSTEM.CATALOG back for the other tests if it was closed, without
masking the failure of
+ * the test, if any.
+ */
+ private static void restoreSystemCatalog(Admin admin, byte[] regionName,
Future<Long> close,
+ Throwable failure) throws Exception {
+ if (close == null) {
+ return;
+ }
+ try {
+ try {
+ close.get(WAIT_MILLIS, TimeUnit.MILLISECONDS);
+ } catch (ExecutionException | TimeoutException e) {
+ // The close failed or is still running; assign below only if the
region is offline.
+ }
+ if (utility.getHBaseCluster().getRegions(SYSTEM_CATALOG).isEmpty()) {
+ admin.assign(regionName);
+ }
+ utility.waitUntilAllRegionsAssigned(SYSTEM_CATALOG);
+ } catch (Exception e) {
+ if (failure == null) {
+ throw e;
+ }
+ failure.addSuppressed(e);
+ }
+ }
+
+ private static long timedUnassign(Admin admin, byte[] regionName) throws
IOException {
+ long start = System.currentTimeMillis();
+ admin.unassign(regionName);
+ return System.currentTimeMillis() - start;
+ }
+
+ private static boolean waitUntilClosing(HRegion region) {
+ long deadline = System.currentTimeMillis() + MAX_CLOSE_MILLIS;
+ while (!region.isClosing() && !region.isClosed()) {
+ if (System.currentTimeMillis() >= deadline) {
+ return false;
+ }
+ try {
+ Thread.sleep(10);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static void createTableWithTTL(Connection conn, String tableName)
throws Exception {
+ conn.createStatement().execute("CREATE TABLE " + tableName
+ + " (ID VARCHAR NOT NULL PRIMARY KEY, V1 VARCHAR, V2 VARCHAR)
TTL=86400");
+ TestUtil.addCoprocessor(conn, PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME,
+ ScanCountingObserver.class);
+ }
+
+ private static void createViewHierarchy(String tableName, String
parentViewName,
+ String childViewName) throws Exception {
+ try (Connection conn = DriverManager.getConnection(url)) {
+ createTableWithTTL(conn, tableName);
+ conn.createStatement()
+ .execute("CREATE VIEW " + parentViewName + " AS SELECT * FROM " +
tableName);
+ conn.createStatement()
+ .execute("CREATE VIEW " + childViewName + " AS SELECT * FROM " +
parentViewName);
+ }
+ }
+
+ private static void watchTableKeys(String... names) {
+ ScanCountingObserver.WATCHED_KEYS.clear();
+ for (String name : names) {
+ ScanCountingObserver.WATCHED_KEYS.add(SchemaUtil.getTableKey(null, null,
name));
+ }
+ }
+
+ private static PhoenixConnection clearCachesAndResetCount(Connection conn)
throws Exception {
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+ // Clear the client and server metadata caches so that getTable rebuilds
the PTables.
+ pconn.getQueryServices().clearCache();
+ ScanCountingObserver.SCANS.set(0);
+ return pconn;
+ }
+
+ private static void assertNoSystemCatalogClientScans() throws IOException {
+ assertEquals("TTL lookup must not issue client scans to SYSTEM.CATALOG", 0,
+ ScanCountingObserver.SCANS.get());
+ // Positive control: a client scan of a watched key is counted.
+ try (Table sysCat = utility.getConnection().getTable(SYSTEM_CATALOG);
ResultScanner scanner =
+ sysCat.getScanner(new
Scan().withStartRow(ScanCountingObserver.WATCHED_KEYS.get(0)))) {
+ scanner.next();
+ }
+ assertEquals(1, ScanCountingObserver.SCANS.get());
+ }
+}
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/end2end/SystemCatalogTTLLookupSplitIT.java
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/SystemCatalogTTLLookupSplitIT.java
new file mode 100644
index 0000000000..fcdcc33431
--- /dev/null
+++
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/SystemCatalogTTLLookupSplitIT.java
@@ -0,0 +1,167 @@
+/*
+ * 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.end2end;
+
+import static
org.apache.phoenix.schema.LiteralTTLExpression.TTL_EXPRESSION_NOT_DEFINED;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.util.Properties;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.RegionLocator;
+import
org.apache.phoenix.end2end.SystemCatalogTTLLookupIT.ScanCountingObserver;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.util.PhoenixRuntime;
+import org.apache.phoenix.util.SchemaUtil;
+import org.apache.phoenix.util.TestUtil;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+/**
+ * Verifies that resolving TTL from a view hierarchy spread over several
SYSTEM.CATALOG regions
+ * reads the ancestors held by other regions through client scans (and only
those), and resolves the
+ * table's TTL.
+ */
+@Category(NeedsOwnMiniClusterTest.class)
+public class SystemCatalogTTLLookupSplitIT extends SplitSystemCatalogIT {
+ private static final TableName SYSTEM_CATALOG =
+ TableName.valueOf(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME);
+
+ @Test
+ public void testTenantViewOnTableInAnotherRegion() throws Exception {
+ String tableName = SchemaUtil.getTableName(SCHEMA1, generateUniqueName());
+ String viewName = SchemaUtil.getTableName(SCHEMA2, generateUniqueName());
+ createTableWithTTL(tableName);
+ createTenantView(viewName, tableName);
+ byte[] tableKey = SchemaUtil.getTableKeyFromFullName(tableName);
+ byte[] viewKey = getTenantTableKey(viewName);
+ assertInDifferentRegions(tableKey, viewKey);
+
+ try (Connection conn = getTenantConnection()) {
+ // The view is read from the local region.
+ assertScansWhileBuilding(conn, viewName, viewKey, 0);
+ // The table is client-scanned in its own region.
+ PTable view = assertScansWhileBuilding(conn, viewName, tableKey, 1);
+ assertTTLOf(tableName, view);
+ }
+ }
+
+ @Test
+ public void testTenantViewHierarchyAcrossRegions() throws Exception {
+ String tableName = SchemaUtil.getTableName(SCHEMA1, generateUniqueName());
+ String parentViewName = SchemaUtil.getTableName(SCHEMA2,
generateUniqueName());
+ String childViewName = SchemaUtil.getTableName(SCHEMA3,
generateUniqueName());
+ createTableWithTTL(tableName);
+ createTenantView(parentViewName, tableName);
+ createTenantView(childViewName, parentViewName);
+ byte[] tableKey = SchemaUtil.getTableKeyFromFullName(tableName);
+ byte[] parentViewKey = getTenantTableKey(parentViewName);
+ byte[] childViewKey = getTenantTableKey(childViewName);
+ assertInDifferentRegions(tableKey, parentViewKey, childViewKey);
+
+ try (Connection conn = getTenantConnection()) {
+ // The child view is read from the local region.
+ assertScansWhileBuilding(conn, childViewName, childViewKey, 0);
+ // The parent view is client-scanned in the parent's region, only by the
child view's TTL
+ // recursion. (Without the fix, building the parent view, which the
client resolves along
+ // with the child view, would also client-scan it.)
+ assertScansWhileBuilding(conn, childViewName, parentViewKey, 1);
+ // The table is client-scanned in its own region, by the child view's
TTL recursion and,
+ // depending on how the client resolves the parent view, by building the
parent view.
+ ScanCountingObserver.WATCHED_KEYS.clear();
+ ScanCountingObserver.WATCHED_KEYS.add(tableKey);
+ PTable childView =
clearCachesAndResetCount(conn).getTableNoCache(childViewName);
+ assertTrue(ScanCountingObserver.SCANS.get() >= 1);
+ assertTTLOf(tableName, childView);
+ }
+ }
+
+ /**
+ * Rebuilds the given table or view and asserts the number of client scans
of SYSTEM.CATALOG
+ * starting at the given row key.
+ */
+ private static PTable assertScansWhileBuilding(Connection conn, String name,
byte[] key,
+ int expectedScans) throws Exception {
+ ScanCountingObserver.WATCHED_KEYS.clear();
+ ScanCountingObserver.WATCHED_KEYS.add(key);
+ PTable table = clearCachesAndResetCount(conn).getTableNoCache(name);
+ assertEquals(expectedScans, ScanCountingObserver.SCANS.get());
+ return table;
+ }
+
+ private static void createTableWithTTL(String tableName) throws Exception {
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ conn.createStatement()
+ .execute("CREATE TABLE " + tableName
+ + " (TENANT_ID VARCHAR NOT NULL, ID VARCHAR NOT NULL, V1 VARCHAR"
+ + " CONSTRAINT PK PRIMARY KEY (TENANT_ID, ID)) MULTI_TENANT=true,
TTL=86400");
+ TestUtil.addCoprocessor(conn,
PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME,
+ ScanCountingObserver.class);
+ }
+ }
+
+ private static void createTenantView(String viewName, String parentName)
throws Exception {
+ try (Connection conn = getTenantConnection()) {
+ conn.createStatement().execute("CREATE VIEW " + viewName + " AS SELECT *
FROM " + parentName);
+ }
+ }
+
+ private static Connection getTenantConnection() throws Exception {
+ Properties props = new Properties();
+ props.setProperty(PhoenixRuntime.TENANT_ID_ATTRIB, TENANT1);
+ return DriverManager.getConnection(getUrl(), props);
+ }
+
+ private static byte[] getTenantTableKey(String fullName) {
+ return SchemaUtil.getTableKey(TENANT1,
SchemaUtil.getSchemaNameFromFullName(fullName),
+ SchemaUtil.getTableNameFromFullName(fullName));
+ }
+
+ private static void assertInDifferentRegions(byte[]... keys) throws
Exception {
+ try (RegionLocator locator =
getUtility().getConnection().getRegionLocator(SYSTEM_CATALOG)) {
+ for (int i = 0; i < keys.length; i++) {
+ for (int j = i + 1; j < keys.length; j++) {
+ assertNotEquals("Expected the keys in different SYSTEM.CATALOG
regions",
+ locator.getRegionLocation(keys[i], true).getRegion(),
+ locator.getRegionLocation(keys[j], true).getRegion());
+ }
+ }
+ }
+ }
+
+ private static PhoenixConnection clearCachesAndResetCount(Connection conn)
throws Exception {
+ PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
+ // Clear the client and server metadata caches so that getTable rebuilds
the PTables.
+ pconn.getQueryServices().clearCache();
+ ScanCountingObserver.SCANS.set(0);
+ return pconn;
+ }
+
+ private static void assertTTLOf(String tableName, PTable view) throws
Exception {
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ PTable table =
conn.unwrap(PhoenixConnection.class).getTableNoCache(tableName);
+ assertNotEquals(TTL_EXPRESSION_NOT_DEFINED, table.getTTLExpression());
+ assertEquals(table.getTTLExpression(), view.getTTLExpression());
+ }
+ }
+}
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/util/ServerUtilTest.java
b/phoenix-core/src/test/java/org/apache/phoenix/util/ServerUtilTest.java
new file mode 100644
index 0000000000..ba8821f080
--- /dev/null
+++ b/phoenix-core/src/test/java/org/apache/phoenix/util/ServerUtilTest.java
@@ -0,0 +1,117 @@
+/*
+ * 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 static org.junit.Assert.assertEquals;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.query.QueryServicesOptions;
+import org.junit.Test;
+
+public class ServerUtilTest {
+
+ @Test
+ public void testMetadataConfigDefaults() {
+ Configuration conf = new Configuration(false);
+ conf.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 150);
+ conf.setInt(HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD, 60000);
+ conf.set("some.other.key", "value");
+
+ Configuration metadataConf = ServerUtil.getMetadataConfig(conf);
+
+ int timeoutMs =
QueryServicesOptions.DEFAULT_METADATA_SERVER_LOOKUP_TIMEOUT_MS;
+ assertEquals(timeoutMs,
+ metadataConf.getInt(HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD, -1));
+ assertEquals(timeoutMs,
metadataConf.getInt(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, -1));
+ assertEquals(timeoutMs,
metadataConf.getInt(HConstants.HBASE_RPC_READ_TIMEOUT_KEY, -1));
+ assertEquals(timeoutMs,
+ metadataConf.getInt(HConstants.HBASE_CLIENT_META_OPERATION_TIMEOUT, -1));
+ assertEquals(timeoutMs,
metadataConf.getInt("hbase.client.meta.read.rpc.timeout", -1));
+ assertEquals(timeoutMs,
metadataConf.getInt("hbase.client.meta.scanner.timeout.period", -1));
+ assertEquals("value", metadataConf.get("some.other.key"));
+ // backoffs of 0.1+0.2+0.3+0.5+1+2+4 = 8.1s fit in 10s, the next 10s
backoff does not
+ assertEquals(7,
metadataConf.getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, -1));
+ // the source configuration is not modified
+ assertEquals(150, conf.getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, -1));
+ assertEquals(60000,
conf.getInt(HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD, -1));
+ }
+
+ @Test
+ public void testMetadataConfigOverrides() {
+ Configuration conf = new Configuration(false);
+ conf.setInt(QueryServices.METADATA_SERVER_LOOKUP_TIMEOUT_MS, 2500);
+
+ Configuration metadataConf = ServerUtil.getMetadataConfig(conf);
+
+ assertEquals(2500,
metadataConf.getInt(HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD, -1));
+ assertEquals(2500,
metadataConf.getInt(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, -1));
+ assertEquals(2500,
metadataConf.getInt(HConstants.HBASE_RPC_READ_TIMEOUT_KEY, -1));
+ // backoffs of 0.1+0.2+0.3+0.5+1 = 2.1s fit in 2.5s
+ assertEquals(5,
metadataConf.getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, -1));
+ }
+
+ @Test
+ public void testMetadataConfigRetriesUseLargerPause() {
+ Configuration conf = new Configuration(false);
+ conf.setLong(HConstants.HBASE_CLIENT_PAUSE, 1000);
+ // backoffs of 1+2+3 = 6s fit in 10s
+ assertEquals(3,
+
ServerUtil.getMetadataConfig(conf).getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,
-1));
+
+ conf.setLong(HConstants.HBASE_CLIENT_PAUSE, 100);
+ conf.setLong("hbase.client.pause.server.overloaded", 1000);
+ assertEquals(3,
+
ServerUtil.getMetadataConfig(conf).getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,
-1));
+ }
+
+ @Test
+ public void testMetadataConfigNonPositiveTimeoutUsesDefault() {
+ Configuration conf = new Configuration(false);
+ conf.setInt(QueryServices.METADATA_SERVER_LOOKUP_TIMEOUT_MS, 0);
+
+ Configuration metadataConf = ServerUtil.getMetadataConfig(conf);
+
+
assertEquals(QueryServicesOptions.DEFAULT_METADATA_SERVER_LOOKUP_TIMEOUT_MS,
+ metadataConf.getInt(HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD, -1));
+ assertEquals(7,
metadataConf.getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, -1));
+ }
+
+ @Test
+ public void testRetriesWithinTimeout() {
+ assertEquals(0, ServerUtil.getRetriesWithinTimeout(50, 100));
+ assertEquals(1, ServerUtil.getRetriesWithinTimeout(100, 100));
+ assertEquals(7, ServerUtil.getRetriesWithinTimeout(10000, 100));
+ // a larger pause allows fewer retries in the same time
+ assertEquals(3, ServerUtil.getRetriesWithinTimeout(10000, 1000));
+ // a zero pause must not loop forever
+ assertEquals(13, ServerUtil.getRetriesWithinTimeout(1000, 0));
+ }
+
+ @Test
+ public void testMetadataLookupConnectionUsesMetadataConfig() {
+ Configuration conf = new Configuration(false);
+ conf.setInt(QueryServices.METADATA_SERVER_LOOKUP_TIMEOUT_MS, 2500);
+
+ Configuration connConf = ServerUtil.ConnectionFactory
+
.getTypeSpecificConfiguration(ServerUtil.ConnectionType.METADATA_LOOKUP_CONNECTION,
conf);
+
+ assertEquals(2500,
connConf.getInt(HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD, -1));
+ }
+}
