You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by ja...@apache.org on 2017/08/31 00:32:26 UTC
[1/3] phoenix git commit: PHOENIX-3815 Only disable indexes on which
write failures occurred (Vincent Poon)
Repository: phoenix
Updated Branches:
refs/heads/4.x-HBase-1.1 6ac72c4c7 -> 55ff60189
PHOENIX-3815 Only disable indexes on which write failures occurred (Vincent Poon)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/55ff6018
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/55ff6018
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/55ff6018
Branch: refs/heads/4.x-HBase-1.1
Commit: 55ff601892d6f54b3542a43efffd7537bcc7a2de
Parents: cab3a39
Author: James Taylor <jt...@salesforce.com>
Authored: Wed Aug 30 16:57:52 2017 -0700
Committer: James Taylor <jt...@salesforce.com>
Committed: Wed Aug 30 17:32:24 2017 -0700
----------------------------------------------------------------------
.../end2end/index/MutableIndexFailureIT.java | 25 +-
.../phoenix/hbase/index/write/IndexWriter.java | 4 +-
.../write/ParallelWriterIndexCommitter.java | 234 ------------------
.../hbase/index/write/RecoveryIndexWriter.java | 1 -
.../TrackingParallelWriterIndexCommitter.java | 243 +++++++++++++++++++
.../recovery/StoreFailuresInCachePolicy.java | 1 +
.../TrackingParallelWriterIndexCommitter.java | 236 ------------------
.../index/PhoenixIndexFailurePolicy.java | 15 ++
.../hbase/index/write/TestIndexWriter.java | 6 +-
.../index/write/TestParalleIndexWriter.java | 4 +-
.../write/TestParalleWriterIndexCommitter.java | 4 +-
11 files changed, 286 insertions(+), 487 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/55ff6018/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/MutableIndexFailureIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/MutableIndexFailureIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/MutableIndexFailureIT.java
index 0abd5ae..82972dc 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/MutableIndexFailureIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/MutableIndexFailureIT.java
@@ -46,7 +46,9 @@ import org.apache.hadoop.hbase.regionserver.MiniBatchOperationInProgress;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.phoenix.end2end.NeedsOwnMiniClusterTest;
import org.apache.phoenix.execute.CommitException;
+import org.apache.phoenix.hbase.index.write.IndexWriter;
import org.apache.phoenix.hbase.index.write.IndexWriterUtils;
+import org.apache.phoenix.hbase.index.write.TrackingParallelWriterIndexCommitter;
import org.apache.phoenix.index.PhoenixIndexFailurePolicy;
import org.apache.phoenix.query.BaseTest;
import org.apache.phoenix.query.QueryConstants;
@@ -173,7 +175,7 @@ public class MutableIndexFailureIT extends BaseTest {
@Test
public void testIndexWriteFailure() throws Exception {
String secondIndexName = "B_" + FailingRegionObserver.FAIL_INDEX_NAME;
-// String thirdIndexName = "C_" + INDEX_NAME;
+ String thirdIndexName = "C_" + "IDX";
// String thirdFullIndexName = SchemaUtil.getTableName(schema, thirdIndexName);
Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
props.put(QueryServices.IS_NAMESPACE_MAPPING_ENABLED, String.valueOf(isNamespaceMapped));
@@ -197,8 +199,8 @@ public class MutableIndexFailureIT extends BaseTest {
// check the drop index.
conn.createStatement().execute(
"CREATE " + (!localIndex ? "LOCAL " : "") + " INDEX " + secondIndexName + " ON " + fullTableName + " (v2) INCLUDE (v1)");
-// conn.createStatement().execute(
-// "CREATE " + (localIndex ? "LOCAL " : "") + " INDEX " + thirdIndexName + " ON " + fullTableName + " (v1) INCLUDE (v2)");
+ conn.createStatement().execute(
+ "CREATE " + (localIndex ? "LOCAL " : "") + " INDEX " + thirdIndexName + " ON " + fullTableName + " (v1) INCLUDE (v2)");
query = "SELECT * FROM " + fullIndexName;
rs = conn.createStatement().executeQuery(query);
@@ -248,6 +250,10 @@ public class MutableIndexFailureIT extends BaseTest {
} else {
String indexState = rs.getString("INDEX_STATE");
assertTrue(PIndexState.DISABLE.toString().equals(indexState) || PIndexState.INACTIVE.toString().equals(indexState));
+ // non-failing index should remain active
+ ResultSet thirdRs = conn.createStatement().executeQuery(getSysCatQuery(thirdIndexName));
+ assertTrue(thirdRs.next());
+ assertEquals(PIndexState.ACTIVE.getSerializedValue(), thirdRs.getString(1));
}
assertFalse(rs.next());
@@ -308,10 +314,7 @@ public class MutableIndexFailureIT extends BaseTest {
waitForIndexRebuild(conn, fullIndexName, PIndexState.DISABLE);
// verify that the index was marked as disabled and the index disable
// timestamp set to 0
- String q =
- "SELECT INDEX_STATE, INDEX_DISABLE_TIMESTAMP FROM SYSTEM.CATALOG WHERE TABLE_SCHEM = '"
- + schema + "' AND TABLE_NAME = '" + indexName + "'"
- + " AND COLUMN_NAME IS NULL AND COLUMN_FAMILY IS NULL";
+ String q = getSysCatQuery(indexName);
try (ResultSet r = conn.createStatement().executeQuery(q)) {
assertTrue(r.next());
assertEquals(PIndexState.DISABLE.getSerializedValue(), r.getString(1));
@@ -325,6 +328,14 @@ public class MutableIndexFailureIT extends BaseTest {
}
}
+ private String getSysCatQuery(String iName) {
+ String q =
+ "SELECT INDEX_STATE, INDEX_DISABLE_TIMESTAMP FROM SYSTEM.CATALOG WHERE TABLE_SCHEM = '"
+ + schema + "' AND TABLE_NAME = '" + iName + "'"
+ + " AND COLUMN_NAME IS NULL AND COLUMN_FAMILY IS NULL";
+ return q;
+ }
+
private void waitForIndexRebuild(Connection conn, String fullIndexName, PIndexState expectedIndexState) throws InterruptedException, SQLException {
if (!transactional) {
TestUtil.waitForIndexRebuild(conn, fullIndexName, expectedIndexState);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/55ff6018/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/IndexWriter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/IndexWriter.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/IndexWriter.java
index a037e92..d8f5527 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/IndexWriter.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/IndexWriter.java
@@ -47,7 +47,7 @@ import com.google.common.collect.Multimap;
public class IndexWriter implements Stoppable {
private static final Log LOG = LogFactory.getLog(IndexWriter.class);
- private static final String INDEX_COMMITTER_CONF_KEY = "index.writer.commiter.class";
+ public static final String INDEX_COMMITTER_CONF_KEY = "index.writer.commiter.class";
public static final String INDEX_FAILURE_POLICY_CONF_KEY = "index.writer.failurepolicy.class";
private AtomicBoolean stopped = new AtomicBoolean(false);
private IndexCommitter writer;
@@ -69,7 +69,7 @@ public class IndexWriter implements Stoppable {
Configuration conf = env.getConfiguration();
try {
IndexCommitter committer =
- conf.getClass(INDEX_COMMITTER_CONF_KEY, ParallelWriterIndexCommitter.class,
+ conf.getClass(INDEX_COMMITTER_CONF_KEY, TrackingParallelWriterIndexCommitter.class,
IndexCommitter.class).newInstance();
return committer;
} catch (InstantiationException e) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/55ff6018/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
deleted file mode 100644
index cb2ffe8..0000000
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/ParallelWriterIndexCommitter.java
+++ /dev/null
@@ -1,234 +0,0 @@
-/*
- * 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.hbase.index.write;
-
-import java.io.IOException;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map.Entry;
-import java.util.Set;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.Abortable;
-import org.apache.hadoop.hbase.Stoppable;
-import org.apache.hadoop.hbase.client.HTableInterface;
-import org.apache.hadoop.hbase.client.Mutation;
-import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
-import org.apache.phoenix.hbase.index.exception.SingleIndexWriteFailureException;
-import org.apache.phoenix.hbase.index.parallel.EarlyExitFailure;
-import org.apache.phoenix.hbase.index.parallel.QuickFailingTaskRunner;
-import org.apache.phoenix.hbase.index.parallel.Task;
-import org.apache.phoenix.hbase.index.parallel.TaskBatch;
-import org.apache.phoenix.hbase.index.parallel.ThreadPoolBuilder;
-import org.apache.phoenix.hbase.index.parallel.ThreadPoolManager;
-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.util.IndexUtil;
-
-import com.google.common.collect.Multimap;
-
-/**
- * Write index updates to the index tables in parallel. We attempt to early exit from the writes if any of the index
- * updates fails. Completion is determined by the following criteria: *
- * <ol>
- * <li>All index writes have returned, OR</li>
- * <li>Any single index write has failed</li>
- * </ol>
- * We attempt to quickly determine if any write has failed and not write to the remaining indexes to ensure a timely
- * recovery of the failed index writes.
- */
-public class ParallelWriterIndexCommitter implements IndexCommitter {
-
- public static final String NUM_CONCURRENT_INDEX_WRITER_THREADS_CONF_KEY = "index.writer.threads.max";
- private static final int DEFAULT_CONCURRENT_INDEX_WRITER_THREADS = 10;
- 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 Stoppable stopped;
- private QuickFailingTaskRunner pool;
- private KeyValueBuilder kvBuilder;
- private RegionCoprocessorEnvironment env;
-
- public ParallelWriterIndexCommitter() {}
-
- // For testing
- public ParallelWriterIndexCommitter(String hbaseVersion) {
- kvBuilder = KeyValueBuilder.get(hbaseVersion);
- }
-
- @Override
- public void setup(IndexWriter parent, RegionCoprocessorEnvironment env, String name) {
- this.env = env;
- Configuration conf = env.getConfiguration();
- setup(IndexWriterUtils.getDefaultDelegateHTableFactory(env),
- ThreadPoolManager.getExecutor(
- new ThreadPoolBuilder(name, conf).setMaxThread(NUM_CONCURRENT_INDEX_WRITER_THREADS_CONF_KEY,
- DEFAULT_CONCURRENT_INDEX_WRITER_THREADS).setCoreTimeout(
- INDEX_WRITER_KEEP_ALIVE_TIME_CONF_KEY), env), env.getRegionServerServices(), parent, env);
- this.kvBuilder = KeyValueBuilder.get(env.getHBaseVersion());
- }
-
- /**
- * Setup <tt>this</tt>.
- * <p>
- * Exposed for TESTING
- */
- void setup(HTableFactory factory, ExecutorService pool, Abortable abortable, Stoppable stop, RegionCoprocessorEnvironment env) {
- this.factory = factory;
- this.pool = new QuickFailingTaskRunner(pool);
- this.stopped = stop;
- }
-
- @Override
- public void write(Multimap<HTableInterfaceReference, Mutation> toWrite, final boolean allowLocalUpdates) throws SingleIndexWriteFailureException {
- /*
- * This bit here is a little odd, so let's explain what's going on. Basically, we want to do the writes in
- * parallel to each index table, so each table gets its own task and is submitted to the pool. Where it gets
- * tricky is that we want to block the calling thread until one of two things happens: (1) all index tables get
- * successfully updated, or (2) any one of the index table writes fail; in either case, we should return as
- * quickly as possible. We get a little more complicated in that if we do get a single failure, but any of the
- * index writes hasn't been started yet (its been queued up, but not submitted to a thread) we want to that task
- * to fail immediately as we know that write is a waste and will need to be replayed anyways.
- */
-
- Set<Entry<HTableInterfaceReference, Collection<Mutation>>> entries = toWrite.asMap().entrySet();
- TaskBatch<Void> tasks = new TaskBatch<Void>(entries.size());
- for (Entry<HTableInterfaceReference, Collection<Mutation>> entry : entries) {
- // get the mutations for each table. We leak the implementation here a little bit to save
- // doing a complete copy over of all the index update for each table.
- final List<Mutation> mutations = kvBuilder.cloneIfNecessary((List<Mutation>)entry.getValue());
- final HTableInterfaceReference tableReference = entry.getKey();
- if (env != null
- && !allowLocalUpdates
- && tableReference.getTableName().equals(
- env.getRegion().getTableDesc().getNameAsString())) {
- continue;
- }
- /*
- * Write a batch of index updates to an index table. This operation stops (is cancelable) via two
- * mechanisms: (1) setting aborted or stopped on the IndexWriter or, (2) interrupting the running thread.
- * The former will only work if we are not in the midst of writing the current batch to the table, though we
- * do check these status variables before starting and before writing the batch. The latter usage,
- * interrupting the thread, will work in the previous situations as was at some points while writing the
- * batch, depending on the underlying writer implementation (HTableInterface#batch is blocking, but doesn't
- * elaborate when is supports an interrupt).
- */
- tasks.add(new Task<Void>() {
-
- /**
- * Do the actual write to the primary table.
- *
- * @return
- */
- @SuppressWarnings("deprecation")
- @Override
- public Void call() throws Exception {
- // this may have been queued, so another task infront of us may have failed, so we should
- // early exit, if that's the case
- throwFailureIfDone();
-
- if (LOG.isTraceEnabled()) {
- LOG.trace("Writing index update:" + mutations + " to table: " + tableReference);
- }
- HTableInterface table = null;
- try {
- if (allowLocalUpdates && env != null && tableReference.getTableName()
- .equals(env.getRegion().getTableDesc().getNameAsString())) {
- try {
- throwFailureIfDone();
- IndexUtil.writeLocalUpdates(env.getRegion(), mutations, true);
- return null;
- } catch (IOException ignord) {
- // when it's failed we fall back to the standard & slow way
- if (LOG.isDebugEnabled()) {
- LOG.debug(
- "indexRegion.batchMutate failed and fall back to HTable.batch(). Got error="
- + ignord);
- }
- }
- }
- table = factory.getTable(tableReference.get());
- throwFailureIfDone();
- table.batch(mutations);
- } catch (SingleIndexWriteFailureException e) {
- throw e;
- } catch (IOException e) {
- throw new SingleIndexWriteFailureException(tableReference.toString(), mutations, e);
- } catch (InterruptedException e) {
- // reset the interrupt status on the thread
- Thread.currentThread().interrupt();
- throw new SingleIndexWriteFailureException(tableReference.toString(), mutations, e);
- }
- finally{
- if (table != null) {
- table.close();
- }
- }
- return null;
- }
-
- private void throwFailureIfDone() throws SingleIndexWriteFailureException {
- if (this.isBatchFailed() || Thread.currentThread().isInterrupted()) { throw new SingleIndexWriteFailureException(
- "Pool closed, not attempting to write to the index!", null); }
-
- }
- });
- }
-
- // actually submit the tasks to the pool and wait for them to finish/fail
- try {
- pool.submitUninterruptible(tasks);
- } catch (EarlyExitFailure e) {
- propagateFailure(e);
- } catch (ExecutionException e) {
- LOG.error("Found a failed index update!");
- propagateFailure(e.getCause());
- }
-
- }
-
- private void propagateFailure(Throwable throwable) throws SingleIndexWriteFailureException {
- try {
- throw throwable;
- } catch (SingleIndexWriteFailureException e1) {
- throw e1;
- } catch (Throwable e1) {
- throw new SingleIndexWriteFailureException("Got an abort notification while writing to the index!", e1);
- }
-
- }
-
- /**
- * {@inheritDoc}
- * <p>
- * This method should only be called <b>once</b>. Stopped state ({@link #isStopped()}) is managed by the external
- * {@link Stoppable}. This call does not delegate the stop down to the {@link Stoppable} passed in the constructor.
- *
- * @param why
- * the reason for stopping
- */
- @Override
- public void stop(String why) {
- LOG.info("Shutting down " + this.getClass().getSimpleName() + " because " + why);
- this.pool.stop(why);
- this.factory.shutdown();
- }
-
- @Override
- public boolean isStopped() {
- return this.stopped.isStopped();
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/55ff6018/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/RecoveryIndexWriter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/RecoveryIndexWriter.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/RecoveryIndexWriter.java
index be542bb..e340784 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/RecoveryIndexWriter.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/RecoveryIndexWriter.java
@@ -35,7 +35,6 @@ import org.apache.hadoop.hbase.util.Pair;
import org.apache.phoenix.hbase.index.exception.MultiIndexWriteFailureException;
import org.apache.phoenix.hbase.index.table.HTableInterfaceReference;
import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
-import org.apache.phoenix.hbase.index.write.recovery.TrackingParallelWriterIndexCommitter;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Multimap;
http://git-wip-us.apache.org/repos/asf/phoenix/blob/55ff6018/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
new file mode 100644
index 0000000..0052f8c
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/TrackingParallelWriterIndexCommitter.java
@@ -0,0 +1,243 @@
+/*
+ * 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.hbase.index.write;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.Abortable;
+import org.apache.hadoop.hbase.Stoppable;
+import org.apache.hadoop.hbase.client.HTableInterface;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.phoenix.hbase.index.CapturingAbortable;
+import org.apache.phoenix.hbase.index.exception.MultiIndexWriteFailureException;
+import org.apache.phoenix.hbase.index.exception.SingleIndexWriteFailureException;
+import org.apache.phoenix.hbase.index.parallel.EarlyExitFailure;
+import org.apache.phoenix.hbase.index.parallel.Task;
+import org.apache.phoenix.hbase.index.parallel.TaskBatch;
+import org.apache.phoenix.hbase.index.parallel.TaskRunner;
+import org.apache.phoenix.hbase.index.parallel.ThreadPoolBuilder;
+import org.apache.phoenix.hbase.index.parallel.ThreadPoolManager;
+import org.apache.phoenix.hbase.index.parallel.WaitForCompletionTaskRunner;
+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.util.IndexUtil;
+
+import com.google.common.collect.Multimap;
+
+/**
+ * Like the {@link ParallelWriterIndexCommitter}, but blocks until all writes have attempted to allow the caller to
+ * retrieve the failed and succeeded index updates. Therefore, this class will be a lot slower, in the face of failures,
+ * when compared to the {@link ParallelWriterIndexCommitter} (though as fast for writes), so it should be used only when
+ * you need to at least attempt all writes and know their result; for instance, this is fine for doing WAL recovery -
+ * it's not a performance intensive situation and we want to limit the the edits we need to retry.
+ * <p>
+ * On failure to {@link #write(Multimap)}, we return a {@link MultiIndexWriteFailureException} that contains the list of
+ * {@link HTableInterfaceReference} that didn't complete successfully.
+ * <p>
+ * Failures to write to the index can happen several different ways:
+ * <ol>
+ * <li><tt>this</tt> is {@link #stop(String) stopped} or aborted (via the passed {@link Abortable}. This causing any
+ * pending tasks to fail whatever they are doing as fast as possible. Any writes that have not begun are not even
+ * attempted and marked as failures.</li>
+ * <li>A batch write fails. This is the generic HBase write failure - it may occur because the index table is not
+ * available, .META. or -ROOT- is unavailable, or any other (of many) possible HBase exceptions.</li>
+ * </ol>
+ * Regardless of how the write fails, we still wait for all writes to complete before passing the failure back to the
+ * client.
+ */
+public class TrackingParallelWriterIndexCommitter implements IndexCommitter {
+ private static final Log LOG = LogFactory.getLog(TrackingParallelWriterIndexCommitter.class);
+
+ public static final String NUM_CONCURRENT_INDEX_WRITER_THREADS_CONF_KEY = "index.writer.threads.max";
+ private static final int DEFAULT_CONCURRENT_INDEX_WRITER_THREADS = 10;
+ private static final String INDEX_WRITER_KEEP_ALIVE_TIME_CONF_KEY = "index.writer.threads.keepalivetime";
+
+ private TaskRunner pool;
+ private HTableFactory factory;
+ private CapturingAbortable abortable;
+ private Stoppable stopped;
+ private RegionCoprocessorEnvironment env;
+ private KeyValueBuilder kvBuilder;
+
+ // for testing
+ public TrackingParallelWriterIndexCommitter(String hbaseVersion) {
+ kvBuilder = KeyValueBuilder.get(hbaseVersion);
+ }
+
+ public TrackingParallelWriterIndexCommitter() {
+ }
+
+ @Override
+ public void setup(IndexWriter parent, RegionCoprocessorEnvironment env, String name) {
+ this.env = env;
+ Configuration conf = env.getConfiguration();
+ setup(IndexWriterUtils.getDefaultDelegateHTableFactory(env),
+ ThreadPoolManager.getExecutor(
+ new ThreadPoolBuilder(name, conf).setMaxThread(NUM_CONCURRENT_INDEX_WRITER_THREADS_CONF_KEY,
+ DEFAULT_CONCURRENT_INDEX_WRITER_THREADS).setCoreTimeout(
+ INDEX_WRITER_KEEP_ALIVE_TIME_CONF_KEY), env), env.getRegionServerServices(), parent, env);
+ this.kvBuilder = KeyValueBuilder.get(env.getHBaseVersion());
+ }
+
+ /**
+ * Setup <tt>this</tt>.
+ * <p>
+ * Exposed for TESTING
+ */
+ void setup(HTableFactory factory, ExecutorService pool, Abortable abortable, Stoppable stop,
+ RegionCoprocessorEnvironment env) {
+ this.pool = new WaitForCompletionTaskRunner(pool);
+ this.factory = factory;
+ this.abortable = new CapturingAbortable(abortable);
+ this.stopped = stop;
+ }
+
+ @Override
+ public void write(Multimap<HTableInterfaceReference, Mutation> toWrite, final boolean allowLocalUpdates) throws MultiIndexWriteFailureException {
+ Set<Entry<HTableInterfaceReference, Collection<Mutation>>> entries = toWrite.asMap().entrySet();
+ TaskBatch<Boolean> tasks = new TaskBatch<Boolean>(entries.size());
+ List<HTableInterfaceReference> tables = new ArrayList<HTableInterfaceReference>(entries.size());
+ for (Entry<HTableInterfaceReference, Collection<Mutation>> entry : entries) {
+ // get the mutations for each table. We leak the implementation here a little bit to save
+ // doing a complete copy over of all the index update for each table.
+ final List<Mutation> mutations = kvBuilder.cloneIfNecessary((List<Mutation>)entry.getValue());
+ // track each reference so we can get at it easily later, when determing failures
+ final HTableInterfaceReference tableReference = entry.getKey();
+ final RegionCoprocessorEnvironment env = this.env;
+ if (env != null
+ && !allowLocalUpdates
+ && tableReference.getTableName().equals(
+ env.getRegion().getTableDesc().getNameAsString())) {
+ continue;
+ }
+ tables.add(tableReference);
+
+ /*
+ * Write a batch of index updates to an index table. This operation stops (is cancelable) via two
+ * mechanisms: (1) setting aborted or stopped on the IndexWriter or, (2) interrupting the running thread.
+ * The former will only work if we are not in the midst of writing the current batch to the table, though we
+ * do check these status variables before starting and before writing the batch. The latter usage,
+ * interrupting the thread, will work in the previous situations as was at some points while writing the
+ * batch, depending on the underlying writer implementation (HTableInterface#batch is blocking, but doesn't
+ * elaborate when is supports an interrupt).
+ */
+ tasks.add(new Task<Boolean>() {
+
+ /**
+ * Do the actual write to the primary table.
+ */
+ @SuppressWarnings("deprecation")
+ @Override
+ public Boolean call() throws Exception {
+ HTableInterface table = null;
+ try {
+ // this may have been queued, but there was an abort/stop so we try to early exit
+ throwFailureIfDone();
+ if (allowLocalUpdates && env!=null && tableReference.getTableName().equals(
+ env.getRegion().getTableDesc().getNameAsString())) {
+ try {
+ throwFailureIfDone();
+ IndexUtil.writeLocalUpdates(env.getRegion(), mutations, true);
+ return Boolean.TRUE;
+ } catch (IOException ignord) {
+ // when it's failed we fall back to the standard & slow way
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("indexRegion.batchMutate failed and fall back to HTable.batch(). Got error="
+ + ignord);
+ }
+ }
+ }
+
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Writing index update:" + mutations + " to table: " + tableReference);
+ }
+
+ table = factory.getTable(tableReference.get());
+ throwFailureIfDone();
+ table.batch(mutations);
+ } catch (InterruptedException e) {
+ // reset the interrupt status on the thread
+ Thread.currentThread().interrupt();
+ throw e;
+ } catch (Exception e) {
+ throw e;
+ } finally {
+ if (table != null) {
+ table.close();
+ }
+ }
+ return Boolean.TRUE;
+ }
+
+ private void throwFailureIfDone() throws SingleIndexWriteFailureException {
+ if (stopped.isStopped() || abortable.isAborted() || Thread.currentThread().isInterrupted()) { throw new SingleIndexWriteFailureException(
+ "Pool closed, not attempting to write to the index!", null); }
+
+ }
+ });
+ }
+
+ List<Boolean> results = null;
+ try {
+ LOG.debug("Waiting on index update tasks to complete...");
+ results = this.pool.submitUninterruptible(tasks);
+ } catch (ExecutionException e) {
+ throw new RuntimeException("Should not fail on the results while using a WaitForCompletionTaskRunner", e);
+ } catch (EarlyExitFailure e) {
+ throw new RuntimeException("Stopped while waiting for batch, quiting!", e);
+ }
+
+ // track the failures. We only ever access this on return from our calls, so no extra
+ // synchronization is needed. We could update all the failures as we find them, but that add a
+ // lot of locking overhead, and just doing the copy later is about as efficient.
+ List<HTableInterfaceReference> failures = new ArrayList<HTableInterfaceReference>();
+ int index = 0;
+ for (Boolean result : results) {
+ // there was a failure
+ if (result == null) {
+ // we know which table failed by the index of the result
+ failures.add(tables.get(index));
+ }
+ index++;
+ }
+
+ // if any of the tasks failed, then we need to propagate the failure
+ if (failures.size() > 0) {
+ // make the list unmodifiable to avoid any more synchronization concerns
+ throw new MultiIndexWriteFailureException(Collections.unmodifiableList(failures));
+ }
+ return;
+ }
+
+ @Override
+ public void stop(String why) {
+ LOG.info("Shutting down " + this.getClass().getSimpleName());
+ this.pool.stop(why);
+ this.factory.shutdown();
+ }
+
+ @Override
+ public boolean isStopped() {
+ return this.stopped.isStopped();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/55ff6018/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/recovery/StoreFailuresInCachePolicy.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/recovery/StoreFailuresInCachePolicy.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/recovery/StoreFailuresInCachePolicy.java
index 189f970..e28a0bd 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/recovery/StoreFailuresInCachePolicy.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/recovery/StoreFailuresInCachePolicy.java
@@ -30,6 +30,7 @@ import org.apache.phoenix.hbase.index.exception.MultiIndexWriteFailureException;
import org.apache.phoenix.hbase.index.table.HTableInterfaceReference;
import org.apache.phoenix.hbase.index.write.IndexFailurePolicy;
import org.apache.phoenix.hbase.index.write.KillServerOnFailurePolicy;
+import org.apache.phoenix.hbase.index.write.TrackingParallelWriterIndexCommitter;
/**
* Tracks any failed writes in The {@link PerRegionIndexWriteCache}, given a
http://git-wip-us.apache.org/repos/asf/phoenix/blob/55ff6018/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/recovery/TrackingParallelWriterIndexCommitter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/recovery/TrackingParallelWriterIndexCommitter.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/recovery/TrackingParallelWriterIndexCommitter.java
deleted file mode 100644
index 2b15a5e..0000000
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/recovery/TrackingParallelWriterIndexCommitter.java
+++ /dev/null
@@ -1,236 +0,0 @@
-/*
- * 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.hbase.index.write.recovery;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map.Entry;
-import java.util.Set;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.Abortable;
-import org.apache.hadoop.hbase.Stoppable;
-import org.apache.hadoop.hbase.client.HTableInterface;
-import org.apache.hadoop.hbase.client.Mutation;
-import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
-import org.apache.phoenix.hbase.index.CapturingAbortable;
-import org.apache.phoenix.hbase.index.exception.MultiIndexWriteFailureException;
-import org.apache.phoenix.hbase.index.exception.SingleIndexWriteFailureException;
-import org.apache.phoenix.hbase.index.parallel.EarlyExitFailure;
-import org.apache.phoenix.hbase.index.parallel.Task;
-import org.apache.phoenix.hbase.index.parallel.TaskBatch;
-import org.apache.phoenix.hbase.index.parallel.TaskRunner;
-import org.apache.phoenix.hbase.index.parallel.ThreadPoolBuilder;
-import org.apache.phoenix.hbase.index.parallel.ThreadPoolManager;
-import org.apache.phoenix.hbase.index.parallel.WaitForCompletionTaskRunner;
-import org.apache.phoenix.hbase.index.table.HTableFactory;
-import org.apache.phoenix.hbase.index.table.HTableInterfaceReference;
-import org.apache.phoenix.hbase.index.write.IndexCommitter;
-import org.apache.phoenix.hbase.index.write.IndexWriter;
-import org.apache.phoenix.hbase.index.write.IndexWriterUtils;
-import org.apache.phoenix.hbase.index.write.ParallelWriterIndexCommitter;
-import org.apache.phoenix.util.IndexUtil;
-
-import com.google.common.collect.Multimap;
-
-/**
- * Like the {@link ParallelWriterIndexCommitter}, but blocks until all writes have attempted to allow the caller to
- * retrieve the failed and succeeded index updates. Therefore, this class will be a lot slower, in the face of failures,
- * when compared to the {@link ParallelWriterIndexCommitter} (though as fast for writes), so it should be used only when
- * you need to at least attempt all writes and know their result; for instance, this is fine for doing WAL recovery -
- * it's not a performance intensive situation and we want to limit the the edits we need to retry.
- * <p>
- * On failure to {@link #write(Multimap)}, we return a {@link MultiIndexWriteFailureException} that contains the list of
- * {@link HTableInterfaceReference} that didn't complete successfully.
- * <p>
- * Failures to write to the index can happen several different ways:
- * <ol>
- * <li><tt>this</tt> is {@link #stop(String) stopped} or aborted (via the passed {@link Abortable}. This causing any
- * pending tasks to fail whatever they are doing as fast as possible. Any writes that have not begun are not even
- * attempted and marked as failures.</li>
- * <li>A batch write fails. This is the generic HBase write failure - it may occur because the index table is not
- * available, .META. or -ROOT- is unavailable, or any other (of many) possible HBase exceptions.</li>
- * </ol>
- * Regardless of how the write fails, we still wait for all writes to complete before passing the failure back to the
- * client.
- */
-public class TrackingParallelWriterIndexCommitter implements IndexCommitter {
- private static final Log LOG = LogFactory.getLog(TrackingParallelWriterIndexCommitter.class);
-
- public static final String NUM_CONCURRENT_INDEX_WRITER_THREADS_CONF_KEY = "index.trackingwriter.threads.max";
- private static final int DEFAULT_CONCURRENT_INDEX_WRITER_THREADS = 10;
- private static final String INDEX_WRITER_KEEP_ALIVE_TIME_CONF_KEY = "index.trackingwriter.threads.keepalivetime";
-
- private TaskRunner pool;
- private HTableFactory factory;
- private CapturingAbortable abortable;
- private Stoppable stopped;
- private RegionCoprocessorEnvironment env;
-
- @Override
- public void setup(IndexWriter parent, RegionCoprocessorEnvironment env, String name) {
- this.env = env;
- Configuration conf = env.getConfiguration();
- setup(IndexWriterUtils.getDefaultDelegateHTableFactory(env),
- ThreadPoolManager.getExecutor(
- new ThreadPoolBuilder(name, conf).setMaxThread(NUM_CONCURRENT_INDEX_WRITER_THREADS_CONF_KEY,
- DEFAULT_CONCURRENT_INDEX_WRITER_THREADS).setCoreTimeout(
- INDEX_WRITER_KEEP_ALIVE_TIME_CONF_KEY), env), env.getRegionServerServices(), parent, env);
- }
-
- /**
- * Setup <tt>this</tt>.
- * <p>
- * Exposed for TESTING
- */
- void setup(HTableFactory factory, ExecutorService pool, Abortable abortable, Stoppable stop,
- RegionCoprocessorEnvironment env) {
- this.pool = new WaitForCompletionTaskRunner(pool);
- this.factory = factory;
- this.abortable = new CapturingAbortable(abortable);
- this.stopped = stop;
- }
-
- @Override
- public void write(Multimap<HTableInterfaceReference, Mutation> toWrite, final boolean allowLocalUpdates) throws MultiIndexWriteFailureException {
- Set<Entry<HTableInterfaceReference, Collection<Mutation>>> entries = toWrite.asMap().entrySet();
- TaskBatch<Boolean> tasks = new TaskBatch<Boolean>(entries.size());
- List<HTableInterfaceReference> tables = new ArrayList<HTableInterfaceReference>(entries.size());
- for (Entry<HTableInterfaceReference, Collection<Mutation>> entry : entries) {
- // get the mutations for each table. We leak the implementation here a little bit to save
- // doing a complete copy over of all the index update for each table.
- final List<Mutation> mutations = (List<Mutation>)entry.getValue();
- // track each reference so we can get at it easily later, when determing failures
- final HTableInterfaceReference tableReference = entry.getKey();
- final RegionCoprocessorEnvironment env = this.env;
- if (env != null
- && !allowLocalUpdates
- && tableReference.getTableName().equals(
- env.getRegion().getTableDesc().getNameAsString())) {
- continue;
- }
- tables.add(tableReference);
-
- /*
- * Write a batch of index updates to an index table. This operation stops (is cancelable) via two
- * mechanisms: (1) setting aborted or stopped on the IndexWriter or, (2) interrupting the running thread.
- * The former will only work if we are not in the midst of writing the current batch to the table, though we
- * do check these status variables before starting and before writing the batch. The latter usage,
- * interrupting the thread, will work in the previous situations as was at some points while writing the
- * batch, depending on the underlying writer implementation (HTableInterface#batch is blocking, but doesn't
- * elaborate when is supports an interrupt).
- */
- tasks.add(new Task<Boolean>() {
-
- /**
- * Do the actual write to the primary table.
- */
- @SuppressWarnings("deprecation")
- @Override
- public Boolean call() throws Exception {
- HTableInterface table = null;
- try {
- // this may have been queued, but there was an abort/stop so we try to early exit
- throwFailureIfDone();
- if (allowLocalUpdates && env!=null && tableReference.getTableName().equals(
- env.getRegion().getTableDesc().getNameAsString())) {
- try {
- throwFailureIfDone();
- IndexUtil.writeLocalUpdates(env.getRegion(), mutations, true);
- return Boolean.TRUE;
- } catch (IOException ignord) {
- // when it's failed we fall back to the standard & slow way
- if (LOG.isTraceEnabled()) {
- LOG.trace("indexRegion.batchMutate failed and fall back to HTable.batch(). Got error="
- + ignord);
- }
- }
- }
-
- if (LOG.isTraceEnabled()) {
- LOG.trace("Writing index update:" + mutations + " to table: " + tableReference);
- }
-
- table = factory.getTable(tableReference.get());
- throwFailureIfDone();
- table.batch(mutations);
- } catch (InterruptedException e) {
- // reset the interrupt status on the thread
- Thread.currentThread().interrupt();
- throw e;
- } catch (Exception e) {
- throw e;
- } finally {
- if (table != null) {
- table.close();
- }
- }
- return Boolean.TRUE;
- }
-
- private void throwFailureIfDone() throws SingleIndexWriteFailureException {
- if (stopped.isStopped() || abortable.isAborted() || Thread.currentThread().isInterrupted()) { throw new SingleIndexWriteFailureException(
- "Pool closed, not attempting to write to the index!", null); }
-
- }
- });
- }
-
- List<Boolean> results = null;
- try {
- LOG.debug("Waiting on index update tasks to complete...");
- results = this.pool.submitUninterruptible(tasks);
- } catch (ExecutionException e) {
- throw new RuntimeException("Should not fail on the results while using a WaitForCompletionTaskRunner", e);
- } catch (EarlyExitFailure e) {
- throw new RuntimeException("Stopped while waiting for batch, quiting!", e);
- }
-
- // track the failures. We only ever access this on return from our calls, so no extra
- // synchronization is needed. We could update all the failures as we find them, but that add a
- // lot of locking overhead, and just doing the copy later is about as efficient.
- List<HTableInterfaceReference> failures = new ArrayList<HTableInterfaceReference>();
- int index = 0;
- for (Boolean result : results) {
- // there was a failure
- if (result == null) {
- // we know which table failed by the index of the result
- failures.add(tables.get(index));
- }
- index++;
- }
-
- // if any of the tasks failed, then we need to propagate the failure
- if (failures.size() > 0) {
- // make the list unmodifiable to avoid any more synchronization concerns
- throw new MultiIndexWriteFailureException(Collections.unmodifiableList(failures));
- }
- return;
- }
-
- @Override
- public void stop(String why) {
- LOG.info("Shutting down " + this.getClass().getSimpleName());
- this.pool.stop(why);
- this.factory.shutdown();
- }
-
- @Override
- public boolean isStopped() {
- return this.stopped.isStopped();
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/55ff6018/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixIndexFailurePolicy.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixIndexFailurePolicy.java b/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixIndexFailurePolicy.java
index c91e36e..a36e7bb 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixIndexFailurePolicy.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixIndexFailurePolicy.java
@@ -41,6 +41,7 @@ import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.phoenix.coprocessor.MetaDataProtocol.MetaDataMutationResult;
import org.apache.phoenix.coprocessor.MetaDataProtocol.MutationCode;
+import org.apache.phoenix.hbase.index.exception.MultiIndexWriteFailureException;
import org.apache.phoenix.hbase.index.table.HTableInterfaceReference;
import org.apache.phoenix.hbase.index.write.DelegateIndexFailurePolicy;
import org.apache.phoenix.hbase.index.write.KillServerOnFailurePolicy;
@@ -59,6 +60,7 @@ import org.apache.phoenix.util.SchemaUtil;
import org.apache.phoenix.util.ServerUtil;
import com.google.common.collect.Multimap;
+import com.google.common.collect.Sets;
/**
*
@@ -165,7 +167,20 @@ public class PhoenixIndexFailurePolicy extends DelegateIndexFailurePolicy {
// start by looking at all the tables to which we attempted to write
long timestamp = 0;
boolean leaveIndexActive = blockDataTableWritesOnFailure || !disableIndexOnFailure;
+ // if using TrackingParallelWriter, we know which indexes failed and only disable those
+ Set<HTableInterfaceReference> failedTables = Sets.newHashSetWithExpectedSize(attempted.size());
+ try {
+ throw cause;
+ } catch (MultiIndexWriteFailureException miwfe) {
+ failedTables.addAll(miwfe.getFailedTables());
+ } catch (Exception e) {
+ // do nothing
+ }
+
for (HTableInterfaceReference ref : refs) {
+ if (failedTables.size() > 0 && !failedTables.contains(ref)) {
+ continue; // leave index active if its writes succeeded
+ }
long minTimeStamp = 0;
// get the minimum timestamp across all the mutations we attempted on that table
http://git-wip-us.apache.org/repos/asf/phoenix/blob/55ff6018/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/write/TestIndexWriter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/write/TestIndexWriter.java b/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/write/TestIndexWriter.java
index 012f08e..282072d 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/write/TestIndexWriter.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/write/TestIndexWriter.java
@@ -131,7 +131,7 @@ public class TestIndexWriter {
tables.put(new ImmutableBytesPtr(tableName), table);
// setup the writer and failure policy
- ParallelWriterIndexCommitter committer = new ParallelWriterIndexCommitter(VersionInfo.getVersion());
+ TrackingParallelWriterIndexCommitter committer = new TrackingParallelWriterIndexCommitter(VersionInfo.getVersion());
committer.setup(factory, exec, abort, stop, e);
KillServerOnFailurePolicy policy = new KillServerOnFailurePolicy();
policy.setup(stop, abort);
@@ -199,7 +199,7 @@ public class TestIndexWriter {
tables.put(new ImmutableBytesPtr(tableName), table);
tables.put(new ImmutableBytesPtr(tableName2), table2);
- ParallelWriterIndexCommitter committer = new ParallelWriterIndexCommitter(VersionInfo.getVersion());
+ TrackingParallelWriterIndexCommitter committer = new TrackingParallelWriterIndexCommitter(VersionInfo.getVersion());
committer.setup(factory, exec, abort, stop, e);
KillServerOnFailurePolicy policy = new KillServerOnFailurePolicy();
policy.setup(stop, abort);
@@ -279,7 +279,7 @@ public class TestIndexWriter {
indexUpdates.add(new Pair<Mutation, byte[]>(m, tableName));
// setup the writer
- ParallelWriterIndexCommitter committer = new ParallelWriterIndexCommitter(VersionInfo.getVersion());
+ TrackingParallelWriterIndexCommitter committer = new TrackingParallelWriterIndexCommitter(VersionInfo.getVersion());
committer.setup(factory, exec, abort, stop, e );
KillServerOnFailurePolicy policy = new KillServerOnFailurePolicy();
policy.setup(stop, abort);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/55ff6018/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/write/TestParalleIndexWriter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/write/TestParalleIndexWriter.java b/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/write/TestParalleIndexWriter.java
index e62af7a..3e2b47c 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/write/TestParalleIndexWriter.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/write/TestParalleIndexWriter.java
@@ -66,7 +66,7 @@ public class TestParalleIndexWriter {
Mockito.when(e.getSharedData()).thenReturn(new ConcurrentHashMap<String,Object>());
FakeTableFactory factory = new FakeTableFactory(
Collections.<ImmutableBytesPtr, HTableInterface> emptyMap());
- ParallelWriterIndexCommitter writer = new ParallelWriterIndexCommitter(VersionInfo.getVersion());
+ TrackingParallelWriterIndexCommitter writer = new TrackingParallelWriterIndexCommitter(VersionInfo.getVersion());
Abortable mockAbort = Mockito.mock(Abortable.class);
Stoppable mockStop = Mockito.mock(Stoppable.class);
// create a simple writer
@@ -116,7 +116,7 @@ public class TestParalleIndexWriter {
tables.put(tableName, table);
// setup the writer and failure policy
- ParallelWriterIndexCommitter writer = new ParallelWriterIndexCommitter(VersionInfo.getVersion());
+ TrackingParallelWriterIndexCommitter writer = new TrackingParallelWriterIndexCommitter(VersionInfo.getVersion());
writer.setup(factory, exec, abort, stop, e);
writer.write(indexUpdates, true);
assertTrue("Writer returned before the table batch completed! Likely a race condition tripped",
http://git-wip-us.apache.org/repos/asf/phoenix/blob/55ff6018/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/write/TestParalleWriterIndexCommitter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/write/TestParalleWriterIndexCommitter.java b/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/write/TestParalleWriterIndexCommitter.java
index 789e7a1..32a6661 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/write/TestParalleWriterIndexCommitter.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/write/TestParalleWriterIndexCommitter.java
@@ -62,7 +62,7 @@ public class TestParalleWriterIndexCommitter {
ExecutorService exec = Executors.newFixedThreadPool(1);
FakeTableFactory factory = new FakeTableFactory(
Collections.<ImmutableBytesPtr, HTableInterface> emptyMap());
- ParallelWriterIndexCommitter writer = new ParallelWriterIndexCommitter(VersionInfo.getVersion());
+ TrackingParallelWriterIndexCommitter writer = new TrackingParallelWriterIndexCommitter(VersionInfo.getVersion());
Abortable mockAbort = Mockito.mock(Abortable.class);
Stoppable mockStop = Mockito.mock(Stoppable.class);
RegionCoprocessorEnvironment e =Mockito.mock(RegionCoprocessorEnvironment.class);
@@ -117,7 +117,7 @@ public class TestParalleWriterIndexCommitter {
tables.put(tableName, table);
// setup the writer and failure policy
- ParallelWriterIndexCommitter writer = new ParallelWriterIndexCommitter(VersionInfo.getVersion());
+ TrackingParallelWriterIndexCommitter writer = new TrackingParallelWriterIndexCommitter(VersionInfo.getVersion());
writer.setup(factory, exec, abort, stop, e);
writer.write(indexUpdates, true);
assertTrue("Writer returned before the table batch completed! Likely a race condition tripped",
[3/3] phoenix git commit: PHOENIX-4128 CNF
org/apache/commons/math3/exception/OutOfRangeException (Ethan Wang)
Posted by ja...@apache.org.
PHOENIX-4128 CNF org/apache/commons/math3/exception/OutOfRangeException (Ethan Wang)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/3572cfc1
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/3572cfc1
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/3572cfc1
Branch: refs/heads/4.x-HBase-1.1
Commit: 3572cfc1e5ca7e5e36fcfa6a980de5723fa41fbc
Parents: 6ac72c4
Author: James Taylor <jt...@salesforce.com>
Authored: Wed Aug 30 16:40:04 2017 -0700
Committer: James Taylor <jt...@salesforce.com>
Committed: Wed Aug 30 17:32:24 2017 -0700
----------------------------------------------------------------------
.../src/main/java/org/apache/phoenix/parse/ConcreteTableNode.java | 3 +--
1 file changed, 1 insertion(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/3572cfc1/phoenix-core/src/main/java/org/apache/phoenix/parse/ConcreteTableNode.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/parse/ConcreteTableNode.java b/phoenix-core/src/main/java/org/apache/phoenix/parse/ConcreteTableNode.java
index 6a4ed6f..c9fd51b 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/parse/ConcreteTableNode.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/parse/ConcreteTableNode.java
@@ -17,7 +17,6 @@
*/
package org.apache.phoenix.parse;
-import org.apache.commons.math3.exception.OutOfRangeException;
import org.apache.phoenix.util.SchemaUtil;
/**
@@ -43,7 +42,7 @@ public abstract class ConcreteTableNode extends TableNode {
if(tableSamplingRate==null){
this.tableSamplingRate=DEFAULT_TABLE_SAMPLING_RATE;
}else if(tableSamplingRate<0d||tableSamplingRate>100d){
- throw new OutOfRangeException(tableSamplingRate, 0d, 100d);
+ throw new IllegalArgumentException("TableSamplingRate is out of bound of 0 and 100");
}else{
this.tableSamplingRate=tableSamplingRate;
}
[2/3] phoenix git commit: PHOENIX-418 Include APPROX_COUNT_DISTINCT
in server jar for support of approximate COUNT DISTINCT (Ethan Wang)
Posted by ja...@apache.org.
PHOENIX-418 Include APPROX_COUNT_DISTINCT in server jar for support of approximate COUNT DISTINCT (Ethan Wang)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/cab3a398
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/cab3a398
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/cab3a398
Branch: refs/heads/4.x-HBase-1.1
Commit: cab3a398b459a6c773e9bf1bec5036911a218511
Parents: 3572cfc
Author: James Taylor <jt...@salesforce.com>
Authored: Wed Aug 30 16:42:44 2017 -0700
Committer: James Taylor <jt...@salesforce.com>
Committed: Wed Aug 30 17:32:24 2017 -0700
----------------------------------------------------------------------
phoenix-core/pom.xml | 2 +-
phoenix-server/pom.xml | 1 +
pom.xml | 7 ++++++-
3 files changed, 8 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/cab3a398/phoenix-core/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-core/pom.xml b/phoenix-core/pom.xml
index 7ad32af..c2ee7e7 100644
--- a/phoenix-core/pom.xml
+++ b/phoenix-core/pom.xml
@@ -473,7 +473,7 @@
<dependency>
<groupId>com.clearspring.analytics</groupId>
<artifactId>stream</artifactId>
- <version>2.9.5</version>
+ <version>${stream.version}</version>
</dependency>
</dependencies>
</project>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/cab3a398/phoenix-server/pom.xml
----------------------------------------------------------------------
diff --git a/phoenix-server/pom.xml b/phoenix-server/pom.xml
index 3db172c..71957ca 100644
--- a/phoenix-server/pom.xml
+++ b/phoenix-server/pom.xml
@@ -133,6 +133,7 @@
<include>it.unimi.dsi:fastutil</include>
<include>io.dropwizard.metrics:metrics-core</include>
<include>org.apache.thrift:libthrift</include>
+ <include>com.clearspring.analytics:stream</include>
</includes>
<excludes>
<exclude>org.apache.phoenix:phoenix-server</exclude>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/cab3a398/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 0e5ab88..af09598 100644
--- a/pom.xml
+++ b/pom.xml
@@ -105,7 +105,7 @@
<spark.version>2.0.2</spark.version>
<scala.version>2.11.8</scala.version>
<scala.binary.version>2.11</scala.binary.version>
-
+ <stream.version>2.9.5</stream.version>
<!-- Test Dependencies -->
<mockito-all.version>1.8.5</mockito-all.version>
<junit.version>4.12</junit.version>
@@ -970,6 +970,11 @@
<artifactId>curator-client</artifactId>
<version>${curator.version}</version>
</dependency>
+ <dependency>
+ <groupId>com.clearspring.analytics</groupId>
+ <artifactId>stream</artifactId>
+ <version>${stream.version}</version>
+ </dependency>
</dependencies>
</dependencyManagement>