Revert "HBASE-23185 Fix high cpu usage because getTable()#put() gets config value every time (#731)"
Causes failure in TestMasterNoCluster and TestCatalogJanitor
This reverts commit ce65db36c4
.
This commit is contained in:
parent
41f671336c
commit
db2ce23a93
|
@ -323,60 +323,35 @@ class AsyncProcess {
|
|||
|
||||
this.id = COUNTER.incrementAndGet();
|
||||
|
||||
ConnectionConfiguration connConf =
|
||||
hc.getConfiguration() == conf
|
||||
? hc.getConnectionConfiguration()
|
||||
// Slow: parse conf in ConnectionConfiguration constructor
|
||||
: new ConnectionConfiguration(conf);
|
||||
if (connConf == null) {
|
||||
// Slow: parse conf in ConnectionConfiguration constructor
|
||||
connConf = new ConnectionConfiguration(conf);
|
||||
this.pause = conf.getLong(HConstants.HBASE_CLIENT_PAUSE,
|
||||
HConstants.DEFAULT_HBASE_CLIENT_PAUSE);
|
||||
long configuredPauseForCQTBE = conf.getLong(HConstants.HBASE_CLIENT_PAUSE_FOR_CQTBE, pause);
|
||||
if (configuredPauseForCQTBE < pause) {
|
||||
LOG.warn("The " + HConstants.HBASE_CLIENT_PAUSE_FOR_CQTBE + " setting: "
|
||||
+ configuredPauseForCQTBE + " is smaller than " + HConstants.HBASE_CLIENT_PAUSE
|
||||
+ ", will use " + pause + " instead.");
|
||||
this.pauseForCQTBE = pause;
|
||||
} else {
|
||||
this.pauseForCQTBE = configuredPauseForCQTBE;
|
||||
}
|
||||
|
||||
this.pause = connConf.getPause();
|
||||
this.pauseForCQTBE = connConf.getPauseForCQTBE();
|
||||
|
||||
this.numTries = connConf.getRetriesNumber();
|
||||
this.numTries = conf.getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,
|
||||
HConstants.DEFAULT_HBASE_CLIENT_RETRIES_NUMBER);
|
||||
this.rpcTimeout = rpcTimeout;
|
||||
this.operationTimeout = connConf.getOperationTimeout();
|
||||
this.operationTimeout = conf.getInt(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT,
|
||||
HConstants.DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT);
|
||||
this.primaryCallTimeoutMicroseconds = conf.getInt(PRIMARY_CALL_TIMEOUT_KEY, 10000);
|
||||
|
||||
// Parse config once and reuse config values of hc's AsyncProcess in AsyncProcess for put
|
||||
// Can be null when constructing hc's AsyncProcess or it's not reusable
|
||||
AsyncProcess globalAsyncProcess = hc.getConfiguration() == conf ? hc.getAsyncProcess() : null;
|
||||
|
||||
this.primaryCallTimeoutMicroseconds =
|
||||
globalAsyncProcess == null
|
||||
? conf.getInt(PRIMARY_CALL_TIMEOUT_KEY, 10000)
|
||||
: globalAsyncProcess.primaryCallTimeoutMicroseconds;
|
||||
|
||||
this.maxTotalConcurrentTasks =
|
||||
globalAsyncProcess == null
|
||||
? conf.getInt(HConstants.HBASE_CLIENT_MAX_TOTAL_TASKS,
|
||||
HConstants.DEFAULT_HBASE_CLIENT_MAX_TOTAL_TASKS)
|
||||
: globalAsyncProcess.maxTotalConcurrentTasks;
|
||||
this.maxConcurrentTasksPerServer =
|
||||
globalAsyncProcess == null
|
||||
? conf.getInt(HConstants.HBASE_CLIENT_MAX_PERSERVER_TASKS,
|
||||
HConstants.DEFAULT_HBASE_CLIENT_MAX_PERSERVER_TASKS)
|
||||
: globalAsyncProcess.maxConcurrentTasksPerServer;
|
||||
this.maxConcurrentTasksPerRegion =
|
||||
globalAsyncProcess == null
|
||||
? conf.getInt(HConstants.HBASE_CLIENT_MAX_PERREGION_TASKS,
|
||||
HConstants.DEFAULT_HBASE_CLIENT_MAX_PERREGION_TASKS)
|
||||
: globalAsyncProcess.maxConcurrentTasksPerRegion;
|
||||
this.maxHeapSizePerRequest =
|
||||
globalAsyncProcess == null
|
||||
? conf.getLong(HBASE_CLIENT_MAX_PERREQUEST_HEAPSIZE,
|
||||
DEFAULT_HBASE_CLIENT_MAX_PERREQUEST_HEAPSIZE)
|
||||
: globalAsyncProcess.maxHeapSizePerRequest;
|
||||
this.maxHeapSizeSubmit =
|
||||
globalAsyncProcess == null
|
||||
? conf.getLong(HBASE_CLIENT_MAX_SUBMIT_HEAPSIZE, DEFAULT_HBASE_CLIENT_MAX_SUBMIT_HEAPSIZE)
|
||||
: globalAsyncProcess.maxHeapSizeSubmit;
|
||||
this.maxTotalConcurrentTasks = conf.getInt(HConstants.HBASE_CLIENT_MAX_TOTAL_TASKS,
|
||||
HConstants.DEFAULT_HBASE_CLIENT_MAX_TOTAL_TASKS);
|
||||
this.maxConcurrentTasksPerServer = conf.getInt(HConstants.HBASE_CLIENT_MAX_PERSERVER_TASKS,
|
||||
HConstants.DEFAULT_HBASE_CLIENT_MAX_PERSERVER_TASKS);
|
||||
this.maxConcurrentTasksPerRegion = conf.getInt(HConstants.HBASE_CLIENT_MAX_PERREGION_TASKS,
|
||||
HConstants.DEFAULT_HBASE_CLIENT_MAX_PERREGION_TASKS);
|
||||
this.maxHeapSizePerRequest = conf.getLong(HBASE_CLIENT_MAX_PERREQUEST_HEAPSIZE,
|
||||
DEFAULT_HBASE_CLIENT_MAX_PERREQUEST_HEAPSIZE);
|
||||
this.maxHeapSizeSubmit = conf.getLong(HBASE_CLIENT_MAX_SUBMIT_HEAPSIZE, DEFAULT_HBASE_CLIENT_MAX_SUBMIT_HEAPSIZE);
|
||||
this.startLogErrorsCnt =
|
||||
globalAsyncProcess == null
|
||||
? conf.getInt(START_LOG_ERRORS_AFTER_COUNT_KEY, DEFAULT_START_LOG_ERRORS_AFTER_COUNT)
|
||||
: globalAsyncProcess.startLogErrorsCnt;
|
||||
conf.getInt(START_LOG_ERRORS_AFTER_COUNT_KEY, DEFAULT_START_LOG_ERRORS_AFTER_COUNT);
|
||||
|
||||
if (this.maxTotalConcurrentTasks <= 0) {
|
||||
throw new IllegalArgumentException("maxTotalConcurrentTasks=" + maxTotalConcurrentTasks);
|
||||
|
@ -412,16 +387,11 @@ class AsyncProcess {
|
|||
|
||||
this.rpcCallerFactory = rpcCaller;
|
||||
this.rpcFactory = rpcFactory;
|
||||
this.logBatchErrorDetails =
|
||||
globalAsyncProcess == null
|
||||
? conf.getBoolean(LOG_DETAILS_FOR_BATCH_ERROR, false)
|
||||
: globalAsyncProcess.logBatchErrorDetails;
|
||||
this.logBatchErrorDetails = conf.getBoolean(LOG_DETAILS_FOR_BATCH_ERROR, false);
|
||||
|
||||
this.thresholdToLogUndoneTaskDetails =
|
||||
globalAsyncProcess == null
|
||||
? conf.getInt(THRESHOLD_TO_LOG_UNDONE_TASK_DETAILS,
|
||||
DEFAULT_THRESHOLD_TO_LOG_UNDONE_TASK_DETAILS)
|
||||
: globalAsyncProcess.thresholdToLogUndoneTaskDetails;
|
||||
conf.getInt(THRESHOLD_TO_LOG_UNDONE_TASK_DETAILS,
|
||||
DEFAULT_THRESHOLD_TO_LOG_UNDONE_TASK_DETAILS);
|
||||
}
|
||||
|
||||
public void setRpcTimeout(int rpcTimeout) {
|
||||
|
|
|
@ -34,6 +34,7 @@ import java.util.concurrent.atomic.AtomicLong;
|
|||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import org.apache.hadoop.conf.Configuration;
|
||||
import org.apache.hadoop.hbase.HConstants; // Needed for write rpc timeout
|
||||
import org.apache.hadoop.hbase.TableName;
|
||||
import org.apache.hadoop.hbase.classification.InterfaceAudience;
|
||||
import org.apache.hadoop.hbase.classification.InterfaceStability;
|
||||
|
@ -111,32 +112,32 @@ public class BufferedMutatorImpl implements BufferedMutator {
|
|||
this.pool = params.getPool();
|
||||
this.listener = params.getListener();
|
||||
|
||||
ConnectionConfiguration connConf = conn.getConnectionConfiguration();
|
||||
if (connConf == null) {
|
||||
// Slow: parse conf in ConnectionConfiguration constructor
|
||||
connConf = new ConnectionConfiguration(conf);
|
||||
}
|
||||
ConnectionConfiguration tableConf = new ConnectionConfiguration(conf);
|
||||
this.writeBufferSize = params.getWriteBufferSize() != BufferedMutatorParams.UNSET ?
|
||||
params.getWriteBufferSize() : connConf.getWriteBufferSize();
|
||||
params.getWriteBufferSize() : tableConf.getWriteBufferSize();
|
||||
|
||||
// Set via the setter because it does value validation and starts/stops the TimerTask
|
||||
long newWriteBufferPeriodicFlushTimeoutMs =
|
||||
params.getWriteBufferPeriodicFlushTimeoutMs() != UNSET
|
||||
? params.getWriteBufferPeriodicFlushTimeoutMs()
|
||||
: connConf.getWriteBufferPeriodicFlushTimeoutMs();
|
||||
: tableConf.getWriteBufferPeriodicFlushTimeoutMs();
|
||||
long newWriteBufferPeriodicFlushTimerTickMs =
|
||||
params.getWriteBufferPeriodicFlushTimerTickMs() != UNSET
|
||||
? params.getWriteBufferPeriodicFlushTimerTickMs()
|
||||
: connConf.getWriteBufferPeriodicFlushTimerTickMs();
|
||||
: tableConf.getWriteBufferPeriodicFlushTimerTickMs();
|
||||
this.setWriteBufferPeriodicFlush(
|
||||
newWriteBufferPeriodicFlushTimeoutMs,
|
||||
newWriteBufferPeriodicFlushTimerTickMs);
|
||||
|
||||
this.maxKeyValueSize = params.getMaxKeyValueSize() != BufferedMutatorParams.UNSET ?
|
||||
params.getMaxKeyValueSize() : connConf.getMaxKeyValueSize();
|
||||
params.getMaxKeyValueSize() : tableConf.getMaxKeyValueSize();
|
||||
|
||||
this.writeRpcTimeout = connConf.getWriteRpcTimeout();
|
||||
this.operationTimeout = connConf.getOperationTimeout();
|
||||
this.writeRpcTimeout = conn.getConfiguration().getInt(HConstants.HBASE_RPC_WRITE_TIMEOUT_KEY,
|
||||
conn.getConfiguration().getInt(HConstants.HBASE_RPC_TIMEOUT_KEY,
|
||||
HConstants.DEFAULT_HBASE_RPC_TIMEOUT));
|
||||
this.operationTimeout = conn.getConfiguration().getInt(
|
||||
HConstants.HBASE_CLIENT_OPERATION_TIMEOUT,
|
||||
HConstants.DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT);
|
||||
// puts need to track errors globally due to how the APIs currently work.
|
||||
ap = new AsyncProcess(connection, conf, pool, rpcCallerFactory, true, rpcFactory, writeRpcTimeout);
|
||||
}
|
||||
|
|
|
@ -12,8 +12,6 @@
|
|||
package org.apache.hadoop.hbase.client;
|
||||
|
||||
import com.google.common.annotations.VisibleForTesting;
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import org.apache.hadoop.conf.Configuration;
|
||||
import org.apache.hadoop.hbase.HConstants;
|
||||
import org.apache.hadoop.hbase.classification.InterfaceAudience;
|
||||
|
@ -28,7 +26,6 @@ import org.apache.hadoop.hbase.classification.InterfaceAudience;
|
|||
*/
|
||||
@InterfaceAudience.Private
|
||||
public class ConnectionConfiguration {
|
||||
static final Log LOG = LogFactory.getLog(ConnectionConfiguration.class);
|
||||
|
||||
public static final String WRITE_BUFFER_SIZE_KEY = "hbase.client.write.buffer";
|
||||
public static final long WRITE_BUFFER_SIZE_DEFAULT = 2097152;
|
||||
|
@ -53,10 +50,6 @@ public class ConnectionConfiguration {
|
|||
private final int metaReplicaCallTimeoutMicroSecondScan;
|
||||
private final int retries;
|
||||
private final int maxKeyValueSize;
|
||||
private final int readRpcTimeout;
|
||||
private final int writeRpcTimeout;
|
||||
private final long pause;
|
||||
private final long pauseForCQTBE;
|
||||
|
||||
/**
|
||||
* Constructor
|
||||
|
@ -97,28 +90,9 @@ public class ConnectionConfiguration {
|
|||
HConstants.HBASE_CLIENT_META_REPLICA_SCAN_TIMEOUT_DEFAULT);
|
||||
|
||||
this.retries = conf.getInt(
|
||||
HConstants.HBASE_CLIENT_RETRIES_NUMBER, HConstants.DEFAULT_HBASE_CLIENT_RETRIES_NUMBER);
|
||||
HConstants.HBASE_CLIENT_RETRIES_NUMBER, HConstants.DEFAULT_HBASE_CLIENT_RETRIES_NUMBER);
|
||||
|
||||
this.maxKeyValueSize = conf.getInt(MAX_KEYVALUE_SIZE_KEY, MAX_KEYVALUE_SIZE_DEFAULT);
|
||||
|
||||
this.readRpcTimeout = conf.getInt(HConstants.HBASE_RPC_READ_TIMEOUT_KEY,
|
||||
conf.getInt(HConstants.HBASE_RPC_TIMEOUT_KEY,
|
||||
HConstants.DEFAULT_HBASE_RPC_TIMEOUT));
|
||||
|
||||
this.writeRpcTimeout = conf.getInt(HConstants.HBASE_RPC_WRITE_TIMEOUT_KEY,
|
||||
conf.getInt(HConstants.HBASE_RPC_TIMEOUT_KEY,
|
||||
HConstants.DEFAULT_HBASE_RPC_TIMEOUT));
|
||||
|
||||
this.pause = conf.getLong(HConstants.HBASE_CLIENT_PAUSE, HConstants.DEFAULT_HBASE_CLIENT_PAUSE);
|
||||
long configuredPauseForCQTBE = conf.getLong(HConstants.HBASE_CLIENT_PAUSE_FOR_CQTBE, pause);
|
||||
if (configuredPauseForCQTBE < pause) {
|
||||
LOG.warn("The " + HConstants.HBASE_CLIENT_PAUSE_FOR_CQTBE + " setting: "
|
||||
+ configuredPauseForCQTBE + " is smaller than " + HConstants.HBASE_CLIENT_PAUSE
|
||||
+ ", will use " + pause + " instead.");
|
||||
this.pauseForCQTBE = pause;
|
||||
} else {
|
||||
this.pauseForCQTBE = configuredPauseForCQTBE;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
@ -141,10 +115,6 @@ public class ConnectionConfiguration {
|
|||
HConstants.HBASE_CLIENT_META_REPLICA_SCAN_TIMEOUT_DEFAULT;
|
||||
this.retries = HConstants.DEFAULT_HBASE_CLIENT_RETRIES_NUMBER;
|
||||
this.maxKeyValueSize = MAX_KEYVALUE_SIZE_DEFAULT;
|
||||
this.readRpcTimeout = HConstants.DEFAULT_HBASE_RPC_TIMEOUT;
|
||||
this.writeRpcTimeout = HConstants.DEFAULT_HBASE_RPC_TIMEOUT;
|
||||
this.pause = HConstants.DEFAULT_HBASE_CLIENT_PAUSE;
|
||||
this.pauseForCQTBE = HConstants.DEFAULT_HBASE_CLIENT_PAUSE;
|
||||
}
|
||||
|
||||
public long getWriteBufferSize() {
|
||||
|
@ -194,20 +164,4 @@ public class ConnectionConfiguration {
|
|||
public long getScannerMaxResultSize() {
|
||||
return scannerMaxResultSize;
|
||||
}
|
||||
|
||||
public int getReadRpcTimeout() {
|
||||
return readRpcTimeout;
|
||||
}
|
||||
|
||||
public int getWriteRpcTimeout() {
|
||||
return writeRpcTimeout;
|
||||
}
|
||||
|
||||
public long getPause() {
|
||||
return pause;
|
||||
}
|
||||
|
||||
public long getPauseForCQTBE() {
|
||||
return pauseForCQTBE;
|
||||
}
|
||||
}
|
||||
|
|
|
@ -669,8 +669,17 @@ class ConnectionManager {
|
|||
this.managed = managed;
|
||||
this.connectionConfig = new ConnectionConfiguration(conf);
|
||||
this.closed = false;
|
||||
this.pause = connectionConfig.getPause();
|
||||
this.pauseForCQTBE = connectionConfig.getPauseForCQTBE();
|
||||
this.pause = conf.getLong(HConstants.HBASE_CLIENT_PAUSE,
|
||||
HConstants.DEFAULT_HBASE_CLIENT_PAUSE);
|
||||
long configuredPauseForCQTBE = conf.getLong(HConstants.HBASE_CLIENT_PAUSE_FOR_CQTBE, pause);
|
||||
if (configuredPauseForCQTBE < pause) {
|
||||
LOG.warn("The " + HConstants.HBASE_CLIENT_PAUSE_FOR_CQTBE + " setting: "
|
||||
+ configuredPauseForCQTBE + " is smaller than " + HConstants.HBASE_CLIENT_PAUSE
|
||||
+ ", will use " + pause + " instead.");
|
||||
this.pauseForCQTBE = pause;
|
||||
} else {
|
||||
this.pauseForCQTBE = configuredPauseForCQTBE;
|
||||
}
|
||||
this.useMetaReplicas = conf.getBoolean(HConstants.USE_META_REPLICAS,
|
||||
HConstants.DEFAULT_USE_META_REPLICAS);
|
||||
this.metaReplicaCallTimeoutScanInMicroSecond =
|
||||
|
|
|
@ -366,8 +366,12 @@ public class HTable implements HTableInterface, RegionLocator {
|
|||
}
|
||||
this.operationTimeout = tableName.isSystemTable() ?
|
||||
connConfiguration.getMetaOperationTimeout() : connConfiguration.getOperationTimeout();
|
||||
this.readRpcTimeout = connConfiguration.getReadRpcTimeout();
|
||||
this.writeRpcTimeout = connConfiguration.getWriteRpcTimeout();
|
||||
this.readRpcTimeout = configuration.getInt(HConstants.HBASE_RPC_READ_TIMEOUT_KEY,
|
||||
configuration.getInt(HConstants.HBASE_RPC_TIMEOUT_KEY,
|
||||
HConstants.DEFAULT_HBASE_RPC_TIMEOUT));
|
||||
this.writeRpcTimeout = configuration.getInt(HConstants.HBASE_RPC_WRITE_TIMEOUT_KEY,
|
||||
configuration.getInt(HConstants.HBASE_RPC_TIMEOUT_KEY,
|
||||
HConstants.DEFAULT_HBASE_RPC_TIMEOUT));
|
||||
this.scannerCaching = connConfiguration.getScannerCaching();
|
||||
this.scannerMaxResultSize = connConfiguration.getScannerMaxResultSize();
|
||||
if (this.rpcCallerFactory == null) {
|
||||
|
|
Loading…
Reference in New Issue