Repository: phoenix Updated Branches: refs/heads/5.x-HBase-2.0 53b35099c -> f640e2dd5
PHOENIX-4601 Perform server-side retries if client version < 4.14 Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/f640e2dd Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/f640e2dd Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/f640e2dd Branch: refs/heads/5.x-HBase-2.0 Commit: f640e2dd517bf4247d32c7dec77ce303db3196ad Parents: 53b3509 Author: Vincent Poon <[email protected]> Authored: Fri Apr 20 14:22:10 2018 -0700 Committer: Vincent Poon <[email protected]> Committed: Fri Apr 20 14:44:49 2018 -0700 ---------------------------------------------------------------------- .../hbase/index/write/IndexWriterUtils.java | 39 ++++++++++++++++---- .../write/ParallelWriterIndexCommitter.java | 12 ++++-- .../TrackingParallelWriterIndexCommitter.java | 13 +++++-- .../phoenix/jdbc/PhoenixDatabaseMetaData.java | 1 + 4 files changed, 50 insertions(+), 15 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/phoenix/blob/f640e2dd/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/IndexWriterUtils.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/IndexWriterUtils.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/IndexWriterUtils.java index d8116b9..958fa3b 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/IndexWriterUtils.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/IndexWriterUtils.java @@ -29,9 +29,11 @@ import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.client.Connection; import org.apache.hadoop.hbase.client.Table; import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment; +import org.apache.hadoop.hbase.HConstants; import org.apache.phoenix.hbase.index.table.HTableFactory; import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr; import org.apache.phoenix.hbase.index.util.IndexManagementUtil; +import org.apache.phoenix.util.PropertiesUtil; public class IndexWriterUtils { @@ -65,15 +67,19 @@ public class IndexWriterUtils { public static final String INDEX_WRITES_THREAD_MAX_PER_REGIONSERVER_KEY = "phoenix.index.writes.threads.max"; public static final String HTABLE_KEEP_ALIVE_KEY = "hbase.htable.threads.keepalivetime"; + @Deprecated public static final String INDEX_WRITER_RPC_RETRIES_NUMBER = "phoenix.index.writes.rpc.retries.number"; - /** - * Retry server-server index write rpc only once, and let the client retry the data write - * instead to avoid typing up the handler - */ - // note in HBase 2+, numTries = numRetries + 1 - // in prior versions, numTries = numRetries - public static final int DEFAULT_INDEX_WRITER_RPC_RETRIES_NUMBER = 1; + /** + * Based on the logic in HBase's AsyncProcess, a default of 11 retries with a pause of 100ms + * approximates 48 sec total retry time (factoring in backoffs). The total time should be less + * than HBase's rpc timeout (default of 60 sec) or else the client will retry before receiving + * the response + */ + @Deprecated + public static final int DEFAULT_INDEX_WRITER_RPC_RETRIES_NUMBER = 11; + @Deprecated public static final String INDEX_WRITER_RPC_PAUSE = "phoenix.index.writes.rpc.pause"; + @Deprecated public static final int DEFAULT_INDEX_WRITER_RPC_PAUSE = 100; private IndexWriterUtils() { @@ -82,13 +88,30 @@ public class IndexWriterUtils { public static HTableFactory getDefaultDelegateHTableFactory(RegionCoprocessorEnvironment env) { // create a simple delegate factory, setup the way we need - Configuration conf = env.getConfiguration(); + Configuration conf = PropertiesUtil.cloneConfig(env.getConfiguration()); + setHTableThreads(conf); + return new CoprocessorHConnectionTableFactory(conf, env); + } + + private static void setHTableThreads(Configuration conf) { // set the number of threads allowed per table. int htableThreads = conf.getInt(IndexWriterUtils.INDEX_WRITER_PER_TABLE_THREADS_CONF_KEY, IndexWriterUtils.DEFAULT_NUM_PER_TABLE_THREADS); LOG.trace("Creating HTableFactory with " + htableThreads + " threads for each HTable."); IndexManagementUtil.setIfNotSet(conf, HTABLE_THREAD_KEY, htableThreads); + } + + /** + * Retry server-server index write rpc only once, and let the client retry the data write + * instead to avoid tying up the handler + */ + public static HTableFactory getNoRetriesHTableFactory(RegionCoprocessorEnvironment env) { + Configuration conf = PropertiesUtil.cloneConfig(env.getConfiguration()); + setHTableThreads(conf); + // note in HBase 2+, numTries = numRetries + 1 + // in prior versions, numTries = numRetries + conf.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 1); return new CoprocessorHConnectionTableFactory(conf, env); } http://git-wip-us.apache.org/repos/asf/phoenix/blob/f640e2dd/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/ParallelWriterIndexCommitter.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/ParallelWriterIndexCommitter.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/ParallelWriterIndexCommitter.java index c9d96f3..e0e49f9 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/ParallelWriterIndexCommitter.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/ParallelWriterIndexCommitter.java @@ -35,6 +35,7 @@ import org.apache.phoenix.hbase.index.table.HTableFactory; import org.apache.phoenix.hbase.index.table.HTableInterfaceReference; import org.apache.phoenix.hbase.index.util.KeyValueBuilder; import org.apache.phoenix.index.PhoenixIndexFailurePolicy; +import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData; import org.apache.phoenix.util.IndexUtil; import com.google.common.collect.Multimap; @@ -56,7 +57,8 @@ public class ParallelWriterIndexCommitter implements IndexCommitter { public static final String INDEX_WRITER_KEEP_ALIVE_TIME_CONF_KEY = "index.writer.threads.keepalivetime"; private static final Log LOG = LogFactory.getLog(ParallelWriterIndexCommitter.class); - private HTableFactory factory; + private HTableFactory retryingFactory; + private HTableFactory noRetriesfactory; private Stoppable stopped; private QuickFailingTaskRunner pool; private KeyValueBuilder kvBuilder; @@ -87,7 +89,8 @@ public class ParallelWriterIndexCommitter implements IndexCommitter { * Exposed for TESTING */ void setup(HTableFactory factory, ExecutorService pool,Stoppable stop, RegionCoprocessorEnvironment env) { - this.factory = factory; + this.retryingFactory = factory; + this.noRetriesfactory = IndexWriterUtils.getNoRetriesHTableFactory(env); this.pool = new QuickFailingTaskRunner(pool); this.stopped = stop; } @@ -161,6 +164,8 @@ public class ParallelWriterIndexCommitter implements IndexCommitter { } } } + // if the client can retry index writes, then we don't need to retry here + HTableFactory factory = clientVersion < PhoenixDatabaseMetaData.MIN_CLIENT_RETRY_INDEX_WRITES ? retryingFactory : noRetriesfactory; table = factory.getTable(tableReference.get()); throwFailureIfDone(); table.batch(mutations, null); @@ -225,7 +230,8 @@ public class ParallelWriterIndexCommitter implements IndexCommitter { public void stop(String why) { LOG.info("Shutting down " + this.getClass().getSimpleName() + " because " + why); this.pool.stop(why); - this.factory.shutdown(); + this.retryingFactory.shutdown(); + this.noRetriesfactory.shutdown(); } @Override http://git-wip-us.apache.org/repos/asf/phoenix/blob/f640e2dd/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/TrackingParallelWriterIndexCommitter.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/TrackingParallelWriterIndexCommitter.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/TrackingParallelWriterIndexCommitter.java index a0ab78c..a441606 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/TrackingParallelWriterIndexCommitter.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/TrackingParallelWriterIndexCommitter.java @@ -40,6 +40,7 @@ import org.apache.phoenix.hbase.index.table.HTableFactory; import org.apache.phoenix.hbase.index.table.HTableInterfaceReference; import org.apache.phoenix.hbase.index.util.KeyValueBuilder; import org.apache.phoenix.index.PhoenixIndexFailurePolicy; +import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData; import org.apache.phoenix.util.IndexUtil; import com.google.common.collect.Multimap; @@ -73,7 +74,8 @@ public class TrackingParallelWriterIndexCommitter implements IndexCommitter { private static final String INDEX_WRITER_KEEP_ALIVE_TIME_CONF_KEY = "index.writer.threads.keepalivetime"; private TaskRunner pool; - private HTableFactory factory; + private HTableFactory retryingFactory; + private HTableFactory noRetriesFactory; private Stoppable stopped; private RegionCoprocessorEnvironment env; private KeyValueBuilder kvBuilder; @@ -106,7 +108,8 @@ public class TrackingParallelWriterIndexCommitter implements IndexCommitter { void setup(HTableFactory factory, ExecutorService pool, Stoppable stop, RegionCoprocessorEnvironment env) { this.pool = new WaitForCompletionTaskRunner(pool); - this.factory = factory; + this.retryingFactory = factory; + this.noRetriesFactory = IndexWriterUtils.getNoRetriesHTableFactory(env); this.stopped = stop; this.env = env; } @@ -172,7 +175,8 @@ public class TrackingParallelWriterIndexCommitter implements IndexCommitter { if (LOG.isTraceEnabled()) { LOG.trace("Writing index update:" + mutations + " to table: " + tableReference); } - + // if the client can retry index writes, then we don't need to retry here + HTableFactory factory = clientVersion < PhoenixDatabaseMetaData.MIN_CLIENT_RETRY_INDEX_WRITES ? retryingFactory : noRetriesFactory; table = factory.getTable(tableReference.get()); throwFailureIfDone(); table.batch(mutations, null); @@ -238,7 +242,8 @@ public class TrackingParallelWriterIndexCommitter implements IndexCommitter { public void stop(String why) { LOG.info("Shutting down " + this.getClass().getSimpleName()); this.pool.stop(why); - this.factory.shutdown(); + this.retryingFactory.shutdown(); + this.noRetriesFactory.shutdown(); } @Override http://git-wip-us.apache.org/repos/asf/phoenix/blob/f640e2dd/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java index 8ef3270..409c0bc 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java @@ -327,6 +327,7 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData { public static final int MIN_RENEW_LEASE_VERSION = VersionUtil.encodeVersion("1", "1", "3"); public static final int MIN_NAMESPACE_MAPPED_PHOENIX_VERSION = VersionUtil.encodeVersion("4", "8", "0"); public static final int MIN_PENDING_ACTIVE_INDEX = VersionUtil.encodeVersion("4", "12", "0"); + public static final int MIN_CLIENT_RETRY_INDEX_WRITES = VersionUtil.encodeVersion("4", "14", "0"); public static final int MIN_TX_CLIENT_SIDE_MAINTENANCE = VersionUtil.encodeVersion("4", "14", "0"); public static final int MIN_PENDING_DISABLE_INDEX = VersionUtil.encodeVersion("4", "14", "0");
