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/25 20:27:39 UTC
[1/3] phoenix git commit: PHOENIX-3953 Clear INDEX_DISABLED_TIMESTAMP
and disable index on compaction
Repository: phoenix
Updated Branches:
refs/heads/master a5fb51948 -> 435441ea8
PHOENIX-3953 Clear INDEX_DISABLED_TIMESTAMP and disable index on compaction
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/435441ea
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/435441ea
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/435441ea
Branch: refs/heads/master
Commit: 435441ea8ba336e1967b03cf84f1868c5ef14790
Parents: 7d3368b
Author: James Taylor <jt...@salesforce.com>
Authored: Fri Aug 25 11:38:41 2017 -0700
Committer: James Taylor <jt...@salesforce.com>
Committed: Fri Aug 25 13:27:23 2017 -0700
----------------------------------------------------------------------
.../end2end/index/PartialIndexRebuilderIT.java | 32 ++++++++++++++
.../org/apache/phoenix/hbase/index/Indexer.java | 45 ++++++++++++++++++++
.../java/org/apache/phoenix/util/IndexUtil.java | 7 ++-
3 files changed, 83 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/435441ea/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/PartialIndexRebuilderIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/PartialIndexRebuilderIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/PartialIndexRebuilderIT.java
index f42d2c9..9483e87 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/PartialIndexRebuilderIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/PartialIndexRebuilderIT.java
@@ -245,6 +245,38 @@ public class PartialIndexRebuilderIT extends BaseUniqueNamesOwnClusterIT {
}
@Test
+ public void testCompactionDuringRebuild() throws Throwable {
+ String schemaName = generateUniqueName();
+ String tableName = generateUniqueName();
+ String indexName = generateUniqueName();
+ final String fullTableName = SchemaUtil.getTableName(schemaName, tableName);
+ String fullIndexName = SchemaUtil.getTableName(schemaName, indexName);
+ final MyClock clock = new MyClock(1000);
+ // Use our own clock to prevent race between partial rebuilder and compaction
+ EnvironmentEdgeManager.injectEdge(clock);
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ conn.createStatement().execute("CREATE TABLE " + fullTableName + "(k INTEGER PRIMARY KEY, v1 INTEGER, v2 INTEGER) COLUMN_ENCODED_BYTES = 0, STORE_NULLS=true, GUIDE_POSTS_WIDTH=1000");
+ clock.time += 1000;
+ conn.createStatement().execute("CREATE INDEX " + indexName + " ON " + fullTableName + " (v1) INCLUDE (v2)");
+ clock.time += 1000;
+ conn.createStatement().execute("UPSERT INTO " + fullTableName + " VALUES(1, 2, 3)");
+ conn.commit();
+ clock.time += 1000;
+ long disableTS = EnvironmentEdgeManager.currentTimeMillis();
+ HTableInterface metaTable = conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES);
+ IndexUtil.updateIndexState(fullIndexName, disableTS, metaTable, PIndexState.DISABLE);
+ TestUtil.doMajorCompaction(conn, fullIndexName);
+ assertFalse(TestUtil.checkIndexState(conn, fullIndexName, PIndexState.DISABLE, 0L));
+ TestUtil.analyzeTable(conn, fullTableName);
+ assertFalse(TestUtil.checkIndexState(conn, fullIndexName, PIndexState.DISABLE, 0L));
+ TestUtil.doMajorCompaction(conn, fullTableName);
+ assertTrue(TestUtil.checkIndexState(conn, fullIndexName, PIndexState.DISABLE, 0L));
+ } finally {
+ EnvironmentEdgeManager.injectEdge(null);
+ }
+ }
+
+ @Test
@Repeat(5)
public void testDeleteAndUpsertAfterFailure() throws Throwable {
final int nRows = 10;
http://git-wip-us.apache.org/repos/asf/phoenix/blob/435441ea/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
index ba8f3d4..b76752d 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
@@ -25,6 +25,7 @@ import static org.apache.phoenix.hbase.index.write.IndexWriterUtils.INDEX_WRITER
import java.io.IOException;
import java.security.PrivilegedExceptionAction;
+import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -87,12 +88,18 @@ import org.apache.phoenix.hbase.index.write.IndexWriter;
import org.apache.phoenix.hbase.index.write.RecoveryIndexWriter;
import org.apache.phoenix.hbase.index.write.recovery.PerRegionIndexWriteCache;
import org.apache.phoenix.hbase.index.write.recovery.StoreFailuresInCachePolicy;
+import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.query.QueryServicesOptions;
+import org.apache.phoenix.schema.PIndexState;
+import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.trace.TracingUtils;
import org.apache.phoenix.trace.util.NullSpan;
import org.apache.phoenix.util.EnvironmentEdgeManager;
+import org.apache.phoenix.util.IndexUtil;
import org.apache.phoenix.util.MetaDataUtil;
+import org.apache.phoenix.util.PhoenixRuntime;
import org.apache.phoenix.util.PropertiesUtil;
+import org.apache.phoenix.util.QueryUtil;
import org.apache.phoenix.util.ServerUtil;
import com.google.common.collect.Lists;
@@ -829,5 +836,43 @@ public class Indexer extends BaseRegionObserver {
properties.put(Indexer.INDEX_BUILDER_CONF_KEY, builder.getName());
desc.addCoprocessor(Indexer.class.getName(), null, priority, properties);
}
+
+ @Override
+ public InternalScanner preCompact(final ObserverContext<RegionCoprocessorEnvironment> c, final Store store,
+ final InternalScanner scanner, final ScanType scanType) throws IOException {
+ // Compaction and split upcalls run with the effective user context of the requesting user.
+ // This will lead to failure of cross cluster RPC if the effective user is not
+ // the login user. Switch to the login user context to ensure we have the expected
+ // security context.
+ return User.runAsLoginUser(new PrivilegedExceptionAction<InternalScanner>() {
+ @Override
+ public InternalScanner run() throws Exception {
+ InternalScanner internalScanner = scanner;
+ if (scanType.equals(ScanType.COMPACT_DROP_DELETES)) {
+ String fullTableName = c.getEnvironment().getRegion().getRegionInfo().getTable().getNameAsString();
+ try {
+ PhoenixConnection conn = QueryUtil.getConnectionOnServer(c.getEnvironment().getConfiguration()).unwrap(PhoenixConnection.class);
+ PTable table = PhoenixRuntime.getTableNoCache(conn, fullTableName);
+ for (PTable index : table.getIndexes()) {
+ if (index.getIndexDisableTimestamp() != 0) {
+ try {
+ IndexUtil.updateIndexState(conn, index.getName().getString(), PIndexState.DISABLE, Long.valueOf(0L));
+ } catch (SQLException e) {
+ LOG.warn("Unable to permanently disable index " + index.getName().getString(), e);
+ }
+ }
+ }
+ } catch (Exception e) {
+ // If we can't reach the stats table, don't interrupt the normal
+ // compaction operation, just log a warning.
+ if (LOG.isWarnEnabled()) {
+ LOG.warn("Unable to permanently disable indexes being partially rebuild for " + fullTableName, e);
+ }
+ }
+ }
+ return internalScanner;
+ }
+ });
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/435441ea/phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java
index 75a7e55..ea6fb96 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java
@@ -711,10 +711,15 @@ public class IndexUtil {
mutations.toArray(new Mutation[mutations.size()]),
HConstants.NO_NONCE, HConstants.NO_NONCE);
}
-
+
public static MetaDataMutationResult updateIndexState(String indexTableName, long minTimeStamp,
HTableInterface metaTable, PIndexState newState) throws Throwable {
byte[] indexTableKey = SchemaUtil.getTableKeyFromFullName(indexTableName);
+ return updateIndexState(indexTableKey, minTimeStamp, metaTable, newState);
+ }
+
+ public static MetaDataMutationResult updateIndexState(byte[] indexTableKey, long minTimeStamp,
+ HTableInterface metaTable, PIndexState newState) throws Throwable {
// Mimic the Put that gets generated by the client on an update of the index state
Put put = new Put(indexTableKey);
put.add(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, PhoenixDatabaseMetaData.INDEX_STATE_BYTES,
[2/3] phoenix git commit: PHOENIX-4124 Two new tests without Apache
licenses
Posted by ja...@apache.org.
PHOENIX-4124 Two new tests without Apache licenses
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/7d3368b6
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/7d3368b6
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/7d3368b6
Branch: refs/heads/master
Commit: 7d3368b62e91aa572cbb5c1e7083336d3f0c6538
Parents: 8b7a836
Author: James Taylor <jt...@salesforce.com>
Authored: Thu Aug 24 16:26:16 2017 -0700
Committer: James Taylor <jt...@salesforce.com>
Committed: Fri Aug 25 13:27:23 2017 -0700
----------------------------------------------------------------------
.../org/apache/phoenix/util/IndexScrutinyIT.java | 17 +++++++++++++++++
.../org/apache/phoenix/util/RunUntilFailure.java | 17 +++++++++++++++++
2 files changed, 34 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7d3368b6/phoenix-core/src/it/java/org/apache/phoenix/util/IndexScrutinyIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/util/IndexScrutinyIT.java b/phoenix-core/src/it/java/org/apache/phoenix/util/IndexScrutinyIT.java
index a703294..a5ec83f 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/util/IndexScrutinyIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/util/IndexScrutinyIT.java
@@ -1,3 +1,20 @@
+/*
+ * 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;
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7d3368b6/phoenix-core/src/test/java/org/apache/phoenix/util/RunUntilFailure.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/util/RunUntilFailure.java b/phoenix-core/src/test/java/org/apache/phoenix/util/RunUntilFailure.java
index 2b378c1..ed0a839 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/util/RunUntilFailure.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/util/RunUntilFailure.java
@@ -1,3 +1,20 @@
+/*
+ * 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 org.junit.Ignore;
[3/3] phoenix git commit: PHOENIX-4122 Prevent
IllegalMonitorStateException from occurring when releasing row locks
Posted by ja...@apache.org.
PHOENIX-4122 Prevent IllegalMonitorStateException from occurring when releasing row locks
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/8b7a836a
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/8b7a836a
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/8b7a836a
Branch: refs/heads/master
Commit: 8b7a836a5e5502c1c217c6e3b6ab7298b980cb81
Parents: a5fb519
Author: James Taylor <jt...@salesforce.com>
Authored: Thu Aug 24 14:16:30 2017 -0700
Committer: James Taylor <jt...@salesforce.com>
Committed: Fri Aug 25 13:27:23 2017 -0700
----------------------------------------------------------------------
.../org/apache/phoenix/hbase/index/Indexer.java | 103 +++++++------------
1 file changed, 37 insertions(+), 66 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/8b7a836a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
index 05e2ae7..ba8f3d4 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
@@ -27,6 +27,7 @@ import java.io.IOException;
import java.security.PrivilegedExceptionAction;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
@@ -71,6 +72,7 @@ import org.apache.htrace.Trace;
import org.apache.htrace.TraceScope;
import org.apache.phoenix.coprocessor.BaseScannerRegionObserver.ReplayWrite;
import org.apache.phoenix.coprocessor.DelegateRegionCoprocessorEnvironment;
+import org.apache.phoenix.hbase.index.LockManager.RowLock;
import org.apache.phoenix.hbase.index.builder.IndexBuildManager;
import org.apache.phoenix.hbase.index.builder.IndexBuilder;
import org.apache.phoenix.hbase.index.metrics.MetricsIndexerSource;
@@ -85,6 +87,7 @@ import org.apache.phoenix.hbase.index.write.IndexWriter;
import org.apache.phoenix.hbase.index.write.RecoveryIndexWriter;
import org.apache.phoenix.hbase.index.write.recovery.PerRegionIndexWriteCache;
import org.apache.phoenix.hbase.index.write.recovery.StoreFailuresInCachePolicy;
+import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.trace.TracingUtils;
import org.apache.phoenix.trace.util.NullSpan;
import org.apache.phoenix.util.EnvironmentEdgeManager;
@@ -119,6 +122,9 @@ import com.google.common.collect.Multimap;
public class Indexer extends BaseRegionObserver {
private static final Log LOG = LogFactory.getLog(Indexer.class);
+ private static final OperationStatus IGNORE = new OperationStatus(OperationStatusCode.SUCCESS);
+ private static final OperationStatus NOWRITE = new OperationStatus(OperationStatusCode.SUCCESS);
+
protected IndexWriter writer;
protected IndexBuildManager builder;
@@ -126,8 +132,13 @@ public class Indexer extends BaseRegionObserver {
// Hack to get around not being able to save any state between
// coprocessor calls. TODO: remove after HBASE-18127 when available
- private ThreadLocal<Collection<Pair<Mutation, byte[]>>> indexUpdates =
- new ThreadLocal<Collection<Pair<Mutation, byte[]>>>();
+ private static class BatchMutateContext {
+ public Collection<Pair<Mutation, byte[]>> indexUpdates = Collections.emptyList();
+ public List<RowLock> rowLocks = Lists.newArrayListWithExpectedSize(QueryServicesOptions.DEFAULT_MUTATE_BATCH_SIZE);
+ }
+
+ private ThreadLocal<BatchMutateContext> batchMutateContext =
+ new ThreadLocal<BatchMutateContext>();
/** Configuration key for the {@link IndexBuilder} to use */
public static final String INDEX_BUILDER_CONF_KEY = "index.builder";
@@ -364,10 +375,6 @@ public class Indexer extends BaseRegionObserver {
"Somehow didn't return an index update but also didn't propagate the failure to the client!");
}
- private static final OperationStatus IGNORE = new OperationStatus(OperationStatusCode.SUCCESS);
- private static final OperationStatus NOWRITE = new OperationStatus(OperationStatusCode.SUCCESS);
- private static final OperationStatus FAILURE = new OperationStatus(OperationStatusCode.FAILURE, "Unable to acquire row lock");
-
// Assume time stamp of mutation a client defined time stamp if it's not within
// a factor of ten of the current time.
// TODO: get rid of this and have client pass LATEST_TIMESTAMP unless an SCN is set
@@ -398,6 +405,8 @@ public class Indexer extends BaseRegionObserver {
* Exclusively lock all rows so we get a consistent read
* while determining the index updates
*/
+ BatchMutateContext context = new BatchMutateContext();
+ setBatchMutateContext(c, context);
Durability durability = Durability.SKIP_WAL;
boolean copyMutations = false;
for (int i = 0; i < miniBatchOp.size(); i++) {
@@ -407,31 +416,7 @@ public class Indexer extends BaseRegionObserver {
continue;
}
if (this.builder.isEnabled(m)) {
- boolean success = false;
- try {
- lockManager.lockRow(m.getRow(), rowLockWaitDuration);
- success = true;
- } finally {
- if (!success) {
- // We're throwing here as a result of either a timeout while waiting
- // for the row lock or an interrupt. Either way, the lock on the
- // current row was unsuccessful and we won't be locking any more rows
- // since we're throwing. By setting the operation status to FAILURE
- // here, we prevent the attempt to unlock rows we've never locked when
- // postBatchMutateIndispensably is executed. We're very limited wrt
- // the state that can be shared between the batch mutate coprocessor
- // calls (see HBASE-18127).
- // Note that we shouldn't necessarily be throwing here, since we're
- // essentially failing the data write because we can't do the locking
- // necessary for performing consistent index maintenance. We'd ideally
- // want to go through the index failure policy to determine what action
- // to perform. We currently cannot ignore this lock failure as we lack
- // the ability to keep that state (PHOENIX-4055).
- for (int j = i; j < miniBatchOp.size(); j++) {
- miniBatchOp.setOperationStatus(j,FAILURE);
- }
- }
- }
+ context.rowLocks.add(lockManager.lockRow(m.getRow(), rowLockWaitDuration));
Durability effectiveDurablity = (m.getDurability() == Durability.USE_DEFAULT) ?
defaultDurability : m.getDurability();
if (effectiveDurablity.ordinal() > durability.ordinal()) {
@@ -563,7 +548,7 @@ public class Indexer extends BaseRegionObserver {
localUpdates.toArray(new Mutation[localUpdates.size()]));
}
if (!indexUpdates.isEmpty()) {
- setIndexUpdates(c, indexUpdates);
+ context.indexUpdates = indexUpdates;
// write index updates to WAL
if (durability != Durability.SKIP_WAL) {
// we have all the WAL durability, so we just update the WAL entry and move on
@@ -575,16 +560,16 @@ public class Indexer extends BaseRegionObserver {
}
}
- private void setIndexUpdates(ObserverContext<RegionCoprocessorEnvironment> c, Collection<Pair<Mutation, byte[]>> indexUpdates) {
- this.indexUpdates.set(indexUpdates);
+ private void setBatchMutateContext(ObserverContext<RegionCoprocessorEnvironment> c, BatchMutateContext context) {
+ this.batchMutateContext.set(context);
}
- private Collection<Pair<Mutation, byte[]>> getIndexUpdates(ObserverContext<RegionCoprocessorEnvironment> c) {
- return this.indexUpdates.get();
+ private BatchMutateContext getBatchMutateContext(ObserverContext<RegionCoprocessorEnvironment> c) {
+ return this.batchMutateContext.get();
}
- private void removeIndexUpdates(ObserverContext<RegionCoprocessorEnvironment> c) {
- this.indexUpdates.remove();
+ private void removeBatchMutateContext(ObserverContext<RegionCoprocessorEnvironment> c) {
+ this.batchMutateContext.remove();
}
@Override
@@ -595,21 +580,21 @@ public class Indexer extends BaseRegionObserver {
return;
}
long start = EnvironmentEdgeManager.currentTimeMillis();
+ BatchMutateContext context = getBatchMutateContext(c);
+ if (context == null) {
+ return;
+ }
try {
- for (int i = 0; i < miniBatchOp.size(); i++) {
- OperationStatus status = miniBatchOp.getOperationStatus(i);
- if (status != IGNORE && status != FAILURE) {
- lockManager.unlockRow(miniBatchOp.getOperation(i).getRow());
- }
+ for (RowLock rowLock : context.rowLocks) {
+ rowLock.release();
}
this.builder.batchCompleted(miniBatchOp);
if (success) { // if miniBatchOp was successfully written, write index updates
- //each batch operation, only the first one will have anything useful, so we can just grab that
- Mutation mutation = miniBatchOp.getOperation(0);
- doPost(c, mutation);
+ doPost(c, context);
}
} finally {
+ removeBatchMutateContext(c);
long duration = EnvironmentEdgeManager.currentTimeMillis() - start;
if (duration >= slowIndexWriteThreshold) {
if (LOG.isDebugEnabled()) {
@@ -621,9 +606,9 @@ public class Indexer extends BaseRegionObserver {
}
}
- private void doPost(ObserverContext<RegionCoprocessorEnvironment> c, Mutation m) throws IOException {
+ private void doPost(ObserverContext<RegionCoprocessorEnvironment> c, BatchMutateContext context) throws IOException {
try {
- doPostWithExceptions(c,m);
+ doPostWithExceptions(c,context);
return;
} catch (Throwable e) {
rethrowIndexingException(e);
@@ -632,10 +617,10 @@ public class Indexer extends BaseRegionObserver {
"Somehow didn't complete the index update, but didn't return succesfully either!");
}
- private void doPostWithExceptions(ObserverContext<RegionCoprocessorEnvironment> c, Mutation m)
+ private void doPostWithExceptions(ObserverContext<RegionCoprocessorEnvironment> c, BatchMutateContext context)
throws IOException {
//short circuit, if we don't need to do any work
- if (!this.builder.isEnabled(m)) {
+ if (context == null || context.indexUpdates.isEmpty()) {
return;
}
@@ -646,23 +631,9 @@ public class Indexer extends BaseRegionObserver {
current = NullSpan.INSTANCE;
}
long start = EnvironmentEdgeManager.currentTimeMillis();
-
- Collection<Pair<Mutation, byte[]>> indexUpdates = getIndexUpdates(c);
- if (indexUpdates == null) {
- return;
- }
-
- // the WAL edit is kept in memory and we already specified the factory when we created the
- // references originally - therefore, we just pass in a null factory here and use the ones
- // already specified on each reference
- try {
- current.addTimelineAnnotation("Actually doing index update for first time");
- writer.writeAndKillYourselfOnFailure(indexUpdates, false);
- } finally { // With a custom kill policy, we may throw instead of kill the server.
- // mark the batch as having been written so it won't be written again
- removeIndexUpdates(c);
- }
+ current.addTimelineAnnotation("Actually doing index update for first time");
+ writer.writeAndKillYourselfOnFailure(context.indexUpdates, false);
long duration = EnvironmentEdgeManager.currentTimeMillis() - start;
if (duration >= slowIndexWriteThreshold) {