You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by kt...@apache.org on 2016/01/28 03:01:40 UTC
[1/5] accumulo git commit: ACCUMULO-4066 Speed up condition checking
for conditional mutations
Repository: accumulo
Updated Branches:
refs/heads/master 97ce2a758 -> 4eadac16b
ACCUMULO-4066 Speed up condition checking for conditional mutations
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/21d2f615
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/21d2f615
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/21d2f615
Branch: refs/heads/master
Commit: 21d2f61524de1f26ed88c4baaddee70c01081196
Parents: ff1e003
Author: Keith Turner <kt...@apache.org>
Authored: Wed Jan 27 14:13:40 2016 -0500
Committer: Keith Turner <kt...@apache.org>
Committed: Wed Jan 27 19:32:29 2016 -0500
----------------------------------------------------------------------
.../core/client/impl/CompressedIterators.java | 14 +-
.../core/client/impl/ConditionalWriterImpl.java | 44 ++++-
.../accumulo/core/iterators/IteratorUtil.java | 93 ++++++-----
.../client/impl/ConditionalComparatorTest.java | 53 ++++++
.../tserver/ConditionCheckerContext.java | 164 +++++++++++++++++++
.../org/apache/accumulo/tserver/Tablet.java | 37 ++++-
.../apache/accumulo/tserver/TabletServer.java | 95 ++++-------
.../accumulo/test/ConditionalWriterIT.java | 137 +++++++++++++++-
8 files changed, 521 insertions(+), 116 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/21d2f615/core/src/main/java/org/apache/accumulo/core/client/impl/CompressedIterators.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/impl/CompressedIterators.java b/core/src/main/java/org/apache/accumulo/core/client/impl/CompressedIterators.java
index 3fcce90..96d58a7 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/CompressedIterators.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/CompressedIterators.java
@@ -24,15 +24,12 @@ import java.util.Map;
import java.util.Map.Entry;
import org.apache.accumulo.core.client.IteratorSetting;
-import org.apache.accumulo.core.data.ArrayByteSequence;
-import org.apache.accumulo.core.data.ByteSequence;
import org.apache.accumulo.core.data.thrift.IterInfo;
import org.apache.accumulo.core.util.UnsynchronizedBuffer;
public class CompressedIterators {
private Map<String,Integer> symbolMap;
private List<String> symbolTable;
- private Map<ByteSequence,IterConfig> cache;
public static class IterConfig {
public List<IterInfo> ssiList = new ArrayList<IterInfo>();
@@ -46,7 +43,6 @@ public class CompressedIterators {
public CompressedIterators(List<String> symbols) {
this.symbolTable = symbols;
- this.cache = new HashMap<ByteSequence,IterConfig>();
}
private int getSymbolID(String symbol) {
@@ -85,14 +81,7 @@ public class CompressedIterators {
}
public IterConfig decompress(ByteBuffer iterators) {
-
- ByteSequence iterKey = new ArrayByteSequence(iterators);
- IterConfig config = cache.get(iterKey);
- if (config != null) {
- return config;
- }
-
- config = new IterConfig();
+ IterConfig config = new IterConfig();
UnsynchronizedBuffer.Reader in = new UnsynchronizedBuffer.Reader(iterators);
@@ -120,7 +109,6 @@ public class CompressedIterators {
}
- cache.put(iterKey, config);
return config;
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/21d2f615/core/src/main/java/org/apache/accumulo/core/client/impl/ConditionalWriterImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/impl/ConditionalWriterImpl.java b/core/src/main/java/org/apache/accumulo/core/client/impl/ConditionalWriterImpl.java
index 730cf08..9030d77 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/ConditionalWriterImpl.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/ConditionalWriterImpl.java
@@ -19,7 +19,9 @@ package org.apache.accumulo.core.client.impl;
import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@@ -70,6 +72,7 @@ import org.apache.accumulo.core.tabletserver.thrift.TabletClientService;
import org.apache.accumulo.core.util.BadArgumentException;
import org.apache.accumulo.core.util.ByteBufferUtil;
import org.apache.accumulo.core.util.LoggingRunnable;
+import org.apache.accumulo.core.util.NamingThreadFactory;
import org.apache.accumulo.core.util.ThriftUtil;
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.accumulo.core.zookeeper.ZooUtil;
@@ -375,7 +378,7 @@ class ConditionalWriterImpl implements ConditionalWriter {
this.credentials = credentials;
this.auths = config.getAuthorizations();
this.ve = new VisibilityEvaluator(config.getAuthorizations());
- this.threadPool = new ScheduledThreadPoolExecutor(config.getMaxWriteThreads());
+ this.threadPool = new ScheduledThreadPoolExecutor(config.getMaxWriteThreads(), new NamingThreadFactory(this.getClass().getSimpleName()));
this.locator = TabletLocator.getLocator(instance, new Text(tableId));
this.serverQueues = new HashMap<String,ServerQueue>();
this.tableId = tableId;
@@ -741,10 +744,47 @@ class ConditionalWriterImpl implements ConditionalWriter {
}
}
+ static class ConditionComparator implements Comparator<Condition> {
+
+ private static final Long MAX = Long.valueOf(Long.MAX_VALUE);
+
+ @Override
+ public int compare(Condition c1, Condition c2) {
+ int comp = c1.getFamily().compareTo(c2.getFamily());
+ if (comp == 0) {
+ comp = c1.getQualifier().compareTo(c2.getQualifier());
+ if (comp == 0) {
+ comp = c1.getVisibility().compareTo(c2.getVisibility());
+ if (comp == 0) {
+ Long l1 = c1.getTimestamp();
+ Long l2 = c2.getTimestamp();
+ if (l1 == null) {
+ l1 = MAX;
+ }
+
+ if (l2 == null) {
+ l2 = MAX;
+ }
+
+ comp = l2.compareTo(l1);
+ }
+ }
+ }
+
+ return comp;
+ }
+ }
+
+ private static final ConditionComparator CONDITION_COMPARATOR = new ConditionComparator();
+
private List<TCondition> convertConditions(ConditionalMutation cm, CompressedIterators compressedIters) {
List<TCondition> conditions = new ArrayList<TCondition>(cm.getConditions().size());
- for (Condition cond : cm.getConditions()) {
+ // sort conditions inorder to get better lookup performance. Sort on client side so tserver does not have to do it.
+ Condition[] ca = cm.getConditions().toArray(new Condition[cm.getConditions().size()]);
+ Arrays.sort(ca, CONDITION_COMPARATOR);
+
+ for (Condition cond : ca) {
long ts = 0;
boolean hasTs = false;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/21d2f615/core/src/main/java/org/apache/accumulo/core/iterators/IteratorUtil.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/iterators/IteratorUtil.java b/core/src/main/java/org/apache/accumulo/core/iterators/IteratorUtil.java
index 98392f6..6f76d77 100644
--- a/core/src/main/java/org/apache/accumulo/core/iterators/IteratorUtil.java
+++ b/core/src/main/java/org/apache/accumulo/core/iterators/IteratorUtil.java
@@ -26,6 +26,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
+import java.util.Set;
import java.util.TreeMap;
import org.apache.accumulo.core.client.IteratorSetting;
@@ -108,21 +109,26 @@ public class IteratorUtil {
return props;
}
- public static int getMaxPriority(IteratorScope scope, AccumuloConfiguration conf) {
- List<IterInfo> iters = new ArrayList<IterInfo>();
- parseIterConf(scope, iters, new HashMap<String,Map<String,String>>(), conf);
+ public static void mergeIteratorConfig(List<IterInfo> destList, Map<String,Map<String,String>> destOpts, List<IterInfo> tableIters,
+ Map<String,Map<String,String>> tableOpts, List<IterInfo> ssi, Map<String,Map<String,String>> ssio) {
+ destList.addAll(tableIters);
+ destList.addAll(ssi);
+ Collections.sort(destList, new IterInfoComparator());
- int max = 0;
-
- for (IterInfo iterInfo : iters) {
- if (iterInfo.priority > max)
- max = iterInfo.priority;
+ Set<Entry<String,Map<String,String>>> es = tableOpts.entrySet();
+ for (Entry<String,Map<String,String>> entry : es) {
+ if (entry.getValue() == null) {
+ destOpts.put(entry.getKey(), null);
+ } else {
+ destOpts.put(entry.getKey(), new HashMap<String,String>(entry.getValue()));
+ }
}
- return max;
+ IteratorUtil.mergeOptions(ssio, destOpts);
+
}
- protected static void parseIterConf(IteratorScope scope, List<IterInfo> iters, Map<String,Map<String,String>> allOptions, AccumuloConfiguration conf) {
+ public static void parseIterConf(IteratorScope scope, List<IterInfo> iters, Map<String,Map<String,String>> allOptions, AccumuloConfiguration conf) {
final Property scopeProperty = IteratorScope.getProperty(scope);
final String scopePropertyKey = scopeProperty.getKey();
@@ -155,24 +161,6 @@ public class IteratorUtil {
Collections.sort(iters, new IterInfoComparator());
}
- public static String findIterator(IteratorScope scope, String className, AccumuloConfiguration conf, Map<String,String> opts) {
- ArrayList<IterInfo> iters = new ArrayList<IterInfo>();
- Map<String,Map<String,String>> allOptions = new HashMap<String,Map<String,String>>();
-
- parseIterConf(scope, iters, allOptions, conf);
-
- for (IterInfo iterInfo : iters)
- if (iterInfo.className.equals(className)) {
- Map<String,String> tmpOpts = allOptions.get(iterInfo.iterName);
- if (tmpOpts != null) {
- opts.putAll(tmpOpts);
- }
- return iterInfo.iterName;
- }
-
- return null;
- }
-
public static <K extends WritableComparable<?>,V extends Writable> SortedKeyValueIterator<K,V> loadIterators(IteratorScope scope,
SortedKeyValueIterator<K,V> source, KeyExtent extent, AccumuloConfiguration conf, IteratorEnvironment env) throws IOException {
List<IterInfo> emptyList = Collections.emptyList();
@@ -209,6 +197,12 @@ public class IteratorUtil {
parseIterConf(scope, iters, allOptions, conf);
+ mergeOptions(ssio, allOptions);
+
+ return loadIterators(source, iters, allOptions, env, useAccumuloClassLoader, conf.get(Property.TABLE_CLASSPATH));
+ }
+
+ private static void mergeOptions(Map<String,Map<String,String>> ssio, Map<String,Map<String,String>> allOptions) {
for (Entry<String,Map<String,String>> entry : ssio.entrySet()) {
if (entry.getValue() == null)
continue;
@@ -219,30 +213,35 @@ public class IteratorUtil {
options.putAll(entry.getValue());
}
}
-
- return loadIterators(source, iters, allOptions, env, useAccumuloClassLoader, conf.get(Property.TABLE_CLASSPATH));
}
- @SuppressWarnings("unchecked")
public static <K extends WritableComparable<?>,V extends Writable> SortedKeyValueIterator<K,V> loadIterators(SortedKeyValueIterator<K,V> source,
Collection<IterInfo> iters, Map<String,Map<String,String>> iterOpts, IteratorEnvironment env, boolean useAccumuloClassLoader, String context)
throws IOException {
+ return loadIterators(source, iters, iterOpts, env, useAccumuloClassLoader, context, null);
+ }
+
+ public static <K extends WritableComparable<?>,V extends Writable> SortedKeyValueIterator<K,V> loadIterators(SortedKeyValueIterator<K,V> source,
+ Collection<IterInfo> iters, Map<String,Map<String,String>> iterOpts, IteratorEnvironment env, boolean useAccumuloClassLoader, String context,
+ Map<String,Class<? extends SortedKeyValueIterator<K,V>>> classCache) throws IOException {
// wrap the source in a SynchronizedIterator in case any of the additional configured iterators want to use threading
SortedKeyValueIterator<K,V> prev = new SynchronizedIterator<K,V>(source);
try {
for (IterInfo iterInfo : iters) {
- Class<? extends SortedKeyValueIterator<K,V>> clazz;
- if (useAccumuloClassLoader) {
- if (context != null && !context.equals(""))
- clazz = (Class<? extends SortedKeyValueIterator<K,V>>) AccumuloVFSClassLoader.getContextManager().loadClass(context, iterInfo.className,
- SortedKeyValueIterator.class);
- else
- clazz = (Class<? extends SortedKeyValueIterator<K,V>>) AccumuloVFSClassLoader.loadClass(iterInfo.className, SortedKeyValueIterator.class);
+ Class<? extends SortedKeyValueIterator<K,V>> clazz = null;
+ if (classCache != null) {
+ clazz = classCache.get(iterInfo.className);
+
+ if (clazz == null) {
+ clazz = loadClass(useAccumuloClassLoader, context, iterInfo);
+ classCache.put(iterInfo.className, clazz);
+ }
} else {
- clazz = (Class<? extends SortedKeyValueIterator<K,V>>) Class.forName(iterInfo.className).asSubclass(SortedKeyValueIterator.class);
+ clazz = loadClass(useAccumuloClassLoader, context, iterInfo);
}
+
SortedKeyValueIterator<K,V> skvi = clazz.newInstance();
Map<String,String> options = iterOpts.get(iterInfo.iterName);
@@ -266,6 +265,22 @@ public class IteratorUtil {
return prev;
}
+ @SuppressWarnings("unchecked")
+ private static <K extends WritableComparable<?>,V extends Writable> Class<? extends SortedKeyValueIterator<K,V>> loadClass(boolean useAccumuloClassLoader,
+ String context, IterInfo iterInfo) throws ClassNotFoundException, IOException {
+ Class<? extends SortedKeyValueIterator<K,V>> clazz;
+ if (useAccumuloClassLoader) {
+ if (context != null && !context.equals(""))
+ clazz = (Class<? extends SortedKeyValueIterator<K,V>>) AccumuloVFSClassLoader.getContextManager().loadClass(context, iterInfo.className,
+ SortedKeyValueIterator.class);
+ else
+ clazz = (Class<? extends SortedKeyValueIterator<K,V>>) AccumuloVFSClassLoader.loadClass(iterInfo.className, SortedKeyValueIterator.class);
+ } else {
+ clazz = (Class<? extends SortedKeyValueIterator<K,V>>) Class.forName(iterInfo.className).asSubclass(SortedKeyValueIterator.class);
+ }
+ return clazz;
+ }
+
public static Range maximizeStartKeyTimeStamp(Range range) {
Range seekRange = range;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/21d2f615/core/src/test/java/org/apache/accumulo/core/client/impl/ConditionalComparatorTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/accumulo/core/client/impl/ConditionalComparatorTest.java b/core/src/test/java/org/apache/accumulo/core/client/impl/ConditionalComparatorTest.java
new file mode 100644
index 0000000..5413893
--- /dev/null
+++ b/core/src/test/java/org/apache/accumulo/core/client/impl/ConditionalComparatorTest.java
@@ -0,0 +1,53 @@
+/*
+ * 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.accumulo.core.client.impl;
+
+import org.apache.accumulo.core.client.impl.ConditionalWriterImpl.ConditionComparator;
+import org.apache.accumulo.core.data.Condition;
+import org.apache.accumulo.core.security.ColumnVisibility;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class ConditionalComparatorTest {
+ @Test
+ public void testComparator() {
+ Condition c1 = new Condition("a", "b");
+ Condition c2 = new Condition("a", "c");
+ Condition c3 = new Condition("b", "c");
+ Condition c4 = new Condition("a", "b").setTimestamp(5);
+ Condition c5 = new Condition("a", "b").setTimestamp(6);
+ Condition c6 = new Condition("a", "b").setVisibility(new ColumnVisibility("A&B"));
+ Condition c7 = new Condition("a", "b").setVisibility(new ColumnVisibility("A&C"));
+
+ ConditionComparator comparator = new ConditionComparator();
+
+ Assert.assertTrue(comparator.compare(c1, c1) == 0);
+ Assert.assertTrue(comparator.compare(c1, c2) < 0);
+ Assert.assertTrue(comparator.compare(c2, c1) > 0);
+ Assert.assertTrue(comparator.compare(c1, c3) < 0);
+ Assert.assertTrue(comparator.compare(c3, c1) > 0);
+ Assert.assertTrue(comparator.compare(c1, c4) < 0);
+ Assert.assertTrue(comparator.compare(c4, c1) > 0);
+ Assert.assertTrue(comparator.compare(c5, c4) < 0);
+ Assert.assertTrue(comparator.compare(c4, c5) > 0);
+ Assert.assertTrue(comparator.compare(c1, c7) < 0);
+ Assert.assertTrue(comparator.compare(c7, c1) > 0);
+ Assert.assertTrue(comparator.compare(c6, c7) < 0);
+ Assert.assertTrue(comparator.compare(c7, c6) > 0);
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/21d2f615/server/tserver/src/main/java/org/apache/accumulo/tserver/ConditionCheckerContext.java
----------------------------------------------------------------------
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/ConditionCheckerContext.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/ConditionCheckerContext.java
new file mode 100644
index 0000000..2e34f38
--- /dev/null
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/ConditionCheckerContext.java
@@ -0,0 +1,164 @@
+/*
+ * 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.accumulo.tserver;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.accumulo.core.client.impl.CompressedIterators;
+import org.apache.accumulo.core.client.impl.CompressedIterators.IterConfig;
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.data.ArrayByteSequence;
+import org.apache.accumulo.core.data.ByteSequence;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.data.thrift.IterInfo;
+import org.apache.accumulo.core.data.thrift.TCMResult;
+import org.apache.accumulo.core.data.thrift.TCMStatus;
+import org.apache.accumulo.core.data.thrift.TCondition;
+import org.apache.accumulo.core.iterators.IteratorUtil;
+import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
+import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
+import org.apache.accumulo.tserver.data.ServerConditionalMutation;
+import org.apache.hadoop.io.Text;
+
+import com.google.common.base.Preconditions;
+
+public class ConditionCheckerContext {
+ private CompressedIterators compressedIters;
+
+ private List<IterInfo> tableIters;
+ private Map<String,Map<String,String>> tableIterOpts;
+ private TabletIteratorEnvironment tie;
+ private String context;
+ private Map<String,Class<? extends SortedKeyValueIterator<Key,Value>>> classCache;
+
+ private static class MergedIterConfig {
+ List<IterInfo> mergedIters;
+ Map<String,Map<String,String>> mergedItersOpts;
+
+ MergedIterConfig(List<IterInfo> mergedIters, Map<String,Map<String,String>> mergedItersOpts) {
+ this.mergedIters = mergedIters;
+ this.mergedItersOpts = mergedItersOpts;
+ }
+ }
+
+ private Map<ByteSequence,MergedIterConfig> mergedIterCache = new HashMap<ByteSequence,MergedIterConfig>();
+
+ ConditionCheckerContext(CompressedIterators compressedIters, AccumuloConfiguration tableConf) {
+ this.compressedIters = compressedIters;
+
+ tableIters = new ArrayList<IterInfo>();
+ tableIterOpts = new HashMap<String,Map<String,String>>();
+
+ // parse table iterator config once
+ IteratorUtil.parseIterConf(IteratorScope.scan, tableIters, tableIterOpts, tableConf);
+
+ context = tableConf.get(Property.TABLE_CLASSPATH);
+
+ classCache = new HashMap<String,Class<? extends SortedKeyValueIterator<Key,Value>>>();
+
+ tie = new TabletIteratorEnvironment(IteratorScope.scan, tableConf);
+ }
+
+ SortedKeyValueIterator<Key,Value> buildIterator(SortedKeyValueIterator<Key,Value> systemIter, TCondition tc) throws IOException {
+
+ ArrayByteSequence key = new ArrayByteSequence(tc.iterators);
+ MergedIterConfig mic = mergedIterCache.get(key);
+ if (mic == null) {
+ IterConfig ic = compressedIters.decompress(tc.iterators);
+
+ List<IterInfo> mergedIters = new ArrayList<IterInfo>(tableIters.size() + ic.ssiList.size());
+ Map<String,Map<String,String>> mergedItersOpts = new HashMap<String,Map<String,String>>(tableIterOpts.size() + ic.ssio.size());
+
+ IteratorUtil.mergeIteratorConfig(mergedIters, mergedItersOpts, tableIters, tableIterOpts, ic.ssiList, ic.ssio);
+
+ mic = new MergedIterConfig(mergedIters, mergedItersOpts);
+
+ mergedIterCache.put(key, mic);
+ }
+
+ return IteratorUtil.loadIterators(systemIter, mic.mergedIters, mic.mergedItersOpts, tie, true, context, classCache);
+ }
+
+ boolean checkConditions(SortedKeyValueIterator<Key,Value> systemIter, ServerConditionalMutation scm) throws IOException {
+ boolean add = true;
+
+ for (TCondition tc : scm.getConditions()) {
+
+ Range range;
+ if (tc.hasTimestamp)
+ range = Range.exact(new Text(scm.getRow()), new Text(tc.getCf()), new Text(tc.getCq()), new Text(tc.getCv()), tc.getTs());
+ else
+ range = Range.exact(new Text(scm.getRow()), new Text(tc.getCf()), new Text(tc.getCq()), new Text(tc.getCv()));
+
+ SortedKeyValueIterator<Key,Value> iter = buildIterator(systemIter, tc);
+
+ ByteSequence cf = new ArrayByteSequence(tc.getCf());
+ iter.seek(range, Collections.singleton(cf), true);
+ Value val = null;
+ if (iter.hasTop()) {
+ val = iter.getTopValue();
+ }
+
+ if ((val == null ^ tc.getVal() == null) || (val != null && !Arrays.equals(tc.getVal(), val.get()))) {
+ add = false;
+ break;
+ }
+ }
+ return add;
+ }
+
+ public class ConditionChecker {
+
+ private List<ServerConditionalMutation> conditionsToCheck;
+ private List<ServerConditionalMutation> okMutations;
+ private List<TCMResult> results;
+ private boolean checked = false;
+
+ public ConditionChecker(List<ServerConditionalMutation> conditionsToCheck, List<ServerConditionalMutation> okMutations, List<TCMResult> results) {
+ this.conditionsToCheck = conditionsToCheck;
+ this.okMutations = okMutations;
+ this.results = results;
+ }
+
+ public void check(SortedKeyValueIterator<Key,Value> systemIter) throws IOException {
+ Preconditions.checkArgument(!checked, "check() method should only be called once");
+ checked = true;
+
+ for (ServerConditionalMutation scm : conditionsToCheck) {
+ if (checkConditions(systemIter, scm)) {
+ okMutations.add(scm);
+ } else {
+ results.add(new TCMResult(scm.getID(), TCMStatus.REJECTED));
+ }
+ }
+ }
+ }
+
+ public ConditionChecker newChecker(List<ServerConditionalMutation> conditionsToCheck, List<ServerConditionalMutation> okMutations, List<TCMResult> results) {
+ return new ConditionChecker(conditionsToCheck, okMutations, results);
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/21d2f615/server/tserver/src/main/java/org/apache/accumulo/tserver/Tablet.java
----------------------------------------------------------------------
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/Tablet.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/Tablet.java
index 3f00c0b..ca4bde6 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/Tablet.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/Tablet.java
@@ -128,6 +128,7 @@ import org.apache.accumulo.trace.instrument.Span;
import org.apache.accumulo.trace.instrument.Trace;
import org.apache.accumulo.tserver.Compactor.CompactionCanceledException;
import org.apache.accumulo.tserver.Compactor.CompactionEnv;
+import org.apache.accumulo.tserver.ConditionCheckerContext.ConditionChecker;
import org.apache.accumulo.tserver.FileManager.ScanFileManager;
import org.apache.accumulo.tserver.InMemoryMap.MemoryIterator;
import org.apache.accumulo.tserver.TabletServer.TservConstraintEnv;
@@ -1597,6 +1598,23 @@ public class Tablet {
void receive(List<KVEntry> matches) throws IOException;
}
+ public void checkConditions(ConditionChecker checker, Authorizations authorizations, AtomicBoolean iFlag) throws IOException {
+
+ ScanDataSource dataSource = new ScanDataSource(this, authorizations, this.defaultSecurityLabel, iFlag);
+
+ try {
+ SortedKeyValueIterator<Key,Value> iter = new SourceSwitchingIterator(dataSource);
+ checker.check(iter);
+ } catch (IOException ioe) {
+ dataSource.close(true);
+ throw ioe;
+ } finally {
+ // code in finally block because always want
+ // to return mapfiles, even when exception is thrown
+ dataSource.close(false);
+ }
+ }
+
class LookupResult {
List<Range> unfinishedRanges = new ArrayList<Range>();
long bytesAdded = 0;
@@ -1929,18 +1947,29 @@ public class Tablet {
private StatsIterator statsIterator;
ScanOptions options;
+ private final boolean loadIters;
ScanDataSource(Authorizations authorizations, byte[] defaultLabels, HashSet<Column> columnSet, List<IterInfo> ssiList, Map<String,Map<String,String>> ssio,
AtomicBoolean interruptFlag) {
expectedDeletionCount = dataSourceDeletions.get();
this.options = new ScanOptions(-1, authorizations, defaultLabels, columnSet, ssiList, ssio, interruptFlag, false);
this.interruptFlag = interruptFlag;
+ this.loadIters = true;
}
ScanDataSource(ScanOptions options) {
expectedDeletionCount = dataSourceDeletions.get();
this.options = options;
this.interruptFlag = options.interruptFlag;
+ this.loadIters = true;
+ }
+
+ ScanDataSource(Tablet tablet, Authorizations authorizations, byte[] defaultLabels, AtomicBoolean iFlag) {
+ expectedDeletionCount = dataSourceDeletions.get();
+ Set<Column> emptycols = Collections.emptySet();
+ this.options = new ScanOptions(-1, authorizations, defaultLabels, emptycols, null, null, iFlag, false);
+ this.interruptFlag = iFlag;
+ this.loadIters = false;
}
@Override
@@ -2035,8 +2064,12 @@ public class Tablet {
VisibilityFilter visFilter = new VisibilityFilter(colFilter, options.authorizations, options.defaultLabels);
- return iterEnv.getTopLevelIterator(IteratorUtil
- .loadIterators(IteratorScope.scan, visFilter, extent, acuTableConf, options.ssiList, options.ssio, iterEnv));
+ if (!loadIters) {
+ return visFilter;
+ } else {
+ return iterEnv.getTopLevelIterator(IteratorUtil.loadIterators(IteratorScope.scan, visFilter, extent, acuTableConf, options.ssiList, options.ssio,
+ iterEnv));
+ }
}
private void close(boolean sawErrors) {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/21d2f615/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
----------------------------------------------------------------------
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
index 4e3e00b..6023ae3 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
@@ -67,7 +67,6 @@ import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.impl.CompressedIterators;
-import org.apache.accumulo.core.client.impl.CompressedIterators.IterConfig;
import org.apache.accumulo.core.client.impl.ScannerImpl;
import org.apache.accumulo.core.client.impl.Tables;
import org.apache.accumulo.core.client.impl.TabletLocator;
@@ -99,7 +98,6 @@ import org.apache.accumulo.core.data.thrift.ScanResult;
import org.apache.accumulo.core.data.thrift.TCMResult;
import org.apache.accumulo.core.data.thrift.TCMStatus;
import org.apache.accumulo.core.data.thrift.TColumn;
-import org.apache.accumulo.core.data.thrift.TCondition;
import org.apache.accumulo.core.data.thrift.TConditionalMutation;
import org.apache.accumulo.core.data.thrift.TConditionalSession;
import org.apache.accumulo.core.data.thrift.TKey;
@@ -199,6 +197,7 @@ import org.apache.accumulo.trace.instrument.Span;
import org.apache.accumulo.trace.instrument.Trace;
import org.apache.accumulo.trace.thrift.TInfo;
import org.apache.accumulo.tserver.Compactor.CompactionInfo;
+import org.apache.accumulo.tserver.ConditionCheckerContext.ConditionChecker;
import org.apache.accumulo.tserver.RowLocks.RowLock;
import org.apache.accumulo.tserver.Tablet.CommitSession;
import org.apache.accumulo.tserver.Tablet.KVEntry;
@@ -1930,79 +1929,57 @@ public class TabletServer extends AbstractMetricsImpl implements org.apache.accu
List<String> symbols) throws IOException {
Iterator<Entry<KeyExtent,List<ServerConditionalMutation>>> iter = updates.entrySet().iterator();
- CompressedIterators compressedIters = new CompressedIterators(symbols);
+ final CompressedIterators compressedIters = new CompressedIterators(symbols);
+ ConditionCheckerContext checkerContext = new ConditionCheckerContext(compressedIters, ServerConfiguration.getTableConfiguration(instance, cs.tableId));
while (iter.hasNext()) {
- Entry<KeyExtent,List<ServerConditionalMutation>> entry = iter.next();
- Tablet tablet = onlineTablets.get(entry.getKey());
+ final Entry<KeyExtent,List<ServerConditionalMutation>> entry = iter.next();
+ final Tablet tablet = onlineTablets.get(entry.getKey());
if (tablet == null || tablet.isClosed()) {
for (ServerConditionalMutation scm : entry.getValue())
results.add(new TCMResult(scm.getID(), TCMStatus.IGNORED));
iter.remove();
} else {
- List<ServerConditionalMutation> okMutations = new ArrayList<ServerConditionalMutation>(entry.getValue().size());
+ final List<ServerConditionalMutation> okMutations = new ArrayList<ServerConditionalMutation>(entry.getValue().size());
+ final List<TCMResult> resultsSubList = results.subList(results.size(), results.size());
- for (ServerConditionalMutation scm : entry.getValue()) {
- if (checkCondition(results, cs, compressedIters, tablet, scm))
- okMutations.add(scm);
- }
-
- entry.setValue(okMutations);
- }
-
- }
- }
-
- boolean checkCondition(ArrayList<TCMResult> results, ConditionalSession cs, CompressedIterators compressedIters, Tablet tablet,
- ServerConditionalMutation scm) throws IOException {
- boolean add = true;
-
- Set<Column> emptyCols = Collections.emptySet();
-
- for (TCondition tc : scm.getConditions()) {
-
- Range range;
- if (tc.hasTimestamp)
- range = Range.exact(new Text(scm.getRow()), new Text(tc.getCf()), new Text(tc.getCq()), new Text(tc.getCv()), tc.getTs());
- else
- range = Range.exact(new Text(scm.getRow()), new Text(tc.getCf()), new Text(tc.getCq()), new Text(tc.getCv()));
-
- IterConfig ic = compressedIters.decompress(tc.iterators);
-
- Scanner scanner = tablet.createScanner(range, 1, emptyCols, cs.auths, ic.ssiList, ic.ssio, false, cs.interruptFlag);
+ ConditionChecker checker = checkerContext.newChecker(entry.getValue(), okMutations, resultsSubList);
+ try {
+ tablet.checkConditions(checker, cs.auths, cs.interruptFlag);
- try {
- ScanBatch batch = scanner.read();
+ if (okMutations.size() > 0) {
+ entry.setValue(okMutations);
+ } else {
+ iter.remove();
+ }
+ } catch (TabletClosedException e) {
+ // clear anything added while checking conditions.
+ resultsSubList.clear();
- Value val = null;
+ for (ServerConditionalMutation scm : entry.getValue()) {
+ results.add(new TCMResult(scm.getID(), TCMStatus.IGNORED));
+ }
+ iter.remove();
+ } catch (IterationInterruptedException e) {
+ // clear anything added while checking conditions.
+ resultsSubList.clear();
- for (KVEntry entry2 : batch.results) {
- val = entry2.getValue();
- break;
- }
+ for (ServerConditionalMutation scm : entry.getValue()) {
+ results.add(new TCMResult(scm.getID(), TCMStatus.IGNORED));
+ }
+ iter.remove();
+ } catch (TooManyFilesException e) {
+ // clear anything added while checking conditions.
+ resultsSubList.clear();
- if ((val == null ^ tc.getVal() == null) || (val != null && !Arrays.equals(tc.getVal(), val.get()))) {
- results.add(new TCMResult(scm.getID(), TCMStatus.REJECTED));
- add = false;
- break;
+ for (ServerConditionalMutation scm : entry.getValue()) {
+ results.add(new TCMResult(scm.getID(), TCMStatus.IGNORED));
+ }
+ iter.remove();
}
-
- } catch (TabletClosedException e) {
- results.add(new TCMResult(scm.getID(), TCMStatus.IGNORED));
- add = false;
- break;
- } catch (IterationInterruptedException iie) {
- results.add(new TCMResult(scm.getID(), TCMStatus.IGNORED));
- add = false;
- break;
- } catch (TooManyFilesException tmfe) {
- results.add(new TCMResult(scm.getID(), TCMStatus.IGNORED));
- add = false;
- break;
}
}
- return add;
}
private void writeConditionalMutations(Map<KeyExtent,List<ServerConditionalMutation>> updates, ArrayList<TCMResult> results, ConditionalSession sess) {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/21d2f615/test/src/test/java/org/apache/accumulo/test/ConditionalWriterIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/ConditionalWriterIT.java b/test/src/test/java/org/apache/accumulo/test/ConditionalWriterIT.java
index 5bfaebf..d32bc43 100644
--- a/test/src/test/java/org/apache/accumulo/test/ConditionalWriterIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/ConditionalWriterIT.java
@@ -19,9 +19,11 @@ package org.apache.accumulo.test;
import static org.junit.Assert.assertTrue;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@@ -63,7 +65,11 @@ import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.iterators.IteratorEnvironment;
+import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
import org.apache.accumulo.core.iterators.LongCombiner.Type;
+import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
+import org.apache.accumulo.core.iterators.WrappingIterator;
import org.apache.accumulo.core.iterators.user.SummingCombiner;
import org.apache.accumulo.core.iterators.user.VersioningIterator;
import org.apache.accumulo.core.security.Authorizations;
@@ -90,6 +96,7 @@ import org.junit.Assert;
import org.junit.Assume;
import org.junit.Test;
+import com.google.common.base.Charsets;
import com.google.common.collect.Iterables;
/**
@@ -501,7 +508,135 @@ public class ConditionalWriterIT extends AccumuloClusterIT {
Assert.assertEquals(expected, actual);
- // TODO test w/ table that has iterators configured
+ cw.close();
+ }
+
+ public static class AddingIterator extends WrappingIterator {
+ long amount = 0;
+
+ @Override
+ public Value getTopValue() {
+ Value val = super.getTopValue();
+ long l = Long.parseLong(val.toString());
+ String newVal = (l + amount) + "";
+ return new Value(newVal.getBytes(Charsets.UTF_8));
+ }
+
+ @Override
+ public void init(SortedKeyValueIterator<Key,Value> source, Map<String,String> options, IteratorEnvironment env) throws IOException {
+ this.setSource(source);
+ amount = Long.parseLong(options.get("amount"));
+ }
+ }
+
+ public static class MultiplyingIterator extends WrappingIterator {
+ long amount = 0;
+
+ @Override
+ public Value getTopValue() {
+ Value val = super.getTopValue();
+ long l = Long.parseLong(val.toString());
+ String newVal = l * amount + "";
+ return new Value(newVal.getBytes(Charsets.UTF_8));
+ }
+
+ @Override
+ public void init(SortedKeyValueIterator<Key,Value> source, Map<String,String> options, IteratorEnvironment env) throws IOException {
+ this.setSource(source);
+ amount = Long.parseLong(options.get("amount"));
+ }
+ }
+
+ @Test
+ public void testTableAndConditionIterators() throws Exception {
+
+ // test w/ table that has iterators configured
+ Connector conn = getConnector();
+ String tableName = getUniqueNames(1)[0];
+
+ IteratorSetting aiConfig1 = new IteratorSetting(30, "AI1", AddingIterator.class);
+ aiConfig1.addOption("amount", "2");
+ IteratorSetting aiConfig2 = new IteratorSetting(35, "MI1", MultiplyingIterator.class);
+ aiConfig2.addOption("amount", "3");
+ IteratorSetting aiConfig3 = new IteratorSetting(40, "AI2", AddingIterator.class);
+ aiConfig3.addOption("amount", "5");
+
+ conn.tableOperations().create(tableName);
+
+ BatchWriter bw = conn.createBatchWriter(tableName, new BatchWriterConfig());
+
+ Mutation m = new Mutation("ACCUMULO-1000");
+ m.put("count", "comments", "6");
+ bw.addMutation(m);
+
+ m = new Mutation("ACCUMULO-1001");
+ m.put("count", "comments", "7");
+ bw.addMutation(m);
+
+ m = new Mutation("ACCUMULO-1002");
+ m.put("count", "comments", "8");
+ bw.addMutation(m);
+
+ bw.close();
+
+ conn.tableOperations().attachIterator(tableName, aiConfig1, EnumSet.of(IteratorScope.scan));
+ conn.tableOperations().offline(tableName, true);
+ conn.tableOperations().online(tableName, true);
+
+ ConditionalWriter cw = conn.createConditionalWriter(tableName, new ConditionalWriterConfig());
+
+ ConditionalMutation cm6 = new ConditionalMutation("ACCUMULO-1000", new Condition("count", "comments").setValue("8"));
+ cm6.put("count", "comments", "7");
+ Assert.assertEquals(Status.ACCEPTED, cw.write(cm6).getStatus());
+
+ Scanner scanner = conn.createScanner(tableName, new Authorizations());
+ scanner.setRange(new Range("ACCUMULO-1000"));
+ scanner.fetchColumn(new Text("count"), new Text("comments"));
+
+ Entry<Key,Value> entry = Iterables.getOnlyElement(scanner);
+ Assert.assertEquals("9", entry.getValue().toString());
+
+ ConditionalMutation cm7 = new ConditionalMutation("ACCUMULO-1000", new Condition("count", "comments").setIterators(aiConfig2).setValue("27"));
+ cm7.put("count", "comments", "8");
+ Assert.assertEquals(Status.ACCEPTED, cw.write(cm7).getStatus());
+
+ entry = Iterables.getOnlyElement(scanner);
+ Assert.assertEquals("10", entry.getValue().toString());
+
+ ConditionalMutation cm8 = new ConditionalMutation("ACCUMULO-1000", new Condition("count", "comments").setIterators(aiConfig2, aiConfig3).setValue("35"));
+ cm8.put("count", "comments", "9");
+ Assert.assertEquals(Status.ACCEPTED, cw.write(cm8).getStatus());
+
+ entry = Iterables.getOnlyElement(scanner);
+ Assert.assertEquals("11", entry.getValue().toString());
+
+ ConditionalMutation cm3 = new ConditionalMutation("ACCUMULO-1000", new Condition("count", "comments").setIterators(aiConfig2).setValue("33"));
+ cm3.put("count", "comments", "3");
+
+ ConditionalMutation cm4 = new ConditionalMutation("ACCUMULO-1001", new Condition("count", "comments").setIterators(aiConfig3).setValue("14"));
+ cm4.put("count", "comments", "3");
+
+ ConditionalMutation cm5 = new ConditionalMutation("ACCUMULO-1002", new Condition("count", "comments").setIterators(aiConfig3).setValue("10"));
+ cm5.put("count", "comments", "3");
+
+ Iterator<Result> results = cw.write(Arrays.asList(cm3, cm4, cm5).iterator());
+ Map<String,Status> actual = new HashMap<String,Status>();
+
+ while (results.hasNext()) {
+ Result result = results.next();
+ String k = new String(result.getMutation().getRow());
+ Assert.assertFalse("Did not expect to see multiple resultus for the row: " + k, actual.containsKey(k));
+ actual.put(k, result.getStatus());
+ }
+
+ cw.close();
+
+ Map<String,Status> expected = new HashMap<String,Status>();
+ expected.put("ACCUMULO-1000", Status.ACCEPTED);
+ expected.put("ACCUMULO-1001", Status.ACCEPTED);
+ expected.put("ACCUMULO-1002", Status.REJECTED);
+
+ Assert.assertEquals(expected, actual);
cw.close();
}
[4/5] accumulo git commit: Merge branch '1.6' into 1.7
Posted by kt...@apache.org.
Merge branch '1.6' into 1.7
Conflicts:
core/src/main/java/org/apache/accumulo/core/client/impl/ConditionalWriterImpl.java
core/src/main/java/org/apache/accumulo/core/iterators/IteratorUtil.java
server/tserver/src/main/java/org/apache/accumulo/tserver/Tablet.java
server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/d4882a15
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/d4882a15
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/d4882a15
Branch: refs/heads/master
Commit: d4882a15fd85482855783e3a56babc14d31bb5bb
Parents: 0faf8b9 21d2f61
Author: Keith Turner <kt...@apache.org>
Authored: Wed Jan 27 19:57:22 2016 -0500
Committer: Keith Turner <kt...@apache.org>
Committed: Wed Jan 27 19:57:22 2016 -0500
----------------------------------------------------------------------
.../core/client/impl/CompressedIterators.java | 14 +-
.../core/client/impl/ConditionalWriterImpl.java | 44 ++++-
.../accumulo/core/iterators/IteratorUtil.java | 93 ++++++-----
.../client/impl/ConditionalComparatorTest.java | 53 ++++++
.../tserver/ConditionCheckerContext.java | 164 +++++++++++++++++++
.../apache/accumulo/tserver/TabletServer.java | 76 ++-------
.../accumulo/tserver/tablet/ScanDataSource.java | 23 ++-
.../apache/accumulo/tserver/tablet/Tablet.java | 18 ++
.../accumulo/test/ConditionalWriterIT.java | 137 +++++++++++++++-
9 files changed, 507 insertions(+), 115 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/d4882a15/core/src/main/java/org/apache/accumulo/core/client/impl/ConditionalWriterImpl.java
----------------------------------------------------------------------
diff --cc core/src/main/java/org/apache/accumulo/core/client/impl/ConditionalWriterImpl.java
index 24040e6,9030d77..c7756ad
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/ConditionalWriterImpl.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/ConditionalWriterImpl.java
@@@ -17,11 -17,11 +17,13 @@@
package org.apache.accumulo.core.client.impl;
+import static java.nio.charset.StandardCharsets.UTF_8;
+
import java.nio.ByteBuffer;
import java.util.ArrayList;
+ import java.util.Arrays;
import java.util.Collections;
+ import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@@@ -69,14 -69,13 +71,15 @@@ import org.apache.accumulo.core.securit
import org.apache.accumulo.core.security.VisibilityParseException;
import org.apache.accumulo.core.tabletserver.thrift.NoSuchScanIDException;
import org.apache.accumulo.core.tabletserver.thrift.TabletClientService;
+import org.apache.accumulo.core.trace.Trace;
+import org.apache.accumulo.core.trace.Tracer;
+import org.apache.accumulo.core.trace.thrift.TInfo;
import org.apache.accumulo.core.util.BadArgumentException;
import org.apache.accumulo.core.util.ByteBufferUtil;
-import org.apache.accumulo.core.util.LoggingRunnable;
+ import org.apache.accumulo.core.util.NamingThreadFactory;
-import org.apache.accumulo.core.util.ThriftUtil;
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.accumulo.core.zookeeper.ZooUtil;
+import org.apache.accumulo.fate.util.LoggingRunnable;
import org.apache.accumulo.fate.zookeeper.ZooCacheFactory;
import org.apache.accumulo.fate.zookeeper.ZooLock;
import org.apache.accumulo.fate.zookeeper.ZooUtil.LockID;
@@@ -379,12 -373,13 +382,12 @@@ class ConditionalWriterImpl implements
}
}
- ConditionalWriterImpl(Instance instance, Credentials credentials, String tableId, ConditionalWriterConfig config) {
- this.instance = instance;
- this.credentials = credentials;
+ ConditionalWriterImpl(ClientContext context, String tableId, ConditionalWriterConfig config) {
+ this.context = context;
this.auths = config.getAuthorizations();
this.ve = new VisibilityEvaluator(config.getAuthorizations());
- this.threadPool = new ScheduledThreadPoolExecutor(config.getMaxWriteThreads());
+ this.threadPool = new ScheduledThreadPoolExecutor(config.getMaxWriteThreads(), new NamingThreadFactory(this.getClass().getSimpleName()));
- this.locator = TabletLocator.getLocator(instance, new Text(tableId));
+ this.locator = TabletLocator.getLocator(context, new Text(tableId));
this.serverQueues = new HashMap<String,ServerQueue>();
this.tableId = tableId;
this.timeout = config.getTimeout(TimeUnit.MILLISECONDS);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/d4882a15/core/src/main/java/org/apache/accumulo/core/iterators/IteratorUtil.java
----------------------------------------------------------------------
diff --cc core/src/main/java/org/apache/accumulo/core/iterators/IteratorUtil.java
index 031d13f,6f76d77..c739e56
--- a/core/src/main/java/org/apache/accumulo/core/iterators/IteratorUtil.java
+++ b/core/src/main/java/org/apache/accumulo/core/iterators/IteratorUtil.java
@@@ -124,22 -109,27 +125,27 @@@ public class IteratorUtil
return props;
}
- public static int getMaxPriority(IteratorScope scope, AccumuloConfiguration conf) {
- List<IterInfo> iters = new ArrayList<IterInfo>();
- parseIterConf(scope, iters, new HashMap<String,Map<String,String>>(), conf);
+ public static void mergeIteratorConfig(List<IterInfo> destList, Map<String,Map<String,String>> destOpts, List<IterInfo> tableIters,
+ Map<String,Map<String,String>> tableOpts, List<IterInfo> ssi, Map<String,Map<String,String>> ssio) {
+ destList.addAll(tableIters);
+ destList.addAll(ssi);
+ Collections.sort(destList, new IterInfoComparator());
- int max = 0;
-
- for (IterInfo iterInfo : iters) {
- if (iterInfo.priority > max)
- max = iterInfo.priority;
+ Set<Entry<String,Map<String,String>>> es = tableOpts.entrySet();
+ for (Entry<String,Map<String,String>> entry : es) {
+ if (entry.getValue() == null) {
+ destOpts.put(entry.getKey(), null);
+ } else {
- destOpts.put(entry.getKey(), new HashMap<String,String>(entry.getValue()));
++ destOpts.put(entry.getKey(), new HashMap<>(entry.getValue()));
+ }
}
- return max;
+ IteratorUtil.mergeOptions(ssio, destOpts);
+
}
- protected static void parseIterConf(IteratorScope scope, List<IterInfo> iters, Map<String,Map<String,String>> allOptions, AccumuloConfiguration conf) {
+ public static void parseIterConf(IteratorScope scope, List<IterInfo> iters, Map<String,Map<String,String>> allOptions, AccumuloConfiguration conf) {
- final Property scopeProperty = IteratorScope.getProperty(scope);
+ final Property scopeProperty = getProperty(scope);
final String scopePropertyKey = scopeProperty.getKey();
for (Entry<String,String> entry : conf.getAllPropertiesWithPrefix(scopeProperty).entrySet()) {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/d4882a15/server/tserver/src/main/java/org/apache/accumulo/tserver/ConditionCheckerContext.java
----------------------------------------------------------------------
diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/ConditionCheckerContext.java
index 0000000,2e34f38..39aa684
mode 000000,100644..100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/ConditionCheckerContext.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/ConditionCheckerContext.java
@@@ -1,0 -1,164 +1,164 @@@
+ /*
+ * 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.accumulo.tserver;
+
+ import java.io.IOException;
+ import java.util.ArrayList;
+ import java.util.Arrays;
+ import java.util.Collections;
+ import java.util.HashMap;
+ import java.util.List;
+ import java.util.Map;
+
+ import org.apache.accumulo.core.client.impl.CompressedIterators;
+ import org.apache.accumulo.core.client.impl.CompressedIterators.IterConfig;
+ import org.apache.accumulo.core.conf.AccumuloConfiguration;
+ import org.apache.accumulo.core.conf.Property;
+ import org.apache.accumulo.core.data.ArrayByteSequence;
+ import org.apache.accumulo.core.data.ByteSequence;
+ import org.apache.accumulo.core.data.Key;
+ import org.apache.accumulo.core.data.Range;
+ import org.apache.accumulo.core.data.Value;
+ import org.apache.accumulo.core.data.thrift.IterInfo;
+ import org.apache.accumulo.core.data.thrift.TCMResult;
+ import org.apache.accumulo.core.data.thrift.TCMStatus;
+ import org.apache.accumulo.core.data.thrift.TCondition;
+ import org.apache.accumulo.core.iterators.IteratorUtil;
+ import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
+ import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
+ import org.apache.accumulo.tserver.data.ServerConditionalMutation;
+ import org.apache.hadoop.io.Text;
+
+ import com.google.common.base.Preconditions;
+
+ public class ConditionCheckerContext {
+ private CompressedIterators compressedIters;
+
+ private List<IterInfo> tableIters;
+ private Map<String,Map<String,String>> tableIterOpts;
+ private TabletIteratorEnvironment tie;
+ private String context;
+ private Map<String,Class<? extends SortedKeyValueIterator<Key,Value>>> classCache;
+
+ private static class MergedIterConfig {
+ List<IterInfo> mergedIters;
+ Map<String,Map<String,String>> mergedItersOpts;
+
+ MergedIterConfig(List<IterInfo> mergedIters, Map<String,Map<String,String>> mergedItersOpts) {
+ this.mergedIters = mergedIters;
+ this.mergedItersOpts = mergedItersOpts;
+ }
+ }
+
- private Map<ByteSequence,MergedIterConfig> mergedIterCache = new HashMap<ByteSequence,MergedIterConfig>();
++ private Map<ByteSequence,MergedIterConfig> mergedIterCache = new HashMap<>();
+
+ ConditionCheckerContext(CompressedIterators compressedIters, AccumuloConfiguration tableConf) {
+ this.compressedIters = compressedIters;
+
- tableIters = new ArrayList<IterInfo>();
- tableIterOpts = new HashMap<String,Map<String,String>>();
++ tableIters = new ArrayList<>();
++ tableIterOpts = new HashMap<>();
+
+ // parse table iterator config once
+ IteratorUtil.parseIterConf(IteratorScope.scan, tableIters, tableIterOpts, tableConf);
+
+ context = tableConf.get(Property.TABLE_CLASSPATH);
+
- classCache = new HashMap<String,Class<? extends SortedKeyValueIterator<Key,Value>>>();
++ classCache = new HashMap<>();
+
+ tie = new TabletIteratorEnvironment(IteratorScope.scan, tableConf);
+ }
+
+ SortedKeyValueIterator<Key,Value> buildIterator(SortedKeyValueIterator<Key,Value> systemIter, TCondition tc) throws IOException {
+
+ ArrayByteSequence key = new ArrayByteSequence(tc.iterators);
+ MergedIterConfig mic = mergedIterCache.get(key);
+ if (mic == null) {
+ IterConfig ic = compressedIters.decompress(tc.iterators);
+
- List<IterInfo> mergedIters = new ArrayList<IterInfo>(tableIters.size() + ic.ssiList.size());
- Map<String,Map<String,String>> mergedItersOpts = new HashMap<String,Map<String,String>>(tableIterOpts.size() + ic.ssio.size());
++ List<IterInfo> mergedIters = new ArrayList<>(tableIters.size() + ic.ssiList.size());
++ Map<String,Map<String,String>> mergedItersOpts = new HashMap<>(tableIterOpts.size() + ic.ssio.size());
+
+ IteratorUtil.mergeIteratorConfig(mergedIters, mergedItersOpts, tableIters, tableIterOpts, ic.ssiList, ic.ssio);
+
+ mic = new MergedIterConfig(mergedIters, mergedItersOpts);
+
+ mergedIterCache.put(key, mic);
+ }
+
+ return IteratorUtil.loadIterators(systemIter, mic.mergedIters, mic.mergedItersOpts, tie, true, context, classCache);
+ }
+
+ boolean checkConditions(SortedKeyValueIterator<Key,Value> systemIter, ServerConditionalMutation scm) throws IOException {
+ boolean add = true;
+
+ for (TCondition tc : scm.getConditions()) {
+
+ Range range;
+ if (tc.hasTimestamp)
+ range = Range.exact(new Text(scm.getRow()), new Text(tc.getCf()), new Text(tc.getCq()), new Text(tc.getCv()), tc.getTs());
+ else
+ range = Range.exact(new Text(scm.getRow()), new Text(tc.getCf()), new Text(tc.getCq()), new Text(tc.getCv()));
+
+ SortedKeyValueIterator<Key,Value> iter = buildIterator(systemIter, tc);
+
+ ByteSequence cf = new ArrayByteSequence(tc.getCf());
+ iter.seek(range, Collections.singleton(cf), true);
+ Value val = null;
+ if (iter.hasTop()) {
+ val = iter.getTopValue();
+ }
+
+ if ((val == null ^ tc.getVal() == null) || (val != null && !Arrays.equals(tc.getVal(), val.get()))) {
+ add = false;
+ break;
+ }
+ }
+ return add;
+ }
+
+ public class ConditionChecker {
+
+ private List<ServerConditionalMutation> conditionsToCheck;
+ private List<ServerConditionalMutation> okMutations;
+ private List<TCMResult> results;
+ private boolean checked = false;
+
+ public ConditionChecker(List<ServerConditionalMutation> conditionsToCheck, List<ServerConditionalMutation> okMutations, List<TCMResult> results) {
+ this.conditionsToCheck = conditionsToCheck;
+ this.okMutations = okMutations;
+ this.results = results;
+ }
+
+ public void check(SortedKeyValueIterator<Key,Value> systemIter) throws IOException {
+ Preconditions.checkArgument(!checked, "check() method should only be called once");
+ checked = true;
+
+ for (ServerConditionalMutation scm : conditionsToCheck) {
+ if (checkConditions(systemIter, scm)) {
+ okMutations.add(scm);
+ } else {
+ results.add(new TCMResult(scm.getID(), TCMStatus.REJECTED));
+ }
+ }
+ }
+ }
+
+ public ConditionChecker newChecker(List<ServerConditionalMutation> conditionsToCheck, List<ServerConditionalMutation> okMutations, List<TCMResult> results) {
+ return new ConditionChecker(conditionsToCheck, okMutations, results);
+ }
+ }
http://git-wip-us.apache.org/repos/asf/accumulo/blob/d4882a15/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
----------------------------------------------------------------------
diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
index 1e0d119,6023ae3..038d3e8
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
@@@ -57,11 -65,8 +57,10 @@@ import java.util.concurrent.locks.Reent
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.Durability;
import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.impl.CompressedIterators;
- import org.apache.accumulo.core.client.impl.CompressedIterators.IterConfig;
+import org.apache.accumulo.core.client.impl.DurabilityImpl;
import org.apache.accumulo.core.client.impl.ScannerImpl;
import org.apache.accumulo.core.client.impl.Tables;
import org.apache.accumulo.core.client.impl.TabletLocator;
@@@ -91,9 -98,9 +90,8 @@@ import org.apache.accumulo.core.data.th
import org.apache.accumulo.core.data.thrift.TCMResult;
import org.apache.accumulo.core.data.thrift.TCMStatus;
import org.apache.accumulo.core.data.thrift.TColumn;
- import org.apache.accumulo.core.data.thrift.TCondition;
import org.apache.accumulo.core.data.thrift.TConditionalMutation;
import org.apache.accumulo.core.data.thrift.TConditionalSession;
-import org.apache.accumulo.core.data.thrift.TKey;
import org.apache.accumulo.core.data.thrift.TKeyExtent;
import org.apache.accumulo.core.data.thrift.TKeyValue;
import org.apache.accumulo.core.data.thrift.TMutation;
@@@ -197,7 -193,21 +195,8 @@@ import org.apache.accumulo.server.zooke
import org.apache.accumulo.server.zookeeper.ZooReaderWriter;
import org.apache.accumulo.start.classloader.vfs.AccumuloVFSClassLoader;
import org.apache.accumulo.start.classloader.vfs.ContextManager;
-import org.apache.accumulo.trace.instrument.Span;
-import org.apache.accumulo.trace.instrument.Trace;
-import org.apache.accumulo.trace.thrift.TInfo;
-import org.apache.accumulo.tserver.Compactor.CompactionInfo;
+ import org.apache.accumulo.tserver.ConditionCheckerContext.ConditionChecker;
import org.apache.accumulo.tserver.RowLocks.RowLock;
-import org.apache.accumulo.tserver.Tablet.CommitSession;
-import org.apache.accumulo.tserver.Tablet.KVEntry;
-import org.apache.accumulo.tserver.Tablet.LookupResult;
-import org.apache.accumulo.tserver.Tablet.MinorCompactionReason;
-import org.apache.accumulo.tserver.Tablet.ScanBatch;
-import org.apache.accumulo.tserver.Tablet.Scanner;
-import org.apache.accumulo.tserver.Tablet.SplitInfo;
-import org.apache.accumulo.tserver.Tablet.TConstraintViolationException;
-import org.apache.accumulo.tserver.Tablet.TabletClosedException;
import org.apache.accumulo.tserver.TabletServerResourceManager.TabletResourceManager;
import org.apache.accumulo.tserver.TabletStatsKeeper.Operation;
import org.apache.accumulo.tserver.compaction.MajorCompactionReason;
@@@ -209,30 -219,10 +208,28 @@@ import org.apache.accumulo.tserver.log.
import org.apache.accumulo.tserver.mastermessage.MasterMessage;
import org.apache.accumulo.tserver.mastermessage.SplitReportMessage;
import org.apache.accumulo.tserver.mastermessage.TabletStatusMessage;
-import org.apache.accumulo.tserver.metrics.TabletServerMBean;
-import org.apache.accumulo.tserver.metrics.TabletServerMinCMetrics;
+import org.apache.accumulo.tserver.metrics.TabletServerMetricsFactory;
import org.apache.accumulo.tserver.metrics.TabletServerScanMetrics;
import org.apache.accumulo.tserver.metrics.TabletServerUpdateMetrics;
+import org.apache.accumulo.tserver.replication.ReplicationServicerHandler;
+import org.apache.accumulo.tserver.replication.ReplicationWorker;
+import org.apache.accumulo.tserver.scan.LookupTask;
+import org.apache.accumulo.tserver.scan.NextBatchTask;
+import org.apache.accumulo.tserver.scan.ScanRunState;
+import org.apache.accumulo.tserver.session.ConditionalSession;
+import org.apache.accumulo.tserver.session.MultiScanSession;
+import org.apache.accumulo.tserver.session.ScanSession;
+import org.apache.accumulo.tserver.session.Session;
+import org.apache.accumulo.tserver.session.SessionManager;
+import org.apache.accumulo.tserver.session.UpdateSession;
+import org.apache.accumulo.tserver.tablet.CommitSession;
+import org.apache.accumulo.tserver.tablet.CompactionInfo;
+import org.apache.accumulo.tserver.tablet.CompactionWatcher;
+import org.apache.accumulo.tserver.tablet.Compactor;
- import org.apache.accumulo.tserver.tablet.KVEntry;
+import org.apache.accumulo.tserver.tablet.ScanBatch;
- import org.apache.accumulo.tserver.tablet.Scanner;
+import org.apache.accumulo.tserver.tablet.SplitInfo;
+import org.apache.accumulo.tserver.tablet.Tablet;
+import org.apache.accumulo.tserver.tablet.TabletClosedException;
import org.apache.commons.collections.map.LRUMap;
import org.apache.hadoop.fs.FSError;
import org.apache.hadoop.fs.FileSystem;
@@@ -256,151 -256,781 +253,150 @@@ public class TabletServer extends Accum
private static final long RECENTLY_SPLIT_MILLIES = 60 * 1000;
private static final long TIME_BETWEEN_GC_CHECKS = 5000;
private static final long TIME_BETWEEN_LOCATOR_CACHE_CLEARS = 60 * 60 * 1000;
- private static final Set<Column> EMPTY_COLUMNS = Collections.emptySet();
- private TabletServerLogger logger;
-
- protected TabletServerMinCMetrics mincMetrics = new TabletServerMinCMetrics();
+ private final GarbageCollectionLogger gcLogger = new GarbageCollectionLogger();
+ private final TransactionWatcher watcher = new TransactionWatcher();
+ private final ZooCache masterLockCache = new ZooCache();
- private ServerConfiguration serverConfig;
- private LogSorter logSorter = null;
+ private final TabletServerLogger logger;
- public TabletServer(ServerConfiguration conf, VolumeManager fs) {
- super();
- this.serverConfig = conf;
- this.instance = conf.getInstance();
- this.fs = fs;
+ private final TabletServerMetricsFactory metricsFactory;
+ private final Metrics updateMetrics;
+ private final Metrics scanMetrics;
+ private final Metrics mincMetrics;
- log.info("Version " + Constants.VERSION);
- log.info("Instance " + instance.getInstanceID());
-
- this.logSorter = new LogSorter(instance, fs, getSystemConfiguration());
- SimpleTimer.getInstance().schedule(new Runnable() {
- @Override
- public void run() {
- synchronized (onlineTablets) {
- long now = System.currentTimeMillis();
- for (Tablet tablet : onlineTablets.values())
- try {
- tablet.updateRates(now);
- } catch (Exception ex) {
- log.error(ex, ex);
- }
- }
- }
- }, TIME_BETWEEN_GC_CHECKS, TIME_BETWEEN_GC_CHECKS);
- SimpleTimer.getInstance().schedule(new Runnable() {
- @Override
- public void run() {
- TabletLocator.clearLocators();
- }
- }, jitter(TIME_BETWEEN_LOCATOR_CACHE_CLEARS), jitter(TIME_BETWEEN_LOCATOR_CACHE_CLEARS));
+ public Metrics getMinCMetrics() {
+ return mincMetrics;
}
- private static long jitter(long ms) {
- Random r = new Random();
- // add a random 10% wait
- return (long) ((1. + (r.nextDouble() / 10)) * ms);
- }
-
- private synchronized static void logGCInfo(AccumuloConfiguration conf) {
- long now = System.currentTimeMillis();
-
- List<GarbageCollectorMXBean> gcmBeans = ManagementFactory.getGarbageCollectorMXBeans();
- Runtime rt = Runtime.getRuntime();
-
- StringBuilder sb = new StringBuilder("gc");
-
- boolean sawChange = false;
-
- long maxIncreaseInCollectionTime = 0;
-
- for (GarbageCollectorMXBean gcBean : gcmBeans) {
- Long prevTime = prevGcTime.get(gcBean.getName());
- long pt = 0;
- if (prevTime != null) {
- pt = prevTime;
- }
-
- long time = gcBean.getCollectionTime();
-
- if (time - pt != 0) {
- sawChange = true;
- }
-
- long increaseInCollectionTime = time - pt;
- sb.append(String.format(" %s=%,.2f(+%,.2f) secs", gcBean.getName(), time / 1000.0, increaseInCollectionTime / 1000.0));
- maxIncreaseInCollectionTime = Math.max(increaseInCollectionTime, maxIncreaseInCollectionTime);
- prevGcTime.put(gcBean.getName(), time);
- }
-
- long mem = rt.freeMemory();
- if (maxIncreaseInCollectionTime == 0) {
- gcTimeIncreasedCount = 0;
- } else {
- gcTimeIncreasedCount++;
- if (gcTimeIncreasedCount > 3 && mem < rt.maxMemory() * 0.05) {
- log.warn("Running low on memory");
- gcTimeIncreasedCount = 0;
- }
- }
-
- if (mem > lastMemorySize) {
- sawChange = true;
- }
-
- String sign = "+";
- if (mem - lastMemorySize <= 0) {
- sign = "";
- }
-
- sb.append(String.format(" freemem=%,d(%s%,d) totalmem=%,d", mem, sign, (mem - lastMemorySize), rt.totalMemory()));
-
- if (sawChange) {
- log.debug(sb.toString());
- }
-
- final long keepAliveTimeout = conf.getTimeInMillis(Property.INSTANCE_ZK_TIMEOUT);
- if (lastMemoryCheckTime > 0 && lastMemoryCheckTime < now) {
- long diff = now - lastMemoryCheckTime;
- if (diff > keepAliveTimeout + 1000) {
- log.warn(String.format("GC pause checker not called in a timely fashion. Expected every %.1f seconds but was %.1f seconds since last check",
- TIME_BETWEEN_GC_CHECKS / 1000., diff / 1000.));
- }
- lastMemoryCheckTime = now;
- return;
- }
-
- if (maxIncreaseInCollectionTime > keepAliveTimeout) {
- Halt.halt("Garbage collection may be interfering with lock keep-alive. Halting.", -1);
- }
-
- lastMemorySize = mem;
- lastMemoryCheckTime = now;
- }
-
- private TabletStatsKeeper statsKeeper;
-
- private static class Session {
- long lastAccessTime;
- long startTime;
- String user;
- String client = TServerUtils.clientAddress.get();
- public boolean reserved;
-
- public boolean cleanup() {
- return true;
- }
- }
-
- private static class SessionManager {
-
- SecureRandom random;
- Map<Long,Session> sessions;
- private long maxIdle;
- private long maxUpdateIdle;
- private List<Session> idleSessions = new ArrayList<Session>();
- private final Long expiredSessionMarker = new Long(-1);
-
- SessionManager(AccumuloConfiguration conf) {
- random = new SecureRandom();
- sessions = new HashMap<Long,Session>();
- maxUpdateIdle = conf.getTimeInMillis(Property.TSERV_UPDATE_SESSION_MAXIDLE);
- maxIdle = conf.getTimeInMillis(Property.TSERV_SESSION_MAXIDLE);
-
- Runnable r = new Runnable() {
- @Override
- public void run() {
- sweep(maxIdle, maxUpdateIdle);
- }
- };
-
- SimpleTimer.getInstance().schedule(r, 0, Math.max(maxIdle / 2, 1000));
- }
-
- synchronized long createSession(Session session, boolean reserve) {
- long sid = random.nextLong();
-
- while (sessions.containsKey(sid)) {
- sid = random.nextLong();
- }
-
- sessions.put(sid, session);
-
- session.reserved = reserve;
-
- session.startTime = session.lastAccessTime = System.currentTimeMillis();
-
- return sid;
- }
-
- long getMaxIdleTime() {
- return maxIdle;
- }
-
- /**
- * while a session is reserved, it cannot be canceled or removed
- */
- synchronized Session reserveSession(long sessionId) {
- Session session = sessions.get(sessionId);
- if (session != null) {
- if (session.reserved)
- throw new IllegalStateException();
- session.reserved = true;
- }
-
- return session;
-
- }
-
- synchronized Session reserveSession(long sessionId, boolean wait) {
- Session session = sessions.get(sessionId);
- if (session != null) {
- while (wait && session.reserved) {
- try {
- wait(1000);
- } catch (InterruptedException e) {
- throw new RuntimeException();
- }
- }
-
- if (session.reserved)
- throw new IllegalStateException();
- session.reserved = true;
- }
-
- return session;
-
- }
-
- synchronized void unreserveSession(Session session) {
- if (!session.reserved)
- throw new IllegalStateException();
- notifyAll();
- session.reserved = false;
- session.lastAccessTime = System.currentTimeMillis();
- }
-
- synchronized void unreserveSession(long sessionId) {
- Session session = getSession(sessionId);
- if (session != null)
- unreserveSession(session);
- }
-
- synchronized Session getSession(long sessionId) {
- Session session = sessions.get(sessionId);
- if (session != null)
- session.lastAccessTime = System.currentTimeMillis();
- return session;
- }
-
- Session removeSession(long sessionId) {
- return removeSession(sessionId, false);
- }
-
- Session removeSession(long sessionId, boolean unreserve) {
- Session session = null;
- synchronized (this) {
- session = sessions.remove(sessionId);
- if (unreserve && session != null)
- unreserveSession(session);
- }
-
- // do clean up out side of lock..
- if (session != null)
- session.cleanup();
-
- return session;
- }
-
- private void sweep(final long maxIdle, final long maxUpdateIdle) {
- ArrayList<Session> sessionsToCleanup = new ArrayList<Session>();
- synchronized (this) {
- Iterator<Session> iter = sessions.values().iterator();
- while (iter.hasNext()) {
- Session session = iter.next();
- long configuredIdle = maxIdle;
- if (session instanceof UpdateSession) {
- configuredIdle = maxUpdateIdle;
- }
- long idleTime = System.currentTimeMillis() - session.lastAccessTime;
- if (idleTime > configuredIdle && !session.reserved) {
- log.info("Closing idle session from user=" + session.user + ", client=" + session.client + ", idle=" + idleTime + "ms");
- iter.remove();
- sessionsToCleanup.add(session);
- }
- }
- }
-
- // do clean up outside of lock for TabletServer in a synchronized block for simplicity vice a synchronized list
-
- synchronized (idleSessions) {
-
- sessionsToCleanup.addAll(idleSessions);
-
- idleSessions.clear();
-
- // perform cleanup for all of the sessions
- for (Session session : sessionsToCleanup) {
- if (!session.cleanup())
- idleSessions.add(session);
- }
- }
-
- }
-
- synchronized void removeIfNotAccessed(final long sessionId, final long delay) {
- Session session = sessions.get(sessionId);
- if (session != null) {
- final long removeTime = session.lastAccessTime;
- TimerTask r = new TimerTask() {
- @Override
- public void run() {
- Session sessionToCleanup = null;
- synchronized (SessionManager.this) {
- Session session2 = sessions.get(sessionId);
- if (session2 != null && session2.lastAccessTime == removeTime && !session2.reserved) {
- log.info("Closing not accessed session from user=" + session2.user + ", client=" + session2.client + ", duration=" + delay + "ms");
- sessions.remove(sessionId);
- sessionToCleanup = session2;
- }
- }
-
- // call clean up outside of lock
- if (sessionToCleanup != null)
- sessionToCleanup.cleanup();
- }
- };
-
- SimpleTimer.getInstance().schedule(r, delay);
- }
- }
-
- public synchronized Map<String,MapCounter<ScanRunState>> getActiveScansPerTable() {
- Map<String,MapCounter<ScanRunState>> counts = new HashMap<String,MapCounter<ScanRunState>>();
- Set<Entry<Long,Session>> copiedIdleSessions = new HashSet<Entry<Long,Session>>();
-
- synchronized (idleSessions) {
- /**
- * Add sessions so that get the list returned in the active scans call
- */
- for (Session session : idleSessions) {
- copiedIdleSessions.add(Maps.immutableEntry(expiredSessionMarker, session));
- }
- }
-
- for (Entry<Long,Session> entry : Iterables.concat(sessions.entrySet(), copiedIdleSessions)) {
-
- Session session = entry.getValue();
- @SuppressWarnings("rawtypes")
- ScanTask nbt = null;
- String tableID = null;
-
- if (session instanceof ScanSession) {
- ScanSession ss = (ScanSession) session;
- nbt = ss.nextBatchTask;
- tableID = ss.extent.getTableId().toString();
- } else if (session instanceof MultiScanSession) {
- MultiScanSession mss = (MultiScanSession) session;
- nbt = mss.lookupTask;
- tableID = mss.threadPoolExtent.getTableId().toString();
- }
-
- if (nbt == null)
- continue;
+ private final LogSorter logSorter;
+ private ReplicationWorker replWorker = null;
+ private final TabletStatsKeeper statsKeeper;
+ private final AtomicInteger logIdGenerator = new AtomicInteger();
- ScanRunState srs = nbt.getScanRunState();
+ private final AtomicLong flushCounter = new AtomicLong(0);
+ private final AtomicLong syncCounter = new AtomicLong(0);
- if (srs == ScanRunState.FINISHED)
- continue;
+ private final VolumeManager fs;
- MapCounter<ScanRunState> stateCounts = counts.get(tableID);
- if (stateCounts == null) {
- stateCounts = new MapCounter<ScanRunState>();
- counts.put(tableID, stateCounts);
- }
+ private final SortedMap<KeyExtent,Tablet> onlineTablets = Collections.synchronizedSortedMap(new TreeMap<KeyExtent,Tablet>());
+ private final SortedSet<KeyExtent> unopenedTablets = Collections.synchronizedSortedSet(new TreeSet<KeyExtent>());
+ private final SortedSet<KeyExtent> openingTablets = Collections.synchronizedSortedSet(new TreeSet<KeyExtent>());
+ @SuppressWarnings("unchecked")
+ private final Map<KeyExtent,Long> recentlyUnloadedCache = Collections.synchronizedMap(new LRUMap(1000));
- stateCounts.increment(srs, 1);
- }
+ private final TabletServerResourceManager resourceManager;
+ private final SecurityOperation security;
- return counts;
- }
+ private final BlockingDeque<MasterMessage> masterMessages = new LinkedBlockingDeque<MasterMessage>();
- public synchronized List<ActiveScan> getActiveScans() {
+ private Thread majorCompactorThread;
- final List<ActiveScan> activeScans = new ArrayList<ActiveScan>();
- final long ct = System.currentTimeMillis();
- final Set<Entry<Long,Session>> copiedIdleSessions = new HashSet<Entry<Long,Session>>();
+ private HostAndPort replicationAddress;
+ private HostAndPort clientAddress;
- synchronized (idleSessions) {
- /**
- * Add sessions so that get the list returned in the active scans call
- */
- for (Session session : idleSessions) {
- copiedIdleSessions.add(Maps.immutableEntry(expiredSessionMarker, session));
- }
- }
+ private volatile boolean serverStopRequested = false;
+ private volatile boolean majorCompactorDisabled = false;
+ private volatile boolean shutdownComplete = false;
- for (Entry<Long,Session> entry : Iterables.concat(sessions.entrySet(), copiedIdleSessions)) {
- Session session = entry.getValue();
- if (session instanceof ScanSession) {
- ScanSession ss = (ScanSession) session;
+ private ZooLock tabletServerLock;
- ScanState state = ScanState.RUNNING;
+ private TServer server;
+ private TServer replServer;
- ScanTask<ScanBatch> nbt = ss.nextBatchTask;
- if (nbt == null) {
- state = ScanState.IDLE;
- } else {
- switch (nbt.getScanRunState()) {
- case QUEUED:
- state = ScanState.QUEUED;
- break;
- case FINISHED:
- state = ScanState.IDLE;
- break;
- case RUNNING:
- default:
- /* do nothing */
- break;
- }
- }
+ private DistributedWorkQueue bulkFailedCopyQ;
- ActiveScan activeScan = new ActiveScan(ss.client, ss.user, ss.extent.getTableId().toString(), ct - ss.startTime, ct - ss.lastAccessTime,
- ScanType.SINGLE, state, ss.extent.toThrift(), Translator.translate(ss.columnSet, Translators.CT), ss.ssiList, ss.ssio,
- ss.auths.getAuthorizationsBB());
+ private String lockID;
- // scanId added by ACCUMULO-2641 is an optional thrift argument and not available in ActiveScan constructor
- activeScan.setScanId(entry.getKey());
- activeScans.add(activeScan);
+ public static final AtomicLong seekCount = new AtomicLong(0);
- } else if (session instanceof MultiScanSession) {
- MultiScanSession mss = (MultiScanSession) session;
+ private final AtomicLong totalMinorCompactions = new AtomicLong(0);
+ private final ServerConfigurationFactory confFactory;
- ScanState state = ScanState.RUNNING;
+ private final ZooAuthenticationKeyWatcher authKeyWatcher;
- ScanTask<MultiScanResult> nbt = mss.lookupTask;
- if (nbt == null) {
- state = ScanState.IDLE;
- } else {
- switch (nbt.getScanRunState()) {
- case QUEUED:
- state = ScanState.QUEUED;
- break;
- case FINISHED:
- state = ScanState.IDLE;
- break;
- case RUNNING:
- default:
- /* do nothing */
- break;
+ public TabletServer(ServerConfigurationFactory confFactory, VolumeManager fs) {
+ super(confFactory);
+ this.confFactory = confFactory;
+ this.fs = fs;
+ AccumuloConfiguration aconf = getConfiguration();
+ Instance instance = getInstance();
+ log.info("Version " + Constants.VERSION);
+ log.info("Instance " + instance.getInstanceID());
+ this.sessionManager = new SessionManager(aconf);
+ this.logSorter = new LogSorter(instance, fs, aconf);
+ this.replWorker = new ReplicationWorker(this, fs);
+ this.statsKeeper = new TabletStatsKeeper();
+ SimpleTimer.getInstance(aconf).schedule(new Runnable() {
+ @Override
+ public void run() {
+ synchronized (onlineTablets) {
+ long now = System.currentTimeMillis();
+ for (Tablet tablet : onlineTablets.values())
+ try {
+ tablet.updateRates(now);
+ } catch (Exception ex) {
+ log.error("Error updating rates for {}", tablet.getExtent(), ex);
}
- }
-
- activeScans.add(new ActiveScan(mss.client, mss.user, mss.threadPoolExtent.getTableId().toString(), ct - mss.startTime, ct - mss.lastAccessTime,
- ScanType.BATCH, state, mss.threadPoolExtent.toThrift(), Translator.translate(mss.columnSet, Translators.CT), mss.ssiList, mss.ssio, mss.auths
- .getAuthorizationsBB()));
}
}
+ }, 5000, 5000);
- return activeScans;
- }
- }
-
- static class TservConstraintEnv implements Environment {
-
- private TCredentials credentials;
- private SecurityOperation security;
- private Authorizations auths;
- private KeyExtent ke;
-
- TservConstraintEnv(SecurityOperation secOp, TCredentials credentials) {
- this.security = secOp;
- this.credentials = credentials;
- }
-
- void setExtent(KeyExtent ke) {
- this.ke = ke;
- }
-
- @Override
- public KeyExtent getExtent() {
- return ke;
- }
-
- @Override
- public String getUser() {
- return credentials.getPrincipal();
- }
-
- @Override
- @Deprecated
- public Authorizations getAuthorizations() {
- if (auths == null)
- try {
- this.auths = security.getUserAuthorizations(credentials);
- } catch (ThriftSecurityException e) {
- throw new RuntimeException(e);
- }
- return auths;
- }
-
- @Override
- public AuthorizationContainer getAuthorizationsContainer() {
- return new AuthorizationContainer() {
- @Override
- public boolean contains(ByteSequence auth) {
- try {
- return security.userHasAuthorizations(credentials,
- Collections.<ByteBuffer> singletonList(ByteBuffer.wrap(auth.getBackingArray(), auth.offset(), auth.length())));
- } catch (ThriftSecurityException e) {
- throw new RuntimeException(e);
- }
- }
- };
- }
- }
-
- private abstract class ScanTask<T> implements RunnableFuture<T> {
-
- protected AtomicBoolean interruptFlag;
- protected ArrayBlockingQueue<Object> resultQueue;
- protected AtomicInteger state;
- protected AtomicReference<ScanRunState> runState;
-
- private static final int INITIAL = 1;
- private static final int ADDED = 2;
- private static final int CANCELED = 3;
-
- ScanTask() {
- interruptFlag = new AtomicBoolean(false);
- runState = new AtomicReference<ScanRunState>(ScanRunState.QUEUED);
- state = new AtomicInteger(INITIAL);
- resultQueue = new ArrayBlockingQueue<Object>(1);
- }
-
- protected void addResult(Object o) {
- if (state.compareAndSet(INITIAL, ADDED))
- resultQueue.add(o);
- else if (state.get() == ADDED)
- throw new IllegalStateException("Tried to add more than one result");
- }
-
- @Override
- public boolean cancel(boolean mayInterruptIfRunning) {
- if (!mayInterruptIfRunning)
- throw new IllegalArgumentException("Cancel will always attempt to interupt running next batch task");
-
- if (state.get() == CANCELED)
- return true;
-
- if (state.compareAndSet(INITIAL, CANCELED)) {
- interruptFlag.set(true);
- resultQueue = null;
- return true;
- }
-
- return false;
- }
-
- @Override
- public T get() throws InterruptedException, ExecutionException {
- throw new UnsupportedOperationException();
- }
-
- @SuppressWarnings("unchecked")
- @Override
- public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
-
- ArrayBlockingQueue<Object> localRQ = resultQueue;
-
- if (state.get() == CANCELED)
- throw new CancellationException();
-
- if (localRQ == null) {
- int st = state.get();
- String stateStr;
- switch (st) {
- case ADDED:
- stateStr = "ADDED";
- break;
- case CANCELED:
- stateStr = "CANCELED";
- break;
- case INITIAL:
- stateStr = "INITIAL";
- break;
- default:
- stateStr = "UNKNOWN";
- break;
- }
- throw new IllegalStateException("Tried to get result twice [state=" + stateStr + "(" + st + ")]");
- }
-
- Object r = localRQ.poll(timeout, unit);
-
- // could have been canceled while waiting
- if (state.get() == CANCELED) {
- if (r != null)
- throw new IllegalStateException("Nothing should have been added when in canceled state");
+ final long walogMaxSize = aconf.getMemoryInBytes(Property.TSERV_WALOG_MAX_SIZE);
+ final long minBlockSize = CachedConfiguration.getInstance().getLong("dfs.namenode.fs-limits.min-block-size", 0);
+ if (minBlockSize != 0 && minBlockSize > walogMaxSize)
+ throw new RuntimeException("Unable to start TabletServer. Logger is set to use blocksize " + walogMaxSize + " but hdfs minimum block size is "
+ + minBlockSize + ". Either increase the " + Property.TSERV_WALOG_MAX_SIZE + " or decrease dfs.namenode.fs-limits.min-block-size in hdfs-site.xml.");
- throw new CancellationException();
+ final long toleratedWalCreationFailures = aconf.getCount(Property.TSERV_WALOG_TOLERATED_CREATION_FAILURES);
+ final long walCreationFailureRetryIncrement = aconf.getTimeInMillis(Property.TSERV_WALOG_TOLERATED_WAIT_INCREMENT);
+ final long walCreationFailureRetryMax = aconf.getTimeInMillis(Property.TSERV_WALOG_TOLERATED_MAXIMUM_WAIT_DURATION);
+ // Tolerate `toleratedWalCreationFailures` failures, waiting `walCreationFailureRetryIncrement` milliseconds after the first failure,
+ // incrementing the next wait period by the same value, for a maximum of `walCreationFailureRetryMax` retries.
+ final RetryFactory walCreationRetryFactory = new RetryFactory(toleratedWalCreationFailures, walCreationFailureRetryIncrement,
+ walCreationFailureRetryIncrement, walCreationFailureRetryMax);
+
+ logger = new TabletServerLogger(this, walogMaxSize, syncCounter, flushCounter, walCreationRetryFactory);
+ this.resourceManager = new TabletServerResourceManager(this, fs);
+ this.security = AuditedSecurityOperation.getInstance(this);
+
+ metricsFactory = new TabletServerMetricsFactory(aconf);
+ updateMetrics = metricsFactory.createUpdateMetrics();
+ scanMetrics = metricsFactory.createScanMetrics();
+ mincMetrics = metricsFactory.createMincMetrics();
+ SimpleTimer.getInstance(aconf).schedule(new Runnable() {
+ @Override
+ public void run() {
+ TabletLocator.clearLocators();
}
+ }, jitter(TIME_BETWEEN_LOCATOR_CACHE_CLEARS), jitter(TIME_BETWEEN_LOCATOR_CACHE_CLEARS));
- if (r == null)
- throw new TimeoutException();
-
- // make this method stop working now that something is being
- // returned
- resultQueue = null;
-
- if (r instanceof Throwable)
- throw new ExecutionException((Throwable) r);
-
- return (T) r;
- }
-
- @Override
- public boolean isCancelled() {
- return state.get() == CANCELED;
- }
-
- @Override
- public boolean isDone() {
- return runState.get().equals(ScanRunState.FINISHED);
- }
-
- public ScanRunState getScanRunState() {
- return runState.get();
- }
-
- }
-
- private static class ConditionalSession extends Session {
- public TCredentials credentials;
- public Authorizations auths;
- public String tableId;
- public AtomicBoolean interruptFlag;
-
- @Override
- public boolean cleanup() {
- interruptFlag.set(true);
- return true;
- }
- }
-
- private static class UpdateSession extends Session {
- public Tablet currentTablet;
- public MapCounter<Tablet> successfulCommits = new MapCounter<Tablet>();
- Map<KeyExtent,Long> failures = new HashMap<KeyExtent,Long>();
- HashMap<KeyExtent,SecurityErrorCode> authFailures = new HashMap<KeyExtent,SecurityErrorCode>();
- public Violations violations;
- public TCredentials credentials;
- public long totalUpdates = 0;
- public long flushTime = 0;
- Stat prepareTimes = new Stat();
- Stat walogTimes = new Stat();
- Stat commitTimes = new Stat();
- Stat authTimes = new Stat();
- public Map<Tablet,List<Mutation>> queuedMutations = new HashMap<Tablet,List<Mutation>>();
- public long queuedMutationSize = 0;
- TservConstraintEnv cenv = null;
- }
-
- private static class ScanSession extends Session {
- public KeyExtent extent;
- public HashSet<Column> columnSet;
- public List<IterInfo> ssiList;
- public Map<String,Map<String,String>> ssio;
- public Authorizations auths;
- public long entriesReturned = 0;
- public Stat nbTimes = new Stat();
- public long batchCount = 0;
- public volatile ScanTask<ScanBatch> nextBatchTask;
- public AtomicBoolean interruptFlag;
- public Scanner scanner;
- public long readaheadThreshold = Constants.SCANNER_DEFAULT_READAHEAD_THRESHOLD;
-
- @Override
- public boolean cleanup() {
- final boolean ret;
- try {
- if (nextBatchTask != null)
- nextBatchTask.cancel(true);
- } finally {
- if (scanner != null)
- ret = scanner.close();
- else
- ret = true;
- }
- return ret;
+ // Create the secret manager
+ setSecretManager(new AuthenticationTokenSecretManager(instance, aconf.getTimeInMillis(Property.GENERAL_DELEGATION_TOKEN_LIFETIME)));
+ if (aconf.getBoolean(Property.INSTANCE_RPC_SASL_ENABLED)) {
+ log.info("SASL is enabled, creating ZooKeeper watcher for AuthenticationKeys");
+ // Watcher to notice new AuthenticationKeys which enable delegation tokens
+ authKeyWatcher = new ZooAuthenticationKeyWatcher(getSecretManager(), ZooReaderWriter.getInstance(), ZooUtil.getRoot(instance)
+ + Constants.ZDELEGATION_TOKEN_KEYS);
+ } else {
+ authKeyWatcher = null;
}
-
}
- private static class MultiScanSession extends Session {
- HashSet<Column> columnSet;
- Map<KeyExtent,List<Range>> queries;
- public List<IterInfo> ssiList;
- public Map<String,Map<String,String>> ssio;
- public Authorizations auths;
-
- // stats
- int numRanges;
- int numTablets;
- int numEntries;
- long totalLookupTime;
-
- public volatile ScanTask<MultiScanResult> lookupTask;
- public KeyExtent threadPoolExtent;
-
- @Override
- public boolean cleanup() {
- if (lookupTask != null)
- lookupTask.cancel(true);
- // the cancellation should provide us the safety to return true here
- return true;
- }
+ private static long jitter(long ms) {
+ Random r = new Random();
+ // add a random 10% wait
+ return (long) ((1. + (r.nextDouble() / 10)) * ms);
}
- /**
- * This little class keeps track of writes in progress and allows readers to wait for writes that started before the read. It assumes that the operation ids
- * are monotonically increasing.
- *
- */
- static class WriteTracker {
- private static AtomicLong operationCounter = new AtomicLong(1);
- private Map<TabletType,TreeSet<Long>> inProgressWrites = new EnumMap<TabletType,TreeSet<Long>>(TabletType.class);
-
- WriteTracker() {
- for (TabletType ttype : TabletType.values()) {
- inProgressWrites.put(ttype, new TreeSet<Long>());
- }
- }
-
- synchronized long startWrite(TabletType ttype) {
- long operationId = operationCounter.getAndIncrement();
- inProgressWrites.get(ttype).add(operationId);
- return operationId;
- }
-
- synchronized void finishWrite(long operationId) {
- if (operationId == -1)
- return;
+ private final SessionManager sessionManager;
- boolean removed = false;
+ private final WriteTracker writeTracker = new WriteTracker();
- for (TabletType ttype : TabletType.values()) {
- removed = inProgressWrites.get(ttype).remove(operationId);
- if (removed)
- break;
- }
+ private final RowLocks rowLocks = new RowLocks();
- if (!removed) {
- throw new IllegalArgumentException("Attempted to finish write not in progress, operationId " + operationId);
- }
-
- this.notifyAll();
- }
-
- synchronized void waitForWrites(TabletType ttype) {
- long operationId = operationCounter.getAndIncrement();
- while (inProgressWrites.get(ttype).floor(operationId) != null) {
- try {
- this.wait();
- } catch (InterruptedException e) {
- log.error(e, e);
- }
- }
- }
-
- public long startWrite(Set<Tablet> keySet) {
- if (keySet.size() == 0)
- return -1;
-
- ArrayList<KeyExtent> extents = new ArrayList<KeyExtent>(keySet.size());
-
- for (Tablet tablet : keySet)
- extents.add(tablet.getExtent());
-
- return startWrite(TabletType.type(extents));
- }
- }
-
- public AccumuloConfiguration getSystemConfiguration() {
- return serverConfig.getConfiguration();
- }
-
- TransactionWatcher watcher = new TransactionWatcher();
+ private final AtomicLong totalQueuedMutationSize = new AtomicLong(0);
+ private final ReentrantLock recoveryLock = new ReentrantLock(true);
private class ThriftClientHandler extends ClientServiceHandler implements TabletClientService.Iface {
@@@ -1071,6 -1930,7 +1067,7 @@@
Iterator<Entry<KeyExtent,List<ServerConditionalMutation>>> iter = updates.entrySet().iterator();
final CompressedIterators compressedIters = new CompressedIterators(symbols);
- ConditionCheckerContext checkerContext = new ConditionCheckerContext(compressedIters, ServerConfiguration.getTableConfiguration(instance, cs.tableId));
++ ConditionCheckerContext checkerContext = new ConditionCheckerContext(compressedIters, confFactory.getTableConfiguration(cs.tableId));
while (iter.hasNext()) {
final Entry<KeyExtent,List<ServerConditionalMutation>> entry = iter.next();
@@@ -1082,65 -1942,44 +1079,28 @@@
iter.remove();
} else {
final List<ServerConditionalMutation> okMutations = new ArrayList<ServerConditionalMutation>(entry.getValue().size());
+ final List<TCMResult> resultsSubList = results.subList(results.size(), results.size());
- for (ServerConditionalMutation scm : entry.getValue()) {
- if (checkCondition(results, cs, compressedIters, tablet, scm))
- okMutations.add(scm);
- }
-
- entry.setValue(okMutations);
- }
-
- }
- }
-
- private boolean checkCondition(ArrayList<TCMResult> results, ConditionalSession cs, CompressedIterators compressedIters, Tablet tablet,
- ServerConditionalMutation scm) throws IOException {
- boolean add = true;
-
- for (TCondition tc : scm.getConditions()) {
-
- Range range;
- if (tc.hasTimestamp)
- range = Range.exact(new Text(scm.getRow()), new Text(tc.getCf()), new Text(tc.getCq()), new Text(tc.getCv()), tc.getTs());
- else
- range = Range.exact(new Text(scm.getRow()), new Text(tc.getCf()), new Text(tc.getCq()), new Text(tc.getCv()));
-
- IterConfig ic = compressedIters.decompress(tc.iterators);
-
- Scanner scanner = tablet.createScanner(range, 1, EMPTY_COLUMNS, cs.auths, ic.ssiList, ic.ssio, false, cs.interruptFlag);
-
- try {
- ScanBatch batch = scanner.read();
-
- Value val = null;
+ ConditionChecker checker = checkerContext.newChecker(entry.getValue(), okMutations, resultsSubList);
+ try {
+ tablet.checkConditions(checker, cs.auths, cs.interruptFlag);
- for (KVEntry entry2 : batch.getResults()) {
- val = entry2.getValue();
- break;
- }
+ if (okMutations.size() > 0) {
+ entry.setValue(okMutations);
+ } else {
+ iter.remove();
+ }
- } catch (TabletClosedException e) {
- // clear anything added while checking conditions.
- resultsSubList.clear();
-
- for (ServerConditionalMutation scm : entry.getValue()) {
- results.add(new TCMResult(scm.getID(), TCMStatus.IGNORED));
- }
- iter.remove();
- } catch (IterationInterruptedException e) {
- // clear anything added while checking conditions.
- resultsSubList.clear();
-
- for (ServerConditionalMutation scm : entry.getValue()) {
- results.add(new TCMResult(scm.getID(), TCMStatus.IGNORED));
- }
- iter.remove();
- } catch (TooManyFilesException e) {
++ } catch (TabletClosedException | IterationInterruptedException | TooManyFilesException e) {
+ // clear anything added while checking conditions.
+ resultsSubList.clear();
- if ((val == null ^ tc.getVal() == null) || (val != null && !Arrays.equals(tc.getVal(), val.get()))) {
- results.add(new TCMResult(scm.getID(), TCMStatus.REJECTED));
- add = false;
- break;
+ for (ServerConditionalMutation scm : entry.getValue()) {
+ results.add(new TCMResult(scm.getID(), TCMStatus.IGNORED));
+ }
+ iter.remove();
}
-
- } catch (TabletClosedException e) {
- results.add(new TCMResult(scm.getID(), TCMStatus.IGNORED));
- add = false;
- break;
- } catch (IterationInterruptedException iie) {
- results.add(new TCMResult(scm.getID(), TCMStatus.IGNORED));
- add = false;
- break;
- } catch (TooManyFilesException tmfe) {
- results.add(new TCMResult(scm.getID(), TCMStatus.IGNORED));
- add = false;
- break;
}
}
- return add;
}
private void writeConditionalMutations(Map<KeyExtent,List<ServerConditionalMutation>> updates, ArrayList<TCMResult> results, ConditionalSession sess) {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/d4882a15/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/ScanDataSource.java
----------------------------------------------------------------------
diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/ScanDataSource.java
index 33277bd,0000000..f586e2e
mode 100644,000000..100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/ScanDataSource.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/ScanDataSource.java
@@@ -1,228 -1,0 +1,247 @@@
+/*
+ * 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.accumulo.tserver.tablet;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
++import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
++import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.accumulo.core.data.Column;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.data.thrift.IterInfo;
+import org.apache.accumulo.core.iterators.IterationInterruptedException;
+import org.apache.accumulo.core.iterators.IteratorEnvironment;
+import org.apache.accumulo.core.iterators.IteratorUtil;
+import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
+import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
+import org.apache.accumulo.core.iterators.system.ColumnFamilySkippingIterator;
+import org.apache.accumulo.core.iterators.system.ColumnQualifierFilter;
+import org.apache.accumulo.core.iterators.system.DeletingIterator;
+import org.apache.accumulo.core.iterators.system.InterruptibleIterator;
+import org.apache.accumulo.core.iterators.system.MultiIterator;
+import org.apache.accumulo.core.iterators.system.SourceSwitchingIterator.DataSource;
+import org.apache.accumulo.core.iterators.system.StatsIterator;
+import org.apache.accumulo.core.iterators.system.VisibilityFilter;
+import org.apache.accumulo.core.metadata.schema.DataFileValue;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.util.Pair;
+import org.apache.accumulo.server.fs.FileRef;
+import org.apache.accumulo.tserver.FileManager.ScanFileManager;
+import org.apache.accumulo.tserver.InMemoryMap.MemoryIterator;
+import org.apache.accumulo.tserver.TabletIteratorEnvironment;
+import org.apache.accumulo.tserver.TabletServer;
+
+class ScanDataSource implements DataSource {
+
+ // data source state
+ private final Tablet tablet;
+ private ScanFileManager fileManager;
+ private SortedKeyValueIterator<Key,Value> iter;
+ private long expectedDeletionCount;
+ private List<MemoryIterator> memIters = null;
+ private long fileReservationId;
+ private AtomicBoolean interruptFlag;
+ private StatsIterator statsIterator;
+
+ private final ScanOptions options;
++ private final boolean loadIters;
++
++ private static final Set<Column> EMPTY_COLS = Collections.emptySet();
+
+ ScanDataSource(Tablet tablet, Authorizations authorizations, byte[] defaultLabels, HashSet<Column> columnSet, List<IterInfo> ssiList,
+ Map<String,Map<String,String>> ssio, AtomicBoolean interruptFlag) {
+ this.tablet = tablet;
+ expectedDeletionCount = tablet.getDataSourceDeletions();
+ this.options = new ScanOptions(-1, authorizations, defaultLabels, columnSet, ssiList, ssio, interruptFlag, false);
+ this.interruptFlag = interruptFlag;
++ this.loadIters = true;
+ }
+
+ ScanDataSource(Tablet tablet, ScanOptions options) {
+ this.tablet = tablet;
+ expectedDeletionCount = tablet.getDataSourceDeletions();
+ this.options = options;
+ this.interruptFlag = options.getInterruptFlag();
++ this.loadIters = true;
++ }
++
++ ScanDataSource(Tablet tablet, Authorizations authorizations, byte[] defaultLabels, AtomicBoolean iFlag) {
++ this.tablet = tablet;
++ expectedDeletionCount = tablet.getDataSourceDeletions();
++ this.options = new ScanOptions(-1, authorizations, defaultLabels, EMPTY_COLS, null, null, iFlag, false);
++ this.interruptFlag = iFlag;
++ this.loadIters = false;
+ }
+
+ @Override
+ public DataSource getNewDataSource() {
+ if (!isCurrent()) {
+ // log.debug("Switching data sources during a scan");
+ if (memIters != null) {
+ tablet.getTabletMemory().returnIterators(memIters);
+ memIters = null;
+ tablet.getDatafileManager().returnFilesForScan(fileReservationId);
+ fileReservationId = -1;
+ }
+
+ if (fileManager != null)
+ fileManager.releaseOpenFiles(false);
+
+ expectedDeletionCount = tablet.getDataSourceDeletions();
+ iter = null;
+
+ return this;
+ } else
+ return this;
+ }
+
+ @Override
+ public boolean isCurrent() {
+ return expectedDeletionCount == tablet.getDataSourceDeletions();
+ }
+
+ @Override
+ public SortedKeyValueIterator<Key,Value> iterator() throws IOException {
+ if (iter == null)
+ iter = createIterator();
+ return iter;
+ }
+
+ private SortedKeyValueIterator<Key,Value> createIterator() throws IOException {
+
+ Map<FileRef,DataFileValue> files;
+
+ synchronized (tablet) {
+
+ if (memIters != null)
+ throw new IllegalStateException("Tried to create new scan iterator w/o releasing memory");
+
+ if (tablet.isClosed())
+ throw new TabletClosedException();
+
+ if (interruptFlag.get())
+ throw new IterationInterruptedException(tablet.getExtent().toString() + " " + interruptFlag.hashCode());
+
+ // only acquire the file manager when we know the tablet is open
+ if (fileManager == null) {
+ fileManager = tablet.getTabletResources().newScanFileManager();
+ tablet.addActiveScans(this);
+ }
+
+ if (fileManager.getNumOpenFiles() != 0)
+ throw new IllegalStateException("Tried to create new scan iterator w/o releasing files");
+
+ // set this before trying to get iterators in case
+ // getIterators() throws an exception
+ expectedDeletionCount = tablet.getDataSourceDeletions();
+
+ memIters = tablet.getTabletMemory().getIterators();
+ Pair<Long,Map<FileRef,DataFileValue>> reservation = tablet.getDatafileManager().reserveFilesForScan();
+ fileReservationId = reservation.getFirst();
+ files = reservation.getSecond();
+ }
+
+ Collection<InterruptibleIterator> mapfiles = fileManager.openFiles(files, options.isIsolated());
+
+ List<SortedKeyValueIterator<Key,Value>> iters = new ArrayList<SortedKeyValueIterator<Key,Value>>(mapfiles.size() + memIters.size());
+
+ iters.addAll(mapfiles);
+ iters.addAll(memIters);
+
+ for (SortedKeyValueIterator<Key,Value> skvi : iters)
+ ((InterruptibleIterator) skvi).setInterruptFlag(interruptFlag);
+
+ MultiIterator multiIter = new MultiIterator(iters, tablet.getExtent());
+
+ TabletIteratorEnvironment iterEnv = new TabletIteratorEnvironment(IteratorScope.scan, tablet.getTableConfiguration(), fileManager, files,
+ options.getAuthorizations());
+
+ statsIterator = new StatsIterator(multiIter, TabletServer.seekCount, tablet.getScannedCounter());
+
+ DeletingIterator delIter = new DeletingIterator(statsIterator, false);
+
+ ColumnFamilySkippingIterator cfsi = new ColumnFamilySkippingIterator(delIter);
+
+ ColumnQualifierFilter colFilter = new ColumnQualifierFilter(cfsi, options.getColumnSet());
+
+ VisibilityFilter visFilter = new VisibilityFilter(colFilter, options.getAuthorizations(), options.getDefaultLabels());
+
- return iterEnv.getTopLevelIterator(IteratorUtil.loadIterators(IteratorScope.scan, visFilter, tablet.getExtent(), tablet.getTableConfiguration(),
- options.getSsiList(), options.getSsio(), iterEnv));
++ if (!loadIters) {
++ return visFilter;
++ } else {
++ return iterEnv.getTopLevelIterator(IteratorUtil.loadIterators(IteratorScope.scan, visFilter, tablet.getExtent(), tablet.getTableConfiguration(),
++ options.getSsiList(), options.getSsio(), iterEnv));
++ }
+ }
+
+ void close(boolean sawErrors) {
+
+ if (memIters != null) {
+ tablet.getTabletMemory().returnIterators(memIters);
+ memIters = null;
+ tablet.getDatafileManager().returnFilesForScan(fileReservationId);
+ fileReservationId = -1;
+ }
+
+ synchronized (tablet) {
+ if (tablet.removeScan(this) == 0)
+ tablet.notifyAll();
+ }
+
+ if (fileManager != null) {
+ fileManager.releaseOpenFiles(sawErrors);
+ fileManager = null;
+ }
+
+ if (statsIterator != null) {
+ statsIterator.report();
+ }
+
+ }
+
+ public void interrupt() {
+ interruptFlag.set(true);
+ }
+
+ @Override
+ public DataSource getDeepCopyDataSource(IteratorEnvironment env) {
+ throw new UnsupportedOperationException();
+ }
+
+ public void reattachFileManager() throws IOException {
+ if (fileManager != null)
+ fileManager.reattach();
+ }
+
+ public void detachFileManager() {
+ if (fileManager != null)
+ fileManager.detach();
+ }
+
+ @Override
+ public void setInterruptFlag(AtomicBoolean flag) {
+ throw new UnsupportedOperationException();
+ }
+
+}
[3/5] accumulo git commit: Merge branch '1.6' into 1.7
Posted by kt...@apache.org.
http://git-wip-us.apache.org/repos/asf/accumulo/blob/d4882a15/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
----------------------------------------------------------------------
diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
index a75b35d,0000000..923b4a1
mode 100644,000000..100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
@@@ -1,2762 -1,0 +1,2780 @@@
+/*
+ * 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.accumulo.tserver.tablet;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.PriorityQueue;
+import java.util.Set;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.Durability;
+import org.apache.accumulo.core.client.IteratorSetting;
+import org.apache.accumulo.core.client.admin.CompactionStrategyConfig;
+import org.apache.accumulo.core.client.impl.DurabilityImpl;
+import org.apache.accumulo.core.client.impl.Tables;
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.conf.ConfigurationCopy;
+import org.apache.accumulo.core.conf.ConfigurationObserver;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.constraints.Violations;
+import org.apache.accumulo.core.data.ByteSequence;
+import org.apache.accumulo.core.data.Column;
+import org.apache.accumulo.core.data.ColumnUpdate;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.data.impl.KeyExtent;
+import org.apache.accumulo.core.data.thrift.IterInfo;
+import org.apache.accumulo.core.data.thrift.MapFileInfo;
+import org.apache.accumulo.core.file.FileOperations;
+import org.apache.accumulo.core.file.FileSKVIterator;
+import org.apache.accumulo.core.iterators.IterationInterruptedException;
+import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
+import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
+import org.apache.accumulo.core.iterators.system.SourceSwitchingIterator;
+import org.apache.accumulo.core.master.thrift.TabletLoadState;
+import org.apache.accumulo.core.metadata.MetadataTable;
+import org.apache.accumulo.core.metadata.RootTable;
+import org.apache.accumulo.core.metadata.schema.DataFileValue;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.LogColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ScanFileColumnFamily;
+import org.apache.accumulo.core.protobuf.ProtobufUtil;
+import org.apache.accumulo.core.replication.ReplicationConfigurationUtil;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.security.ColumnVisibility;
+import org.apache.accumulo.core.tabletserver.log.LogEntry;
+import org.apache.accumulo.core.tabletserver.thrift.TabletStats;
+import org.apache.accumulo.core.trace.ProbabilitySampler;
+import org.apache.accumulo.core.trace.Span;
+import org.apache.accumulo.core.trace.Trace;
+import org.apache.accumulo.core.util.LocalityGroupUtil;
+import org.apache.accumulo.core.util.Pair;
+import org.apache.accumulo.core.util.UtilWaitThread;
+import org.apache.accumulo.server.AccumuloServerContext;
+import org.apache.accumulo.server.ServerConstants;
+import org.apache.accumulo.server.conf.TableConfiguration;
+import org.apache.accumulo.server.fs.FileRef;
+import org.apache.accumulo.server.fs.VolumeManager;
+import org.apache.accumulo.server.fs.VolumeManager.FileType;
+import org.apache.accumulo.server.fs.VolumeUtil;
+import org.apache.accumulo.server.fs.VolumeUtil.TabletFiles;
+import org.apache.accumulo.server.master.state.TServerInstance;
+import org.apache.accumulo.server.master.tableOps.UserCompactionConfig;
+import org.apache.accumulo.server.metrics.Metrics;
+import org.apache.accumulo.server.problems.ProblemReport;
+import org.apache.accumulo.server.problems.ProblemReports;
+import org.apache.accumulo.server.problems.ProblemType;
+import org.apache.accumulo.server.replication.StatusUtil;
+import org.apache.accumulo.server.replication.proto.Replication.Status;
+import org.apache.accumulo.server.tablets.TabletTime;
+import org.apache.accumulo.server.tablets.UniqueNameAllocator;
+import org.apache.accumulo.server.util.FileUtil;
+import org.apache.accumulo.server.util.MasterMetadataUtil;
+import org.apache.accumulo.server.util.MetadataTableUtil;
+import org.apache.accumulo.server.util.ReplicationTableUtil;
+import org.apache.accumulo.server.zookeeper.ZooReaderWriter;
+import org.apache.accumulo.start.classloader.vfs.AccumuloVFSClassLoader;
++import org.apache.accumulo.tserver.ConditionCheckerContext.ConditionChecker;
+import org.apache.accumulo.tserver.InMemoryMap;
+import org.apache.accumulo.tserver.MinorCompactionReason;
+import org.apache.accumulo.tserver.TConstraintViolationException;
+import org.apache.accumulo.tserver.TLevel;
+import org.apache.accumulo.tserver.TabletServer;
+import org.apache.accumulo.tserver.TabletServerResourceManager.TabletResourceManager;
+import org.apache.accumulo.tserver.TabletStatsKeeper;
+import org.apache.accumulo.tserver.TabletStatsKeeper.Operation;
+import org.apache.accumulo.tserver.TooManyFilesException;
+import org.apache.accumulo.tserver.TservConstraintEnv;
+import org.apache.accumulo.tserver.compaction.CompactionPlan;
+import org.apache.accumulo.tserver.compaction.CompactionStrategy;
+import org.apache.accumulo.tserver.compaction.DefaultCompactionStrategy;
+import org.apache.accumulo.tserver.compaction.MajorCompactionReason;
+import org.apache.accumulo.tserver.compaction.MajorCompactionRequest;
+import org.apache.accumulo.tserver.compaction.WriteParameters;
+import org.apache.accumulo.tserver.constraints.ConstraintChecker;
+import org.apache.accumulo.tserver.log.DfsLogger;
+import org.apache.accumulo.tserver.log.MutationReceiver;
+import org.apache.accumulo.tserver.mastermessage.TabletStatusMessage;
+import org.apache.accumulo.tserver.metrics.TabletServerMinCMetrics;
+import org.apache.accumulo.tserver.tablet.Compactor.CompactionCanceledException;
+import org.apache.accumulo.tserver.tablet.Compactor.CompactionEnv;
+import org.apache.commons.codec.DecoderException;
+import org.apache.commons.codec.binary.Hex;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.log4j.Logger;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.KeeperException.NoNodeException;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Optional;
+import com.google.common.base.Preconditions;
+
+/**
+ *
+ * Provide access to a single row range in a living TabletServer.
+ *
+ */
+public class Tablet implements TabletCommitter {
+ static private final Logger log = Logger.getLogger(Tablet.class);
+ static private final List<LogEntry> NO_LOG_ENTRIES = Collections.emptyList();
+
+ private final TabletServer tabletServer;
+ private final KeyExtent extent;
+ private final TabletResourceManager tabletResources;
+ private final DatafileManager datafileManager;
+ private final TableConfiguration tableConfiguration;
+ private final String tabletDirectory;
+ private final Path location; // absolute path of this tablets dir
+
+ private final TabletMemory tabletMemory;
+
+ private final TabletTime tabletTime;
+ private final Object timeLock = new Object();
+ private long persistedTime;
+
+ private TServerInstance lastLocation;
+ private volatile boolean tableDirChecked = false;
+
+ private final AtomicLong dataSourceDeletions = new AtomicLong(0);
+
+ public long getDataSourceDeletions() {
+ return dataSourceDeletions.get();
+ }
+
+ private final Set<ScanDataSource> activeScans = new HashSet<ScanDataSource>();
+
+ private static enum CloseState {
+ OPEN, CLOSING, CLOSED, COMPLETE
+ }
+
+ private volatile CloseState closeState = CloseState.OPEN;
+
+ private boolean updatingFlushID = false;
+
+ private long lastFlushID = -1;
+ private long lastCompactID = -1;
+
+ private static class CompactionWaitInfo {
+ long flushID = -1;
+ long compactionID = -1;
+ }
+
+ // stores info about user initiated major compaction that is waiting on a minor compaction to finish
+ private CompactionWaitInfo compactionWaitInfo = new CompactionWaitInfo();
+
+ static enum CompactionState {
+ WAITING_TO_START, IN_PROGRESS
+ };
+
+ private volatile CompactionState minorCompactionState = null;
+ private volatile CompactionState majorCompactionState = null;
+
+ private final Set<MajorCompactionReason> majorCompactionQueued = Collections.synchronizedSet(EnumSet.noneOf(MajorCompactionReason.class));
+
+ private final AtomicReference<ConstraintChecker> constraintChecker = new AtomicReference<ConstraintChecker>();
+
+ private int writesInProgress = 0;
+
+ private final TabletStatsKeeper timer = new TabletStatsKeeper();
+
+ private final Rate queryRate = new Rate(0.95);
+ private long queryCount = 0;
+
+ private final Rate queryByteRate = new Rate(0.95);
+ private long queryBytes = 0;
+
+ private final Rate ingestRate = new Rate(0.95);
+ private long ingestCount = 0;
+
+ private final Rate ingestByteRate = new Rate(0.95);
+ private long ingestBytes = 0;
+
+ private byte[] defaultSecurityLabel = new byte[0];
+
+ private long lastMinorCompactionFinishTime = 0;
+ private long lastMapFileImportTime = 0;
+
+ private volatile long numEntries = 0;
+ private volatile long numEntriesInMemory = 0;
+
+ private final Rate scannedRate = new Rate(0.95);
+ private final AtomicLong scannedCount = new AtomicLong(0);
+
+ private final ConfigurationObserver configObserver;
+
+ private final int logId;
+
+ @Override
+ public int getLogId() {
+ return logId;
+ }
+
+ public static class LookupResult {
+ public List<Range> unfinishedRanges = new ArrayList<Range>();
+ public long bytesAdded = 0;
+ public long dataSize = 0;
+ public boolean closed = false;
+ }
+
+ FileRef getNextMapFilename(String prefix) throws IOException {
+ String extension = FileOperations.getNewFileExtension(tableConfiguration);
+ checkTabletDir();
+ return new FileRef(location.toString() + "/" + prefix + UniqueNameAllocator.getInstance().getNextName() + "." + extension);
+ }
+
+ private void checkTabletDir() throws IOException {
+ if (!tableDirChecked) {
+ FileStatus[] files = null;
+ try {
+ files = getTabletServer().getFileSystem().listStatus(location);
+ } catch (FileNotFoundException ex) {
+ // ignored
+ }
+
+ if (files == null) {
+ if (location.getName().startsWith(Constants.CLONE_PREFIX))
+ log.debug("Tablet " + extent + " had no dir, creating " + location); // its a clone dir...
+ else
+ log.warn("Tablet " + extent + " had no dir, creating " + location);
+
+ getTabletServer().getFileSystem().mkdirs(location);
+ }
+ tableDirChecked = true;
+ }
+ }
+
+ /**
+ * Only visible for testing
+ */
+ @VisibleForTesting
+ protected Tablet(TabletTime tabletTime, String tabletDirectory, int logId, Path location, DatafileManager datafileManager, TabletServer tabletServer,
+ TabletResourceManager tabletResources, TabletMemory tabletMemory, TableConfiguration tableConfiguration, KeyExtent extent,
+ ConfigurationObserver configObserver) {
+ this.tabletTime = tabletTime;
+ this.tabletDirectory = tabletDirectory;
+ this.logId = logId;
+ this.location = location;
+ this.datafileManager = datafileManager;
+ this.tabletServer = tabletServer;
+ this.tabletResources = tabletResources;
+ this.tabletMemory = tabletMemory;
+ this.tableConfiguration = tableConfiguration;
+ this.extent = extent;
+ this.configObserver = configObserver;
+ }
+
+ public Tablet(TabletServer tabletServer, KeyExtent extent, TabletResourceManager trm, SplitInfo info) throws IOException {
+ this(tabletServer, new Text(info.getDir()), extent, trm, info.getDatafiles(), info.getTime(), info.getInitFlushID(), info.getInitCompactID(), info
+ .getLastLocation());
+ splitCreationTime = System.currentTimeMillis();
+ }
+
+ private Tablet(TabletServer tabletServer, Text location, KeyExtent extent, TabletResourceManager trm, SortedMap<FileRef,DataFileValue> datafiles,
+ String time, long initFlushID, long initCompactID, TServerInstance lastLocation) throws IOException {
+ this(tabletServer, extent, location, trm, NO_LOG_ENTRIES, datafiles, time, lastLocation, new HashSet<FileRef>(), initFlushID, initCompactID);
+ }
+
+ private static String lookupTime(AccumuloConfiguration conf, KeyExtent extent, SortedMap<Key,Value> tabletsKeyValues) {
+ SortedMap<Key,Value> entries;
+
+ if (extent.isRootTablet()) {
+ return null;
+ } else {
+ entries = new TreeMap<Key,Value>();
+ Text rowName = extent.getMetadataEntry();
+ for (Entry<Key,Value> entry : tabletsKeyValues.entrySet()) {
+ if (entry.getKey().compareRow(rowName) == 0 && TabletsSection.ServerColumnFamily.TIME_COLUMN.hasColumns(entry.getKey())) {
+ entries.put(new Key(entry.getKey()), new Value(entry.getValue()));
+ }
+ }
+ }
+
+ if (entries.size() == 1)
+ return entries.values().iterator().next().toString();
+ return null;
+ }
+
+ private static SortedMap<FileRef,DataFileValue> lookupDatafiles(AccumuloServerContext context, VolumeManager fs, KeyExtent extent,
+ SortedMap<Key,Value> tabletsKeyValues) throws IOException {
+
+ TreeMap<FileRef,DataFileValue> datafiles = new TreeMap<FileRef,DataFileValue>();
+
+ if (extent.isRootTablet()) { // the meta0 tablet
+ Path location = new Path(MetadataTableUtil.getRootTabletDir());
+
+ // cleanUpFiles() has special handling for delete. files
+ FileStatus[] files = fs.listStatus(location);
+ Collection<String> goodPaths = RootFiles.cleanupReplacement(fs, files, true);
+ for (String good : goodPaths) {
+ Path path = new Path(good);
+ String filename = path.getName();
+ FileRef ref = new FileRef(location.toString() + "/" + filename, path);
+ DataFileValue dfv = new DataFileValue(0, 0);
+ datafiles.put(ref, dfv);
+ }
+ } else {
+ final Text buffer = new Text();
+ final Text row = extent.getMetadataEntry();
+
+ for (Entry<Key,Value> entry : tabletsKeyValues.entrySet()) {
+ Key k = entry.getKey();
+ k.getRow(buffer);
+ // Double-check that we have the expected row
+ if (row.equals(buffer)) {
+ k.getColumnFamily(buffer);
+ // Ignore anything but file:
+ if (TabletsSection.DataFileColumnFamily.NAME.equals(buffer)) {
+ FileRef ref = new FileRef(fs, k);
+ datafiles.put(ref, new DataFileValue(entry.getValue().get()));
+ }
+ }
+ }
+ }
+ return datafiles;
+ }
+
+ private static List<LogEntry> lookupLogEntries(AccumuloServerContext context, KeyExtent ke, SortedMap<Key,Value> tabletsKeyValues) {
+ List<LogEntry> logEntries = new ArrayList<LogEntry>();
+
+ if (ke.isMeta()) {
+ try {
+ logEntries = MetadataTableUtil.getLogEntries(context, ke);
+ } catch (Exception ex) {
+ throw new RuntimeException("Unable to read tablet log entries", ex);
+ }
+ } else {
+ log.debug("Looking at metadata " + tabletsKeyValues);
+ Text row = ke.getMetadataEntry();
+ for (Entry<Key,Value> entry : tabletsKeyValues.entrySet()) {
+ Key key = entry.getKey();
+ if (key.getRow().equals(row)) {
+ if (key.getColumnFamily().equals(LogColumnFamily.NAME)) {
+ logEntries.add(LogEntry.fromKeyValue(key, entry.getValue()));
+ }
+ }
+ }
+ }
+
+ log.debug("got " + logEntries + " for logs for " + ke);
+ return logEntries;
+ }
+
+ private static Set<FileRef> lookupScanFiles(KeyExtent extent, SortedMap<Key,Value> tabletsKeyValues, VolumeManager fs) throws IOException {
+ HashSet<FileRef> scanFiles = new HashSet<FileRef>();
+
+ Text row = extent.getMetadataEntry();
+ for (Entry<Key,Value> entry : tabletsKeyValues.entrySet()) {
+ Key key = entry.getKey();
+ if (key.getRow().equals(row) && key.getColumnFamily().equals(ScanFileColumnFamily.NAME)) {
+ scanFiles.add(new FileRef(fs, key));
+ }
+ }
+
+ return scanFiles;
+ }
+
+ private static long lookupFlushID(KeyExtent extent, SortedMap<Key,Value> tabletsKeyValues) {
+ Text row = extent.getMetadataEntry();
+ for (Entry<Key,Value> entry : tabletsKeyValues.entrySet()) {
+ Key key = entry.getKey();
+ if (key.getRow().equals(row) && TabletsSection.ServerColumnFamily.FLUSH_COLUMN.equals(key.getColumnFamily(), key.getColumnQualifier()))
+ return Long.parseLong(entry.getValue().toString());
+ }
+
+ return -1;
+ }
+
+ private static long lookupCompactID(KeyExtent extent, SortedMap<Key,Value> tabletsKeyValues) {
+ Text row = extent.getMetadataEntry();
+ for (Entry<Key,Value> entry : tabletsKeyValues.entrySet()) {
+ Key key = entry.getKey();
+ if (key.getRow().equals(row) && TabletsSection.ServerColumnFamily.COMPACT_COLUMN.equals(key.getColumnFamily(), key.getColumnQualifier()))
+ return Long.parseLong(entry.getValue().toString());
+ }
+
+ return -1;
+ }
+
+ private static TServerInstance lookupLastServer(KeyExtent extent, SortedMap<Key,Value> tabletsKeyValues) {
+ for (Entry<Key,Value> entry : tabletsKeyValues.entrySet()) {
+ if (entry.getKey().getColumnFamily().compareTo(TabletsSection.LastLocationColumnFamily.NAME) == 0) {
+ return new TServerInstance(entry.getValue(), entry.getKey().getColumnQualifier());
+ }
+ }
+ return null;
+ }
+
+ public Tablet(TabletServer tabletServer, KeyExtent extent, Text location, TabletResourceManager trm, SortedMap<Key,Value> tabletsKeyValues)
+ throws IOException {
+ this(tabletServer, extent, location, trm, lookupLogEntries(tabletServer, extent, tabletsKeyValues), lookupDatafiles(tabletServer,
+ tabletServer.getFileSystem(), extent, tabletsKeyValues), lookupTime(tabletServer.getConfiguration(), extent, tabletsKeyValues), lookupLastServer(
+ extent, tabletsKeyValues), lookupScanFiles(extent, tabletsKeyValues, tabletServer.getFileSystem()), lookupFlushID(extent, tabletsKeyValues),
+ lookupCompactID(extent, tabletsKeyValues));
+ }
+
+ /**
+ * yet another constructor - this one allows us to avoid costly lookups into the Metadata table if we already know the files we need - as at split time
+ */
+ private Tablet(final TabletServer tabletServer, final KeyExtent extent, final Text location, final TabletResourceManager trm,
+ final List<LogEntry> rawLogEntries, final SortedMap<FileRef,DataFileValue> rawDatafiles, String time, final TServerInstance lastLocation,
+ Set<FileRef> scanFiles, long initFlushID, long initCompactID) throws IOException {
+
+ TableConfiguration tblConf = tabletServer.getTableConfiguration(extent);
+ if (null == tblConf) {
+ Tables.clearCache(tabletServer.getInstance());
+ tblConf = tabletServer.getTableConfiguration(extent);
+ Preconditions.checkNotNull(tblConf, "Could not get table configuration for " + extent.getTableId().toString());
+ }
+
+ this.tableConfiguration = tblConf;
+
+ TabletFiles tabletPaths = VolumeUtil.updateTabletVolumes(tabletServer, tabletServer.getLock(), tabletServer.getFileSystem(), extent, new TabletFiles(
+ location.toString(), rawLogEntries, rawDatafiles), ReplicationConfigurationUtil.isEnabled(extent, this.tableConfiguration));
+
+ Path locationPath;
+
+ if (tabletPaths.dir.contains(":")) {
+ locationPath = new Path(tabletPaths.dir.toString());
+ } else {
+ locationPath = tabletServer.getFileSystem().getFullPath(FileType.TABLE, extent.getTableId().toString() + tabletPaths.dir.toString());
+ }
+
+ final List<LogEntry> logEntries = tabletPaths.logEntries;
+ final SortedMap<FileRef,DataFileValue> datafiles = tabletPaths.datafiles;
+
+ this.location = locationPath;
+ this.lastLocation = lastLocation;
+ this.tabletDirectory = tabletPaths.dir;
+
+ this.extent = extent;
+ this.tabletResources = trm;
+
+ this.lastFlushID = initFlushID;
+ this.lastCompactID = initCompactID;
+
+ if (extent.isRootTablet()) {
+ long rtime = Long.MIN_VALUE;
+ for (FileRef ref : datafiles.keySet()) {
+ Path path = ref.path();
+ FileSystem ns = tabletServer.getFileSystem().getVolumeByPath(path).getFileSystem();
+ FileSKVIterator reader = FileOperations.getInstance().openReader(path.toString(), true, ns, ns.getConf(), tabletServer.getTableConfiguration(extent));
+ long maxTime = -1;
+ try {
+
+ while (reader.hasTop()) {
+ maxTime = Math.max(maxTime, reader.getTopKey().getTimestamp());
+ reader.next();
+ }
+
+ } finally {
+ reader.close();
+ }
+
+ if (maxTime > rtime) {
+ time = TabletTime.LOGICAL_TIME_ID + "" + maxTime;
+ rtime = maxTime;
+ }
+ }
+ }
+ if (time == null && datafiles.isEmpty() && extent.equals(RootTable.OLD_EXTENT)) {
+ // recovery... old root tablet has no data, so time doesn't matter:
+ time = TabletTime.LOGICAL_TIME_ID + "" + Long.MIN_VALUE;
+ }
+
+ this.tabletServer = tabletServer;
+ this.logId = tabletServer.createLogId(extent);
+
+ setupDefaultSecurityLabels(extent);
+
+ tabletMemory = new TabletMemory(this);
+ tabletTime = TabletTime.getInstance(time);
+ persistedTime = tabletTime.getTime();
+
+ tableConfiguration.addObserver(configObserver = new ConfigurationObserver() {
+
+ private void reloadConstraints() {
+ log.debug("Reloading constraints for extent: " + extent);
+ constraintChecker.set(new ConstraintChecker(tableConfiguration));
+ }
+
+ @Override
+ public void propertiesChanged() {
+ reloadConstraints();
+
+ try {
+ setupDefaultSecurityLabels(extent);
+ } catch (Exception e) {
+ log.error("Failed to reload default security labels for extent: " + extent.toString());
+ }
+ }
+
+ @Override
+ public void propertyChanged(String prop) {
+ if (prop.startsWith(Property.TABLE_CONSTRAINT_PREFIX.getKey()))
+ reloadConstraints();
+ else if (prop.equals(Property.TABLE_DEFAULT_SCANTIME_VISIBILITY.getKey())) {
+ try {
+ log.info("Default security labels changed for extent: " + extent.toString());
+ setupDefaultSecurityLabels(extent);
+ } catch (Exception e) {
+ log.error("Failed to reload default security labels for extent: " + extent.toString());
+ }
+ }
+
+ }
+
+ @Override
+ public void sessionExpired() {
+ log.debug("Session expired, no longer updating per table props...");
+ }
+
+ });
+
+ tableConfiguration.getNamespaceConfiguration().addObserver(configObserver);
+
+ // Force a load of any per-table properties
+ configObserver.propertiesChanged();
+
+ if (!logEntries.isEmpty()) {
+ log.info("Starting Write-Ahead Log recovery for " + this.extent);
+ // count[0] = entries used on tablet
+ // count[1] = track max time from walog entries wihtout timestamps
+ final long[] count = new long[2];
+ final CommitSession commitSession = getTabletMemory().getCommitSession();
+ count[1] = Long.MIN_VALUE;
+ try {
+ Set<String> absPaths = new HashSet<String>();
+ for (FileRef ref : datafiles.keySet())
+ absPaths.add(ref.path().toString());
+
+ tabletServer.recover(this.getTabletServer().getFileSystem(), extent, tableConfiguration, logEntries, absPaths, new MutationReceiver() {
+ @Override
+ public void receive(Mutation m) {
+ // LogReader.printMutation(m);
+ Collection<ColumnUpdate> muts = m.getUpdates();
+ for (ColumnUpdate columnUpdate : muts) {
+ if (!columnUpdate.hasTimestamp()) {
+ // if it is not a user set timestamp, it must have been set
+ // by the system
+ count[1] = Math.max(count[1], columnUpdate.getTimestamp());
+ }
+ }
+ getTabletMemory().mutate(commitSession, Collections.singletonList(m));
+ count[0]++;
+ }
+ });
+
+ if (count[1] != Long.MIN_VALUE) {
+ tabletTime.useMaxTimeFromWALog(count[1]);
+ }
+ commitSession.updateMaxCommittedTime(tabletTime.getTime());
+
+ if (count[0] == 0) {
+ log.debug("No replayed mutations applied, removing unused entries for " + extent);
+ MetadataTableUtil.removeUnusedWALEntries(getTabletServer(), extent, logEntries, tabletServer.getLock());
+
+ // No replication update to be made because the fact that this tablet didn't use any mutations
+ // from the WAL implies nothing about use of this WAL by other tablets. Do nothing.
+
+ logEntries.clear();
+ } else if (ReplicationConfigurationUtil.isEnabled(extent, tabletServer.getTableConfiguration(extent))) {
+ // The logs are about to be re-used by this tablet, we need to record that they have data for this extent,
+ // but that they may get more data. logEntries is not cleared which will cause the elements
+ // in logEntries to be added to the currentLogs for this Tablet below.
+ //
+ // This update serves the same purpose as an update during a MinC. We know that the WAL was defined
+ // (written when the WAL was opened) but this lets us know there are mutations written to this WAL
+ // that could potentially be replicated. Because the Tablet is using this WAL, we can be sure that
+ // the WAL isn't closed (WRT replication Status) and thus we're safe to update its progress.
+ Status status = StatusUtil.openWithUnknownLength();
+ for (LogEntry logEntry : logEntries) {
+ log.debug("Writing updated status to metadata table for " + logEntry.logSet + " " + ProtobufUtil.toString(status));
+ ReplicationTableUtil.updateFiles(tabletServer, extent, logEntry.logSet, status);
+ }
+ }
+
+ } catch (Throwable t) {
+ if (tableConfiguration.getBoolean(Property.TABLE_FAILURES_IGNORE)) {
+ log.warn("Error recovering from log files: ", t);
+ } else {
+ throw new RuntimeException(t);
+ }
+ }
+ // make some closed references that represent the recovered logs
+ currentLogs = new HashSet<DfsLogger>();
+ for (LogEntry logEntry : logEntries) {
+ for (String log : logEntry.logSet) {
+ currentLogs.add(new DfsLogger(tabletServer.getServerConfig(), log, logEntry.getColumnQualifier().toString()));
+ }
+ }
+
+ log.info("Write-Ahead Log recovery complete for " + this.extent + " (" + count[0] + " mutations applied, " + getTabletMemory().getNumEntries()
+ + " entries created)");
+ }
+
+ String contextName = tableConfiguration.get(Property.TABLE_CLASSPATH);
+ if (contextName != null && !contextName.equals("")) {
+ // initialize context classloader, instead of possibly waiting for it to initialize for a scan
+ // TODO this could hang, causing other tablets to fail to load - ACCUMULO-1292
+ AccumuloVFSClassLoader.getContextManager().getClassLoader(contextName);
+ }
+
+ // do this last after tablet is completely setup because it
+ // could cause major compaction to start
+ datafileManager = new DatafileManager(this, datafiles);
+
+ computeNumEntries();
+
+ getDatafileManager().removeFilesAfterScan(scanFiles);
+
+ // look for hints of a failure on the previous tablet server
+ if (!logEntries.isEmpty() || needsMajorCompaction(MajorCompactionReason.NORMAL)) {
+ // look for any temp files hanging around
+ removeOldTemporaryFiles();
+ }
+
+ log.log(TLevel.TABLET_HIST, extent + " opened");
+ }
+
+ private void removeOldTemporaryFiles() {
+ // remove any temporary files created by a previous tablet server
+ try {
+ for (FileStatus tmp : getTabletServer().getFileSystem().globStatus(new Path(location, "*_tmp"))) {
+ try {
+ log.debug("Removing old temp file " + tmp.getPath());
+ getTabletServer().getFileSystem().delete(tmp.getPath());
+ } catch (IOException ex) {
+ log.error("Unable to remove old temp file " + tmp.getPath() + ": " + ex);
+ }
+ }
+ } catch (IOException ex) {
+ log.error("Error scanning for old temp files in " + location);
+ }
+ }
+
+ private void setupDefaultSecurityLabels(KeyExtent extent) {
+ if (extent.isMeta()) {
+ defaultSecurityLabel = new byte[0];
+ } else {
+ try {
+ ColumnVisibility cv = new ColumnVisibility(tableConfiguration.get(Property.TABLE_DEFAULT_SCANTIME_VISIBILITY));
+ this.defaultSecurityLabel = cv.getExpression();
+ } catch (Exception e) {
+ log.error(e, e);
+ this.defaultSecurityLabel = new byte[0];
+ }
+ }
+ }
+
+ private LookupResult lookup(SortedKeyValueIterator<Key,Value> mmfi, List<Range> ranges, HashSet<Column> columnSet, List<KVEntry> results, long maxResultsSize)
+ throws IOException {
+
+ LookupResult lookupResult = new LookupResult();
+
+ boolean exceededMemoryUsage = false;
+ boolean tabletClosed = false;
+
+ Set<ByteSequence> cfset = null;
+ if (columnSet.size() > 0)
+ cfset = LocalityGroupUtil.families(columnSet);
+
+ for (Range range : ranges) {
+
+ if (exceededMemoryUsage || tabletClosed) {
+ lookupResult.unfinishedRanges.add(range);
+ continue;
+ }
+
+ int entriesAdded = 0;
+
+ try {
+ if (cfset != null)
+ mmfi.seek(range, cfset, true);
+ else
+ mmfi.seek(range, LocalityGroupUtil.EMPTY_CF_SET, false);
+
+ while (mmfi.hasTop()) {
+ Key key = mmfi.getTopKey();
+
+ KVEntry kve = new KVEntry(key, mmfi.getTopValue());
+ results.add(kve);
+ entriesAdded++;
+ lookupResult.bytesAdded += kve.estimateMemoryUsed();
+ lookupResult.dataSize += kve.numBytes();
+
+ exceededMemoryUsage = lookupResult.bytesAdded > maxResultsSize;
+
+ if (exceededMemoryUsage) {
+ addUnfinishedRange(lookupResult, range, key, false);
+ break;
+ }
+
+ mmfi.next();
+ }
+
+ } catch (TooManyFilesException tmfe) {
+ // treat this as a closed tablet, and let the client retry
+ log.warn("Tablet " + getExtent() + " has too many files, batch lookup can not run");
+ handleTabletClosedDuringScan(results, lookupResult, exceededMemoryUsage, range, entriesAdded);
+ tabletClosed = true;
+ } catch (IOException ioe) {
+ if (shutdownInProgress()) {
+ // assume HDFS shutdown hook caused this exception
+ log.debug("IOException while shutdown in progress ", ioe);
+ handleTabletClosedDuringScan(results, lookupResult, exceededMemoryUsage, range, entriesAdded);
+ tabletClosed = true;
+ } else {
+ throw ioe;
+ }
+ } catch (IterationInterruptedException iie) {
+ if (isClosed()) {
+ handleTabletClosedDuringScan(results, lookupResult, exceededMemoryUsage, range, entriesAdded);
+ tabletClosed = true;
+ } else {
+ throw iie;
+ }
+ } catch (TabletClosedException tce) {
+ handleTabletClosedDuringScan(results, lookupResult, exceededMemoryUsage, range, entriesAdded);
+ tabletClosed = true;
+ }
+
+ }
+
+ return lookupResult;
+ }
+
+ private void handleTabletClosedDuringScan(List<KVEntry> results, LookupResult lookupResult, boolean exceededMemoryUsage, Range range, int entriesAdded) {
+ if (exceededMemoryUsage)
+ throw new IllegalStateException("tablet should not exceed memory usage or close, not both");
+
+ if (entriesAdded > 0)
+ addUnfinishedRange(lookupResult, range, results.get(results.size() - 1).getKey(), false);
+ else
+ lookupResult.unfinishedRanges.add(range);
+
+ lookupResult.closed = true;
+ }
+
+ private void addUnfinishedRange(LookupResult lookupResult, Range range, Key key, boolean inclusiveStartKey) {
+ if (range.getEndKey() == null || key.compareTo(range.getEndKey()) < 0) {
+ Range nlur = new Range(new Key(key), inclusiveStartKey, range.getEndKey(), range.isEndKeyInclusive());
+ lookupResult.unfinishedRanges.add(nlur);
+ }
+ }
+
++ public void checkConditions(ConditionChecker checker, Authorizations authorizations, AtomicBoolean iFlag) throws IOException {
++
++ ScanDataSource dataSource = new ScanDataSource(this, authorizations, this.defaultSecurityLabel, iFlag);
++
++ try {
++ SortedKeyValueIterator<Key,Value> iter = new SourceSwitchingIterator(dataSource);
++ checker.check(iter);
++ } catch (IOException ioe) {
++ dataSource.close(true);
++ throw ioe;
++ } finally {
++ // code in finally block because always want
++ // to return mapfiles, even when exception is thrown
++ dataSource.close(false);
++ }
++ }
++
+ public LookupResult lookup(List<Range> ranges, HashSet<Column> columns, Authorizations authorizations, List<KVEntry> results, long maxResultSize,
+ List<IterInfo> ssiList, Map<String,Map<String,String>> ssio, AtomicBoolean interruptFlag) throws IOException {
+
+ if (ranges.size() == 0) {
+ return new LookupResult();
+ }
+
+ ranges = Range.mergeOverlapping(ranges);
+ if (ranges.size() > 1) {
+ Collections.sort(ranges);
+ }
+
+ Range tabletRange = extent.toDataRange();
+ for (Range range : ranges) {
+ // do a test to see if this range falls within the tablet, if it does not
+ // then clip will throw an exception
+ tabletRange.clip(range);
+ }
+
+ ScanDataSource dataSource = new ScanDataSource(this, authorizations, this.defaultSecurityLabel, columns, ssiList, ssio, interruptFlag);
+
+ LookupResult result = null;
+
+ try {
+ SortedKeyValueIterator<Key,Value> iter = new SourceSwitchingIterator(dataSource);
+ result = lookup(iter, ranges, columns, results, maxResultSize);
+ return result;
+ } catch (IOException ioe) {
+ dataSource.close(true);
+ throw ioe;
+ } finally {
+ // code in finally block because always want
+ // to return mapfiles, even when exception is thrown
+ dataSource.close(false);
+
+ synchronized (this) {
+ queryCount += results.size();
+ if (result != null)
+ queryBytes += result.dataSize;
+ }
+ }
+ }
+
+ Batch nextBatch(SortedKeyValueIterator<Key,Value> iter, Range range, int num, Set<Column> columns) throws IOException {
+
+ // log.info("In nextBatch..");
+
+ List<KVEntry> results = new ArrayList<KVEntry>();
+ Key key = null;
+
+ Value value;
+ long resultSize = 0L;
+ long resultBytes = 0L;
+
+ long maxResultsSize = tableConfiguration.getMemoryInBytes(Property.TABLE_SCAN_MAXMEM);
+
+ if (columns.size() == 0) {
+ iter.seek(range, LocalityGroupUtil.EMPTY_CF_SET, false);
+ } else {
+ iter.seek(range, LocalityGroupUtil.families(columns), true);
+ }
+
+ Key continueKey = null;
+ boolean skipContinueKey = false;
+
+ boolean endOfTabletReached = false;
+ while (iter.hasTop()) {
+
+ value = iter.getTopValue();
+ key = iter.getTopKey();
+
+ KVEntry kvEntry = new KVEntry(key, value); // copies key and value
+ results.add(kvEntry);
+ resultSize += kvEntry.estimateMemoryUsed();
+ resultBytes += kvEntry.numBytes();
+
+ if (resultSize >= maxResultsSize || results.size() >= num) {
+ continueKey = new Key(key);
+ skipContinueKey = true;
+ break;
+ }
+
+ iter.next();
+ }
+
+ if (iter.hasTop() == false) {
+ endOfTabletReached = true;
+ }
+
+ if (endOfTabletReached) {
+ continueKey = null;
+ }
+
+ if (endOfTabletReached && results.size() == 0)
+ results = null;
+
+ return new Batch(skipContinueKey, results, continueKey, resultBytes);
+ }
+
+ /**
+ * Determine if a JVM shutdown is in progress.
+ *
+ */
+ boolean shutdownInProgress() {
+ try {
+ Runtime.getRuntime().removeShutdownHook(new Thread(new Runnable() {
+ @Override
+ public void run() {}
+ }));
+ } catch (IllegalStateException ise) {
+ return true;
+ }
+
+ return false;
+ }
+
+ public Scanner createScanner(Range range, int num, Set<Column> columns, Authorizations authorizations, List<IterInfo> ssiList,
+ Map<String,Map<String,String>> ssio, boolean isolated, AtomicBoolean interruptFlag) {
+ // do a test to see if this range falls within the tablet, if it does not
+ // then clip will throw an exception
+ extent.toDataRange().clip(range);
+
+ ScanOptions opts = new ScanOptions(num, authorizations, this.defaultSecurityLabel, columns, ssiList, ssio, interruptFlag, isolated);
+ return new Scanner(this, range, opts);
+ }
+
+ DataFileValue minorCompact(VolumeManager fs, InMemoryMap memTable, FileRef tmpDatafile, FileRef newDatafile, FileRef mergeFile, boolean hasQueueTime,
+ long queued, CommitSession commitSession, long flushId, MinorCompactionReason mincReason) {
+ boolean failed = false;
+ long start = System.currentTimeMillis();
+ timer.incrementStatusMinor();
+
+ long count = 0;
+
+ try {
+ Span span = Trace.start("write");
+ CompactionStats stats;
+ try {
+ count = memTable.getNumEntries();
+
+ DataFileValue dfv = null;
+ if (mergeFile != null)
+ dfv = getDatafileManager().getDatafileSizes().get(mergeFile);
+
+ MinorCompactor compactor = new MinorCompactor(tabletServer, this, memTable, mergeFile, dfv, tmpDatafile, mincReason, tableConfiguration);
+ stats = compactor.call();
+ } finally {
+ span.stop();
+ }
+ span = Trace.start("bringOnline");
+ try {
+ getDatafileManager().bringMinorCompactionOnline(tmpDatafile, newDatafile, mergeFile, new DataFileValue(stats.getFileSize(), stats.getEntriesWritten()),
+ commitSession, flushId);
+ } finally {
+ span.stop();
+ }
+ return new DataFileValue(stats.getFileSize(), stats.getEntriesWritten());
+ } catch (Exception e) {
+ failed = true;
+ throw new RuntimeException(e);
+ } catch (Error e) {
+ // Weird errors like "OutOfMemoryError" when trying to create the thread for the compaction
+ failed = true;
+ throw new RuntimeException(e);
+ } finally {
+ try {
+ getTabletMemory().finalizeMinC();
+ } catch (Throwable t) {
+ log.error("Failed to free tablet memory", t);
+ }
+
+ if (!failed) {
+ lastMinorCompactionFinishTime = System.currentTimeMillis();
+ }
+ Metrics minCMetrics = getTabletServer().getMinCMetrics();
+ if (minCMetrics.isEnabled())
+ minCMetrics.add(TabletServerMinCMetrics.MINC, (lastMinorCompactionFinishTime - start));
+ if (hasQueueTime) {
+ timer.updateTime(Operation.MINOR, queued, start, count, failed);
+ if (minCMetrics.isEnabled())
+ minCMetrics.add(TabletServerMinCMetrics.QUEUE, (start - queued));
+ } else
+ timer.updateTime(Operation.MINOR, start, count, failed);
+ }
+ }
+
+ private synchronized MinorCompactionTask prepareForMinC(long flushId, MinorCompactionReason mincReason) {
+ CommitSession oldCommitSession = getTabletMemory().prepareForMinC();
+ otherLogs = currentLogs;
+ currentLogs = new HashSet<DfsLogger>();
+
+ FileRef mergeFile = null;
+ if (mincReason != MinorCompactionReason.RECOVERY) {
+ mergeFile = getDatafileManager().reserveMergingMinorCompactionFile();
+ }
+
+ double tracePercent = tabletServer.getConfiguration().getFraction(Property.TSERV_MINC_TRACE_PERCENT);
+
+ return new MinorCompactionTask(this, mergeFile, oldCommitSession, flushId, mincReason, tracePercent);
+
+ }
+
+ public void flush(long tableFlushID) {
+ boolean updateMetadata = false;
+ boolean initiateMinor = false;
+
+ try {
+
+ synchronized (this) {
+
+ // only want one thing at a time to update flush ID to ensure that metadata table and tablet in memory state are consistent
+ if (updatingFlushID)
+ return;
+
+ if (lastFlushID >= tableFlushID)
+ return;
+
+ if (isClosing() || isClosed() || getTabletMemory().memoryReservedForMinC())
+ return;
+
+ if (getTabletMemory().getMemTable().getNumEntries() == 0) {
+ lastFlushID = tableFlushID;
+ updatingFlushID = true;
+ updateMetadata = true;
+ } else
+ initiateMinor = true;
+ }
+
+ if (updateMetadata) {
+ // if multiple threads were allowed to update this outside of a sync block, then it would be
+ // a race condition
+ MetadataTableUtil.updateTabletFlushID(extent, tableFlushID, tabletServer, getTabletServer().getLock());
+ } else if (initiateMinor)
+ initiateMinorCompaction(tableFlushID, MinorCompactionReason.USER);
+
+ } finally {
+ if (updateMetadata) {
+ synchronized (this) {
+ updatingFlushID = false;
+ this.notifyAll();
+ }
+ }
+ }
+
+ }
+
+ public boolean initiateMinorCompaction(MinorCompactionReason mincReason) {
+ if (isClosed()) {
+ // don't bother trying to get flush id if closed... could be closed after this check but that is ok... just trying to cut down on uneeded log messages....
+ return false;
+ }
+
+ // get the flush id before the new memmap is made available for write
+ long flushId;
+ try {
+ flushId = getFlushID();
+ } catch (NoNodeException e) {
+ log.info("Asked to initiate MinC when there was no flush id " + getExtent() + " " + e.getMessage());
+ return false;
+ }
+ return initiateMinorCompaction(flushId, mincReason);
+ }
+
+ public boolean minorCompactNow(MinorCompactionReason mincReason) {
+ long flushId;
+ try {
+ flushId = getFlushID();
+ } catch (NoNodeException e) {
+ log.info("Asked to initiate MinC when there was no flush id " + getExtent() + " " + e.getMessage());
+ return false;
+ }
+ MinorCompactionTask mct = createMinorCompactionTask(flushId, mincReason);
+ if (mct == null)
+ return false;
+ mct.run();
+ return true;
+ }
+
+ boolean initiateMinorCompaction(long flushId, MinorCompactionReason mincReason) {
+ MinorCompactionTask mct = createMinorCompactionTask(flushId, mincReason);
+ if (mct == null)
+ return false;
+ getTabletResources().executeMinorCompaction(mct);
+ return true;
+ }
+
+ private MinorCompactionTask createMinorCompactionTask(long flushId, MinorCompactionReason mincReason) {
+ MinorCompactionTask mct;
+ long t1, t2;
+
+ StringBuilder logMessage = null;
+
+ try {
+ synchronized (this) {
+ t1 = System.currentTimeMillis();
+
+ if (isClosing() || isClosed() || majorCompactionState == CompactionState.WAITING_TO_START || getTabletMemory().memoryReservedForMinC()
+ || getTabletMemory().getMemTable().getNumEntries() == 0 || updatingFlushID) {
+
+ logMessage = new StringBuilder();
+
+ logMessage.append(extent.toString());
+ logMessage.append(" closeState " + closeState);
+ logMessage.append(" majorCompactionState " + majorCompactionState);
+ if (getTabletMemory() != null)
+ logMessage.append(" tabletMemory.memoryReservedForMinC() " + getTabletMemory().memoryReservedForMinC());
+ if (getTabletMemory() != null && getTabletMemory().getMemTable() != null)
+ logMessage.append(" tabletMemory.getMemTable().getNumEntries() " + getTabletMemory().getMemTable().getNumEntries());
+ logMessage.append(" updatingFlushID " + updatingFlushID);
+
+ return null;
+ }
+
+ mct = prepareForMinC(flushId, mincReason);
+ t2 = System.currentTimeMillis();
+ }
+ } finally {
+ // log outside of sync block
+ if (logMessage != null && log.isDebugEnabled())
+ log.debug(logMessage);
+ }
+
+ log.debug(String.format("MinC initiate lock %.2f secs", (t2 - t1) / 1000.0));
+ return mct;
+ }
+
+ public long getFlushID() throws NoNodeException {
+ try {
+ String zTablePath = Constants.ZROOT + "/" + tabletServer.getInstance().getInstanceID() + Constants.ZTABLES + "/" + extent.getTableId()
+ + Constants.ZTABLE_FLUSH_ID;
+ return Long.parseLong(new String(ZooReaderWriter.getInstance().getData(zTablePath, null), UTF_8));
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ } catch (NumberFormatException nfe) {
+ throw new RuntimeException(nfe);
+ } catch (KeeperException ke) {
+ if (ke instanceof NoNodeException) {
+ throw (NoNodeException) ke;
+ } else {
+ throw new RuntimeException(ke);
+ }
+ }
+ }
+
+ long getCompactionCancelID() {
+ String zTablePath = Constants.ZROOT + "/" + tabletServer.getInstance().getInstanceID() + Constants.ZTABLES + "/" + extent.getTableId()
+ + Constants.ZTABLE_COMPACT_CANCEL_ID;
+
+ try {
+ return Long.parseLong(new String(ZooReaderWriter.getInstance().getData(zTablePath, null), UTF_8));
+ } catch (KeeperException e) {
+ throw new RuntimeException(e);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public Pair<Long,UserCompactionConfig> getCompactionID() throws NoNodeException {
+ try {
+ String zTablePath = Constants.ZROOT + "/" + tabletServer.getInstance().getInstanceID() + Constants.ZTABLES + "/" + extent.getTableId()
+ + Constants.ZTABLE_COMPACT_ID;
+
+ String[] tokens = new String(ZooReaderWriter.getInstance().getData(zTablePath, null), UTF_8).split(",");
+ long compactID = Long.parseLong(tokens[0]);
+
+ UserCompactionConfig compactionConfig = new UserCompactionConfig();
+
+ if (tokens.length > 1) {
+ Hex hex = new Hex();
+ ByteArrayInputStream bais = new ByteArrayInputStream(hex.decode(tokens[1].split("=")[1].getBytes(UTF_8)));
+ DataInputStream dis = new DataInputStream(bais);
+
+ try {
+ compactionConfig.readFields(dis);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ KeyExtent ke = new KeyExtent(extent.getTableId(), compactionConfig.getEndRow(), compactionConfig.getStartRow());
+
+ if (!ke.overlaps(extent)) {
+ // only use iterators if compaction range overlaps
+ compactionConfig = new UserCompactionConfig();
+ }
+ }
+
+ return new Pair<Long,UserCompactionConfig>(compactID, compactionConfig);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ } catch (NumberFormatException nfe) {
+ throw new RuntimeException(nfe);
+ } catch (KeeperException ke) {
+ if (ke instanceof NoNodeException) {
+ throw (NoNodeException) ke;
+ } else {
+ throw new RuntimeException(ke);
+ }
+ } catch (DecoderException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private synchronized CommitSession finishPreparingMutations(long time) {
+ if (writesInProgress < 0) {
+ throw new IllegalStateException("waitingForLogs < 0 " + writesInProgress);
+ }
+
+ if (isClosed() || getTabletMemory() == null) {
+ return null;
+ }
+
+ writesInProgress++;
+ CommitSession commitSession = getTabletMemory().getCommitSession();
+ commitSession.incrementCommitsInProgress();
+ commitSession.updateMaxCommittedTime(time);
+ return commitSession;
+ }
+
+ public void checkConstraints() {
+ ConstraintChecker cc = constraintChecker.get();
+
+ if (cc.classLoaderChanged()) {
+ ConstraintChecker ncc = new ConstraintChecker(tableConfiguration);
+ constraintChecker.compareAndSet(cc, ncc);
+ }
+ }
+
+ public CommitSession prepareMutationsForCommit(TservConstraintEnv cenv, List<Mutation> mutations) throws TConstraintViolationException {
+
+ ConstraintChecker cc = constraintChecker.get();
+
+ List<Mutation> violators = null;
+ Violations violations = new Violations();
+ cenv.setExtent(extent);
+ for (Mutation mutation : mutations) {
+ Violations more = cc.check(cenv, mutation);
+ if (more != null) {
+ violations.add(more);
+ if (violators == null)
+ violators = new ArrayList<Mutation>();
+ violators.add(mutation);
+ }
+ }
+
+ long time = tabletTime.setUpdateTimes(mutations);
+
+ if (!violations.isEmpty()) {
+
+ HashSet<Mutation> violatorsSet = new HashSet<Mutation>(violators);
+ ArrayList<Mutation> nonViolators = new ArrayList<Mutation>();
+
+ for (Mutation mutation : mutations) {
+ if (!violatorsSet.contains(mutation)) {
+ nonViolators.add(mutation);
+ }
+ }
+
+ CommitSession commitSession = null;
+
+ if (nonViolators.size() > 0) {
+ // if everything is a violation, then it is expected that
+ // code calling this will not log or commit
+ commitSession = finishPreparingMutations(time);
+ if (commitSession == null)
+ return null;
+ }
+
+ throw new TConstraintViolationException(violations, violators, nonViolators, commitSession);
+ }
+
+ return finishPreparingMutations(time);
+ }
+
+ @Override
+ public synchronized void abortCommit(CommitSession commitSession, List<Mutation> value) {
+ if (writesInProgress <= 0) {
+ throw new IllegalStateException("waitingForLogs <= 0 " + writesInProgress);
+ }
+
+ if (isCloseComplete() || getTabletMemory() == null) {
+ throw new IllegalStateException("aborting commit when tablet is closed");
+ }
+
+ commitSession.decrementCommitsInProgress();
+ writesInProgress--;
+ if (writesInProgress == 0)
+ this.notifyAll();
+ }
+
+ @Override
+ public void commit(CommitSession commitSession, List<Mutation> mutations) {
+
+ int totalCount = 0;
+ long totalBytes = 0;
+
+ // write the mutation to the in memory table
+ for (Mutation mutation : mutations) {
+ totalCount += mutation.size();
+ totalBytes += mutation.numBytes();
+ }
+
+ getTabletMemory().mutate(commitSession, mutations);
+
+ synchronized (this) {
+ if (writesInProgress < 1) {
+ throw new IllegalStateException("commiting mutations after logging, but not waiting for any log messages");
+ }
+
+ if (isCloseComplete()) {
+ throw new IllegalStateException("tablet closed with outstanding messages to the logger");
+ }
+
+ getTabletMemory().updateMemoryUsageStats();
+
+ // decrement here in case an exception is thrown below
+ writesInProgress--;
+ if (writesInProgress == 0)
+ this.notifyAll();
+
+ commitSession.decrementCommitsInProgress();
+
+ numEntries += totalCount;
+ numEntriesInMemory += totalCount;
+ ingestCount += totalCount;
+ ingestBytes += totalBytes;
+ }
+ }
+
+ /**
+ * Closes the mapfiles associated with a Tablet. If saveState is true, a minor compaction is performed.
+ */
+ public void close(boolean saveState) throws IOException {
+ initiateClose(saveState, false, false);
+ completeClose(saveState, true);
+ }
+
+ void initiateClose(boolean saveState, boolean queueMinC, boolean disableWrites) {
+
+ if (!saveState && queueMinC) {
+ throw new IllegalArgumentException("Not saving state on close and requesting minor compactions queue does not make sense");
+ }
+
+ log.debug("initiateClose(saveState=" + saveState + " queueMinC=" + queueMinC + " disableWrites=" + disableWrites + ") " + getExtent());
+
+ MinorCompactionTask mct = null;
+
+ synchronized (this) {
+ if (isClosed() || isClosing()) {
+ String msg = "Tablet " + getExtent() + " already " + closeState;
+ throw new IllegalStateException(msg);
+ }
+
+ // enter the closing state, no splits, minor, or major compactions can start
+ // should cause running major compactions to stop
+ closeState = CloseState.CLOSING;
+ this.notifyAll();
+
+ // determines if inserts and queries can still continue while minor compacting
+ if (disableWrites) {
+ closeState = CloseState.CLOSED;
+ }
+
+ // wait for major compactions to finish, setting closing to
+ // true should cause any running major compactions to abort
+ while (isMajorCompactionRunning()) {
+ try {
+ this.wait(50);
+ } catch (InterruptedException e) {
+ log.error(e.toString());
+ }
+ }
+
+ while (updatingFlushID) {
+ try {
+ this.wait(50);
+ } catch (InterruptedException e) {
+ log.error(e.toString());
+ }
+ }
+
+ if (!saveState || getTabletMemory().getMemTable().getNumEntries() == 0) {
+ return;
+ }
+
+ getTabletMemory().waitForMinC();
+
+ try {
+ mct = prepareForMinC(getFlushID(), MinorCompactionReason.CLOSE);
+ } catch (NoNodeException e) {
+ throw new RuntimeException(e);
+ }
+
+ if (queueMinC) {
+ getTabletResources().executeMinorCompaction(mct);
+ return;
+ }
+
+ }
+
+ // do minor compaction outside of synch block so that tablet can be read and written to while
+ // compaction runs
+ mct.run();
+ }
+
+ private boolean closeCompleting = false;
+
+ synchronized void completeClose(boolean saveState, boolean completeClose) throws IOException {
+
+ if (!isClosing() || isCloseComplete() || closeCompleting) {
+ throw new IllegalStateException("closeState = " + closeState);
+ }
+
+ log.debug("completeClose(saveState=" + saveState + " completeClose=" + completeClose + ") " + getExtent());
+
+ // ensure this method is only called once, also guards against multiple
+ // threads entering the method at the same time
+ closeCompleting = true;
+ closeState = CloseState.CLOSED;
+
+ // modify dataSourceDeletions so scans will try to switch data sources and fail because the tablet is closed
+ dataSourceDeletions.incrementAndGet();
+
+ for (ScanDataSource activeScan : activeScans) {
+ activeScan.interrupt();
+ }
+
+ // wait for reads and writes to complete
+ while (writesInProgress > 0 || activeScans.size() > 0) {
+ try {
+ this.wait(50);
+ } catch (InterruptedException e) {
+ log.error(e.toString());
+ }
+ }
+
+ getTabletMemory().waitForMinC();
+
+ if (saveState && getTabletMemory().getMemTable().getNumEntries() > 0) {
+ try {
+ prepareForMinC(getFlushID(), MinorCompactionReason.CLOSE).run();
+ } catch (NoNodeException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ if (saveState) {
+ // at this point all tablet data is flushed, so do a consistency check
+ RuntimeException err = null;
+ for (int i = 0; i < 5; i++) {
+ try {
+ closeConsistencyCheck();
+ err = null;
+ } catch (RuntimeException t) {
+ err = t;
+ log.error("Consistency check fails, retrying " + t);
+ UtilWaitThread.sleep(500);
+ }
+ }
+ if (err != null) {
+ ProblemReports.getInstance(tabletServer)
+ .report(new ProblemReport(extent.getTableId().toString(), ProblemType.TABLET_LOAD, this.extent.toString(), err));
+ log.error("Tablet closed consistency check has failed for " + this.extent + " giving up and closing");
+ }
+ }
+
+ try {
+ getTabletMemory().getMemTable().delete(0);
+ } catch (Throwable t) {
+ log.error("Failed to delete mem table : " + t.getMessage(), t);
+ }
+
+ getTabletMemory().close();
+
+ // close map files
+ getTabletResources().close();
+
+ log.log(TLevel.TABLET_HIST, extent + " closed");
+
+ tableConfiguration.getNamespaceConfiguration().removeObserver(configObserver);
+ tableConfiguration.removeObserver(configObserver);
+
+ if (completeClose)
+ closeState = CloseState.COMPLETE;
+ }
+
+ private void closeConsistencyCheck() {
+
+ if (getTabletMemory().getMemTable().getNumEntries() != 0) {
+ String msg = "Closed tablet " + extent + " has " + getTabletMemory().getMemTable().getNumEntries() + " entries in memory";
+ log.error(msg);
+ throw new RuntimeException(msg);
+ }
+
+ if (getTabletMemory().memoryReservedForMinC()) {
+ String msg = "Closed tablet " + extent + " has minor compacting memory";
+ log.error(msg);
+ throw new RuntimeException(msg);
+ }
+
+ try {
+ Pair<List<LogEntry>,SortedMap<FileRef,DataFileValue>> fileLog = MetadataTableUtil.getFileAndLogEntries(tabletServer, extent);
+
+ if (fileLog.getFirst().size() != 0) {
+ String msg = "Closed tablet " + extent + " has walog entries in " + MetadataTable.NAME + " " + fileLog.getFirst();
+ log.error(msg);
+ throw new RuntimeException(msg);
+ }
+
+ if (extent.isRootTablet()) {
+ if (!fileLog.getSecond().keySet().equals(getDatafileManager().getDatafileSizes().keySet())) {
+ String msg = "Data file in " + RootTable.NAME + " differ from in memory data " + extent + " " + fileLog.getSecond().keySet() + " "
+ + getDatafileManager().getDatafileSizes().keySet();
+ log.error(msg);
+ throw new RuntimeException(msg);
+ }
+ } else {
+ if (!fileLog.getSecond().equals(getDatafileManager().getDatafileSizes())) {
+ String msg = "Data file in " + MetadataTable.NAME + " differ from in memory data " + extent + " " + fileLog.getSecond() + " "
+ + getDatafileManager().getDatafileSizes();
+ log.error(msg);
+ throw new RuntimeException(msg);
+ }
+ }
+
+ } catch (Exception e) {
+ String msg = "Failed to do close consistency check for tablet " + extent;
+ log.error(msg, e);
+ throw new RuntimeException(msg, e);
+
+ }
+
+ if (otherLogs.size() != 0 || currentLogs.size() != 0) {
+ String msg = "Closed tablet " + extent + " has walog entries in memory currentLogs = " + currentLogs + " otherLogs = " + otherLogs;
+ log.error(msg);
+ throw new RuntimeException(msg);
+ }
+
+ // TODO check lastFlushID and lostCompactID - ACCUMULO-1290
+ }
+
+ /**
+ * Returns a Path object representing the tablet's location on the DFS.
+ *
+ * @return location
+ */
+ public Path getLocation() {
+ return location;
+ }
+
+ public synchronized boolean initiateMajorCompaction(MajorCompactionReason reason) {
+
+ if (isClosing() || isClosed() || !needsMajorCompaction(reason) || isMajorCompactionRunning() || majorCompactionQueued.contains(reason)) {
+ return false;
+ }
+
+ majorCompactionQueued.add(reason);
+
+ getTabletResources().executeMajorCompaction(getExtent(), new CompactionRunner(this, reason));
+
+ return false;
+ }
+
+ /**
+ * Returns true if a major compaction should be performed on the tablet.
+ *
+ */
+ public boolean needsMajorCompaction(MajorCompactionReason reason) {
+ if (isMajorCompactionRunning())
+ return false;
+ if (reason == MajorCompactionReason.CHOP || reason == MajorCompactionReason.USER)
+ return true;
+ return getTabletResources().needsMajorCompaction(getDatafileManager().getDatafileSizes(), reason);
+ }
+
+ /**
+ * Returns an int representing the total block size of the files served by this tablet.
+ *
+ * @return size
+ */
+ // this is the size of just the files
+ public long estimateTabletSize() {
+ long size = 0L;
+
+ for (DataFileValue sz : getDatafileManager().getDatafileSizes().values())
+ size += sz.getSize();
+
+ return size;
+ }
+
+ private boolean sawBigRow = false;
+ private long timeOfLastMinCWhenBigFreakinRowWasSeen = 0;
+ private long timeOfLastImportWhenBigFreakinRowWasSeen = 0;
+ private long splitCreationTime;
+
+ private SplitRowSpec findSplitRow(Collection<FileRef> files) {
+
+ // never split the root tablet
+ // check if we already decided that we can never split
+ // check to see if we're big enough to split
+
+ long splitThreshold = tableConfiguration.getMemoryInBytes(Property.TABLE_SPLIT_THRESHOLD);
+ long maxEndRow = tableConfiguration.getMemoryInBytes(Property.TABLE_MAX_END_ROW_SIZE);
+
+ if (extent.isRootTablet() || estimateTabletSize() <= splitThreshold) {
+ return null;
+ }
+
+ // have seen a big row before, do not bother checking unless a minor compaction or map file import has occurred.
+ if (sawBigRow) {
+ if (timeOfLastMinCWhenBigFreakinRowWasSeen != lastMinorCompactionFinishTime || timeOfLastImportWhenBigFreakinRowWasSeen != lastMapFileImportTime) {
+ // a minor compaction or map file import has occurred... check again
+ sawBigRow = false;
+ } else {
+ // nothing changed, do not split
+ return null;
+ }
+ }
+
+ SortedMap<Double,Key> keys = null;
+
+ try {
+ // we should make .25 below configurable
+ keys = FileUtil.findMidPoint(getTabletServer().getFileSystem(), getTabletServer().getConfiguration(), extent.getPrevEndRow(), extent.getEndRow(),
+ FileUtil.toPathStrings(files), .25);
+ } catch (IOException e) {
+ log.error("Failed to find midpoint " + e.getMessage());
+ return null;
+ }
+
+ // check to see if one row takes up most of the tablet, in which case we can not split
+ try {
+
+ Text lastRow;
+ if (extent.getEndRow() == null) {
+ Key lastKey = (Key) FileUtil.findLastKey(getTabletServer().getFileSystem(), getTabletServer().getConfiguration(), files);
+ lastRow = lastKey.getRow();
+ } else {
+ lastRow = extent.getEndRow();
+ }
+
+ // We expect to get a midPoint for this set of files. If we don't get one, we have a problem.
+ final Key mid = keys.get(.5);
+ if (null == mid) {
+ throw new IllegalStateException("Could not determine midpoint for files");
+ }
+
+ // check to see that the midPoint is not equal to the end key
+ if (mid.compareRow(lastRow) == 0) {
+ if (keys.firstKey() < .5) {
+ Key candidate = keys.get(keys.firstKey());
+ if (candidate.getLength() > maxEndRow) {
+ log.warn("Cannot split tablet " + extent + ", selected split point too long. Length : " + candidate.getLength());
+
+ sawBigRow = true;
+ timeOfLastMinCWhenBigFreakinRowWasSeen = lastMinorCompactionFinishTime;
+ timeOfLastImportWhenBigFreakinRowWasSeen = lastMapFileImportTime;
+
+ return null;
+ }
+ if (candidate.compareRow(lastRow) != 0) {
+ // we should use this ratio in split size estimations
+ if (log.isTraceEnabled())
+ log.trace(String.format("Splitting at %6.2f instead of .5, row at .5 is same as end row%n", keys.firstKey()));
+ return new SplitRowSpec(keys.firstKey(), candidate.getRow());
+ }
+
+ }
+
+ log.warn("Cannot split tablet " + extent + " it contains a big row : " + lastRow);
+
+ sawBigRow = true;
+ timeOfLastMinCWhenBigFreakinRowWasSeen = lastMinorCompactionFinishTime;
+ timeOfLastImportWhenBigFreakinRowWasSeen = lastMapFileImportTime;
+
+ return null;
+ }
+
+ Text text = mid.getRow();
+ SortedMap<Double,Key> firstHalf = keys.headMap(.5);
+ if (firstHalf.size() > 0) {
+ Text beforeMid = firstHalf.get(firstHalf.lastKey()).getRow();
+ Text shorter = new Text();
+ int trunc = longestCommonLength(text, beforeMid);
+ shorter.set(text.getBytes(), 0, Math.min(text.getLength(), trunc + 1));
+ text = shorter;
+ }
+
+ if (text.getLength() > maxEndRow) {
+ log.warn("Cannot split tablet " + extent + ", selected split point too long. Length : " + text.getLength());
+
+ sawBigRow = true;
+ timeOfLastMinCWhenBigFreakinRowWasSeen = lastMinorCompactionFinishTime;
+ timeOfLastImportWhenBigFreakinRowWasSeen = lastMapFileImportTime;
+
+ return null;
+ }
+
+ return new SplitRowSpec(.5, text);
+ } catch (IOException e) {
+ // don't split now, but check again later
+ log.error("Failed to find lastkey " + e.getMessage());
+ return null;
+ }
+
+ }
+
+ private static int longestCommonLength(Text text, Text beforeMid) {
+ int common = 0;
+ while (common < text.getLength() && common < beforeMid.getLength() && text.getBytes()[common] == beforeMid.getBytes()[common]) {
+ common++;
+ }
+ return common;
+ }
+
+ private Map<FileRef,Pair<Key,Key>> getFirstAndLastKeys(SortedMap<FileRef,DataFileValue> allFiles) throws IOException {
+ Map<FileRef,Pair<Key,Key>> result = new HashMap<FileRef,Pair<Key,Key>>();
+ FileOperations fileFactory = FileOperations.getInstance();
+ VolumeManager fs = getTabletServer().getFileSystem();
+ for (Entry<FileRef,DataFileValue> entry : allFiles.entrySet()) {
+ FileRef file = entry.getKey();
+ FileSystem ns = fs.getVolumeByPath(file.path()).getFileSystem();
+ FileSKVIterator openReader = fileFactory.openReader(file.path().toString(), true, ns, ns.getConf(), this.getTableConfiguration());
+ try {
+ Key first = openReader.getFirstKey();
+ Key last = openReader.getLastKey();
+ result.put(file, new Pair<Key,Key>(first, last));
+ } finally {
+ openReader.close();
+ }
+ }
+ return result;
+ }
+
+ List<FileRef> findChopFiles(KeyExtent extent, Map<FileRef,Pair<Key,Key>> firstAndLastKeys, Collection<FileRef> allFiles) throws IOException {
+ List<FileRef> result = new ArrayList<FileRef>();
+ if (firstAndLastKeys == null) {
+ result.addAll(allFiles);
+ return result;
+ }
+
+ for (FileRef file : allFiles) {
+ Pair<Key,Key> pair = firstAndLastKeys.get(file);
+ if (pair == null) {
+ // file was created or imported after we obtained the first and last keys... there
+ // are a few options here... throw an exception which will cause the compaction to
+ // retry and also cause ugly error message that the admin has to ignore... could
+ // go get the first and last key, but this code is called while the tablet lock
+ // is held... or just compact the file....
+ result.add(file);
+ } else {
+ Key first = pair.getFirst();
+ Key last = pair.getSecond();
+ // If first and last are null, it's an empty file. Add it to the compact set so it goes away.
+ if ((first == null && last == null) || (first != null && !extent.contains(first.getRow())) || (last != null && !extent.contains(last.getRow()))) {
+ result.add(file);
+ }
+ }
+ }
+ return result;
+ }
+
+ /**
+ * Returns true if this tablet needs to be split
+ *
+ */
+ public synchronized boolean needsSplit() {
+ if (isClosing() || isClosed())
+ return false;
+ return findSplitRow(getDatafileManager().getFiles()) != null;
+ }
+
+ // BEGIN PRIVATE METHODS RELATED TO MAJOR COMPACTION
+
+ private boolean isCompactionEnabled() {
+ return !isClosing() && !getTabletServer().isMajorCompactionDisabled();
+ }
+
+ private CompactionStats _majorCompact(MajorCompactionReason reason) throws IOException, CompactionCanceledException {
+
+ long t1, t2, t3;
+
+ Pair<Long,UserCompactionConfig> compactionId = null;
+ CompactionStrategy strategy = null;
+ Map<FileRef,Pair<Key,Key>> firstAndLastKeys = null;
+
+ if (reason == MajorCompactionReason.USER) {
+ try {
+ compactionId = getCompactionID();
+ strategy = createCompactionStrategy(compactionId.getSecond().getCompactionStrategy());
+ } catch (NoNodeException e) {
+ throw new RuntimeException(e);
+ }
+ } else if (reason == MajorCompactionReason.NORMAL || reason == MajorCompactionReason.IDLE) {
+ strategy = Property.createTableInstanceFromPropertyName(tableConfiguration, Property.TABLE_COMPACTION_STRATEGY, CompactionStrategy.class,
+ new DefaultCompactionStrategy());
+ strategy.init(Property.getCompactionStrategyOptions(tableConfiguration));
+ } else if (reason == MajorCompactionReason.CHOP) {
+ firstAndLastKeys = getFirstAndLastKeys(getDatafileManager().getDatafileSizes());
+ } else {
+ throw new IllegalArgumentException("Unknown compaction reason " + reason);
+ }
+
+ if (strategy != null) {
+ MajorCompactionRequest request = new MajorCompactionRequest(extent, reason, getTabletServer().getFileSystem(), tableConfiguration);
+ request.setFiles(getDatafileManager().getDatafileSizes());
+ strategy.gatherInformation(request);
+ }
+
+ Map<FileRef,DataFileValue> filesToCompact = null;
+
+ int maxFilesToCompact = tableConfiguration.getCount(Property.TSERV_MAJC_THREAD_MAXOPEN);
+
+ CompactionStats majCStats = new CompactionStats();
+ CompactionPlan plan = null;
+
+ boolean propogateDeletes = false;
+ boolean updateCompactionID = false;
+
+ synchronized (this) {
+ // plan all that work that needs to be done in the sync block... then do the actual work
+ // outside the sync block
+
+ t1 = System.currentTimeMillis();
+
+ majorCompactionState = CompactionState.WAITING_TO_START;
+
+ getTabletMemory().waitForMinC();
+
+ t2 = System.currentTimeMillis();
+
+ majorCompactionState = CompactionState.IN_PROGRESS;
+ notifyAll();
+
+ VolumeManager fs = getTabletServer().getFileSystem();
+ if (extent.isRootTablet()) {
+ // very important that we call this before doing major compaction,
+ // otherwise deleted compacted files could possible be brought back
+ // at some point if the file they were compacted to was legitimately
+ // removed by a major compaction
+ RootFiles.cleanupReplacement(fs, fs.listStatus(this.location), false);
+ }
+ SortedMap<FileRef,DataFileValue> allFiles = getDatafileManager().getDatafileSizes();
+ List<FileRef> inputFiles = new ArrayList<FileRef>();
+ if (reason == MajorCompactionReason.CHOP) {
+ // enforce rules: files with keys outside our range need to be compacted
+ inputFiles.addAll(findChopFiles(extent, firstAndLastKeys, allFiles.keySet()));
+ } else {
+ MajorCompactionRequest request = new MajorCompactionRequest(extent, reason, fs, tableConfiguration);
+ request.setFiles(allFiles);
+ plan = strategy.getCompactionPlan(request);
+ if (plan != null) {
+ plan.validate(allFiles.keySet());
+ inputFiles.addAll(plan.inputFiles);
+ }
+ }
+
+ if (inputFiles.isEmpty()) {
+ if (reason == MajorCompactionReason.USER) {
+ if (compactionId.getSecond().getIterators().isEmpty()) {
+ log.debug("No-op major compaction by USER on 0 input files because no iterators present.");
+ lastCompactID = compactionId.getFirst();
+ updateCompactionID = true;
+ } else {
+ log.debug("Major compaction by USER on 0 input files with iterators.");
+ filesToCompact = new HashMap<>();
+ }
+ } else {
+ return majCStats;
+ }
+ } else {
+ // If no original files will exist at the end of the compaction, we do not have to propogate deletes
+ Set<FileRef> droppedFiles = new HashSet<>();
+ droppedFiles.addAll(inputFiles);
+ if (plan != null)
+ droppedFiles.addAll(plan.deleteFiles);
+ propogateDeletes = !(droppedFiles.equals(allFiles.keySet()));
+ log.debug("Major compaction plan: " + plan + " propogate deletes : " + propogateDeletes);
+ filesToCompact = new HashMap<>(allFiles);
+ filesToCompact.keySet().retainAll(inputFiles);
+
+ getDatafileManager().reserveMajorCompactingFiles(filesToCompact.keySet());
+ }
+
+ t3 = System.currentTimeMillis();
+ }
+
+ try {
+
+ log.debug(String.format("MajC initiate lock %.2f secs, wait %.2f secs", (t3 - t2) / 1000.0, (t2 - t1) / 1000.0));
+
+ if (updateCompactionID) {
+ MetadataTableUtil.updateTabletCompactID(extent, compactionId.getFirst(), tabletServer, getTabletServer().getLock());
+ return majCStats;
+ }
+
+ if (!propogateDeletes && compactionId == null) {
+ // compacting everything, so update the compaction id in metadata
+ try {
+ compactionId = getCompactionID();
+ if (compactionId.getSecond().getCompactionStrategy() != null) {
+ compactionId = null;
+ // TODO maybe return unless chop?
+ }
+
+ } catch (NoNodeException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ List<IteratorSetting> compactionIterators = new ArrayList<IteratorSetting>();
+ if (compactionId != null) {
+ if (reason == MajorCompactionReason.USER) {
+ if (getCompactionCancelID() >= compactionId.getFirst()) {
+ // compaction was canceled
+ return majCStats;
+ }
+ compactionIterators = compactionId.getSecond().getIterators();
+
+ synchronized (this) {
+ if (lastCompactID >= compactionId.getFirst())
+ // already compacted
+ return majCStats;
+ }
+ }
+
+ }
+
+ // need to handle case where only one file is being major compacted
+ // ACCUMULO-3645 run loop at least once, even if filesToCompact.isEmpty()
+ do {
+ int numToCompact = maxFilesToCompact;
+
+ if (filesToCompact.size() > maxFilesToCompact && filesToCompact.size() < 2 * maxFilesToCompact) {
+ // on the second to last compaction pass, compact the minimum amount of files possible
+ numToCompact = filesToCompact.size() - maxFilesToCompact + 1;
+ }
+
+ Set<FileRef> smallestFiles = removeSmallest(filesToCompact, numToCompact);
+
+ FileRef fileName = getNextMapFilename((filesToCompact.size() == 0 && !propogateDeletes) ? "A" : "C");
+ FileRef compactTmpName = new FileRef(fileName.path().toString() + "_tmp");
+
+ AccumuloConfiguration tableConf = createTableConfiguration(tableConfiguration, plan);
+
+ Span span = Trace.start("compactFiles");
+ try {
+
+ CompactionEnv cenv = new CompactionEnv() {
+ @Override
+ public boolean isCompactionEnabled() {
+ return Tablet.this.isCompactionEnabled();
+ }
+
+ @Override
+ public IteratorScope getIteratorScope() {
+ return IteratorScope.majc;
+ }
+ };
+
+ HashMap<FileRef,DataFileValue> copy = new HashMap<FileRef,DataFileValue>(getDatafileManager().getDatafileSizes());
+ if (!copy.keySet().containsAll(smallestFiles))
+ throw new IllegalStateException("Cannot find data file values for " + smallestFiles);
+
+ copy.keySet().retainAll(smallestFiles);
+
+ log.debug("Starting MajC " + extent + " (" + reason + ") " + copy.keySet() + " --> " + compactTmpName + " " + compactionIterators);
+
+ // always propagate deletes, unless last batch
+ boolean lastBatch = filesToCompact.isEmpty();
+ Compactor compactor = new Compactor(tabletServer, this, copy, null, compactTmpName, lastBatch ? propogateDeletes : true, cenv, compactionIterators,
+ reason.ordinal(), tableConf);
+
+ CompactionStats mcs = compactor.call();
+
+ span.data("files", "" + smallestFiles.size());
+ span.data("read", "" + mcs.getEntriesRead());
+ span.data("written", "" + mcs.getEntriesWritten());
+ majCStats.add(mcs);
+
+ if (lastBatch && plan != null && plan.deleteFiles != null) {
+ smallestFiles.addAll(plan.deleteFiles);
+ }
+ getDatafileManager().bringMajorCompactionOnline(smallestFiles, compactTmpName, fileName,
+ filesToCompact.size() == 0 && compactionId != null ? compactionId.getFirst() : null,
+ new DataFileValue(mcs.getFileSize(), mcs.getEntriesWritten()));
+
+ // when major compaction produces a file w/ zero entries, it will be deleted... do not want
+ // to add the deleted file
+ if (filesToCompact.size() > 0 && mcs.getEntriesWritten() > 0) {
+ filesToCompact.put(fileName, new DataFileValue(mcs.getFileSize(), mcs.getEntriesWritten()));
+ }
+ } finally {
+ span.stop();
+ }
+
+ } while (filesToCompact.size() > 0);
+ return majCStats;
+ } finally {
+ synchronized (Tablet.this) {
+ getDatafileManager().clearMajorCompactingFile();
+ }
+ }
+ }
+
+ protected AccumuloConfiguration createTableConfiguration(TableConfiguration base, CompactionPlan plan) {
+ if (plan == null || plan.writeParameters == null)
+ return base;
+ WriteParameters p = plan.writeParameters;
+ ConfigurationCopy result = new ConfigurationCopy(base);
+ if (p.getHdfsBlockSize() > 0)
+ result.set(Property.TABLE_FILE_BLOCK_SIZE, "" + p.getHdfsBlockSize());
+ if (p.getBlockSize() > 0)
+ result.set(Property.TABLE_FILE_COMPRESSED_BLOCK_SIZE, "" + p.getBlockSize());
+ if (p.getIndexBlockSize() > 0)
+ result.set(Property.TABLE_FILE_COMPRESSED_BLOCK_SIZE_INDEX, "" + p.getIndexBlockSize());
+ if (p.getCompressType() != null)
+ result.set(Property.TABLE_FILE_COMPRESSION_TYPE, p.getCompressType());
+ if (p.getReplication() != 0)
+ result.set(Property.TABLE_FILE_REPLICATION, "" + p.getReplication());
+ return result;
+ }
+
+ private Set<FileRef> removeSmallest(Map<FileRef,DataFileValue> filesToCompact, int maxFilesToCompact) {
+ // ensure this method works properly when multiple files have the same size
+
+ // short-circuit; also handles zero files case
+ if (filesToCompact.size() <= maxFilesToCompact) {
+ Set<FileRef> smallestFiles = new HashSet<FileRef>(filesToCompact.keySet());
+ filesToCompact.clear();
+ return smallestFiles;
+ }
+
+ PriorityQueue<Pair<FileRef,Long>> fileHeap = new PriorityQueue<Pair<FileRef,Long>>(filesToCompact.size(), new Comparator<Pair<FileRef,Long>>() {
+ @Override
+ public int compare(Pair<FileRef,Long> o1, Pair<FileRef,Long> o2) {
+ if (o1.getSecond() == o2.getSecond())
+ return o1.getFirst().compareTo(o2.getFirst());
+ if (o1.getSecond() < o2.getSecond())
+ return -1;
+ return 1;
+ }
+ });
+
+ for (Iterator<Entry<FileRef,DataFileValue>> iterator = filesToCompact.entrySet().iterator(); iterator.hasNext();) {
+ Entry<FileRef,DataFileValue> entry = iterator.next();
+ fileHeap.add(new Pair<FileRef,Long>(entry.getKey(), entry.getValue().getSize()));
+ }
+
+ Set<FileRef> smallestFiles = new HashSet<FileRef>();
+ while (smallestFiles.size() < maxFilesToCompact && fileHeap.size() > 0) {
+ Pair<FileRef,Long> pair = fileHeap.remove();
+ filesToCompact.remove(pair.getFirst());
+ smallestFiles.add(pair.getFirst());
+ }
+
+ return smallestFiles;
+ }
+
+ // END PRIVATE METHODS RELATED TO MAJOR COMPACTION
+
+ /**
+ * Performs a major compaction on the tablet. If needsSplit() returns true, the tablet is split and a reference to the new tablet is returned.
+ */
+
+ CompactionStats majorCompact(MajorCompactionReason reason, long queued) {
+ CompactionStats majCStats = null;
+ boolean success = false;
+ long start = System.currentTimeMillis();
+
+ timer.incrementStatusMajor();
+
+ synchronized (this) {
+ // check that compaction is still needed - defer to splitting
+ majorCompactionQueued.remove(reason);
+
+ if (isClosing() || isClosed() || !needsMajorCompaction(reason) || isMajorCompactionRunning() || needsSplit()) {
+ return null;
+ }
+
+ majorCompactionState = CompactionState.WAITING_TO_START;
+ }
+
+ Span span = null;
+
+ try {
+ double tracePercent = tabletServer.getConfiguration().getFraction(Property.TSERV_MAJC_TRACE_PERCENT);
+ ProbabilitySampler sampler = new ProbabilitySampler(tracePercent);
+ span = Trace.on("majorCompaction", sampler);
+
+ majCStats = _majorCompact(reason);
+ if (reason == MajorCompactionReason.CHOP) {
+ MetadataTableUtil.chopped(getTabletServer(), getExtent(), this.getTabletServer().getLock());
+ getTabletServer().enqueueMasterMessage(new TabletStatusMessage(TabletLoadState.CHOPPED, extent));
+ }
+ success = true;
+ } catch (CompactionCanceledException cce) {
+ log.debug("Major compaction canceled, extent = " + getExtent());
+ } catch (IOException ioe) {
+ log.error("MajC Failed, extent = " + getExtent(), ioe);
+ } catch (RuntimeException e) {
+ log.error("MajC Unexpected exception, extent = " + getExtent(), e);
+ } finally {
+ // ensure we always reset boolean, even
+ // when an exception is thrown
+ synchronized (this) {
+ majorCompactionState = null;
+ this.notifyAll();
+ }
+
+ if (span != null) {
+ span.data("extent", "" + getExtent());
+ if (majCStats != null) {
+ span.data("read", "" + majCStats.getEntriesRead());
+ span.data("written", "" + majCStats.getEntriesWritten());
+ }
+ span.stop();
+ }
+ }
+ long count = 0;
+ if (majCStats != null)
+ count = majCStats.getEntriesRead();
+ timer.updateTime(Operation.MAJOR, queued, start, count, !success);
+
+ return majCStats;
+ }
+
+ @Override
+ public KeyExtent getExtent() {
+ return extent;
+ }
+
+ synchronized void computeNumEntries() {
+ Collection<DataFileValue> vals = getDatafileManager().getDatafileSizes().values();
+
+ long numEntries = 0;
+
+ for (DataFileValue tableValue : vals) {
+ numEntries += tableValue.getNumEntries();
+ }
+
+ this.numEntriesInMemory = getTabletMemory().getNumEntries();
+ numEntries += getTabletMemory().getNumEntries();
+
+ this.numEntries = numEntries;
+ }
+
+ public long getNumEntries() {
+ return numEntries;
+ }
+
+ public long getNumEntriesInMemory() {
+ return numEntriesInMemory;
+ }
+
+ public synchronized boolean isClosing() {
+ return closeState == CloseState.CLOSING;
+ }
+
+ public synchronized boolean isClosed() {
+ return closeState == CloseState.CLOSED || closeState == CloseState.COMPLETE;
+ }
+
+ public synchronized boolean isCloseComplete() {
+ return closeState == CloseState.COMPLETE;
+ }
+
+ public boolean isMajorCompactionRunning() {
+ return majorCompactionState != null;
+ }
+
+ public boolean isMinorCompactionQueued() {
+ return minorCompactionState == CompactionState.WAITING_TO_START;
+ }
+
+ public boolean isMinorCompactionRunning() {
+ return minorCompactionState == CompactionState.IN_PROGRESS;
+ }
+
+ public boolean isMajorCompactionQueued() {
+ return majorCompactionQueued.size() > 0;
+ }
+
+ public TreeMap<KeyExtent,SplitInfo> split(byte[] sp) throws IOException {
+
+ if (sp != null && extent.getEndRow() != null && extent.getEndRow().equals(new Text(sp))) {
+ throw new IllegalArgumentException();
+ }
+
+ if (sp != null && sp.length > tableConfiguration.getMemoryInBytes(Property.TABLE_MAX_END_ROW_SIZE)) {
+ String msg = "Cannot split tablet " + extent + ", selected split point too long. Length : " + sp.length;
+ log.warn(msg);
+ throw new IOException(msg);
+ }
+
+ if (extent.isRootTablet()) {
+ String msg = "Cannot split root tablet";
+ log.warn(msg);
+ throw new RuntimeException(msg);
+ }
+
+ try {
+ initiateClose(true, false, false);
+ } catch (IllegalStateException ise) {
+ log.debug("File " + extent + " not splitting : " + ise.getMessage());
+ return null;
+ }
+
+ // obtain this info outside of synch block since it will involve opening
+ // the map files... it is ok if the set of map files changes, because
+ // this info is used for optimization... it is ok if map files are missing
+ // from the set... can still query and insert into the tablet while this
+ // map file operation is happening
+ Map<FileRef,FileUtil.FileInfo> firstAndLastRows = FileUtil.tryToGetFirstAndLastRows(getTabletServer().getFileSystem(),
+ getTabletServer().getConfiguration(), getDatafileManager().getFiles());
+
+ synchronized (this) {
+ // java needs tuples ...
+ TreeMap<KeyExtent,SplitInfo> newTablets = new TreeMap<KeyExtent,SplitInfo>();
+
+ long t1 = System.currentTimeMillis();
+ // choose a split point
+ SplitRowSpec splitPoint;
+ if (sp == null)
+ splitPoint = findSplitRow(getDatafileManager().getFiles());
+ else {
+ Text tsp = new Text(sp);
+ splitPoint =
<TRUNCATED>
[2/5] accumulo git commit: Merge branch '1.6' into 1.7
Posted by kt...@apache.org.
http://git-wip-us.apache.org/repos/asf/accumulo/blob/d4882a15/test/src/test/java/org/apache/accumulo/test/ConditionalWriterIT.java
----------------------------------------------------------------------
diff --cc test/src/test/java/org/apache/accumulo/test/ConditionalWriterIT.java
index b7637a6,d32bc43..15e9d5f
--- a/test/src/test/java/org/apache/accumulo/test/ConditionalWriterIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/ConditionalWriterIT.java
@@@ -85,17 -87,16 +91,18 @@@ import org.apache.accumulo.harness.Accu
import org.apache.accumulo.minicluster.impl.MiniAccumuloClusterImpl;
import org.apache.accumulo.test.functional.BadIterator;
import org.apache.accumulo.test.functional.SlowIterator;
-import org.apache.accumulo.trace.instrument.Span;
-import org.apache.accumulo.trace.instrument.Trace;
+import org.apache.accumulo.tracer.TraceDump;
+import org.apache.accumulo.tracer.TraceDump.Printer;
import org.apache.accumulo.tracer.TraceServer;
import org.apache.hadoop.io.Text;
-import org.apache.log4j.Logger;
import org.junit.Assert;
import org.junit.Assume;
+import org.junit.Before;
import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+ import com.google.common.base.Charsets;
import com.google.common.collect.Iterables;
/**
[5/5] accumulo git commit: Merge branch '1.7'
Posted by kt...@apache.org.
Merge branch '1.7'
Conflicts:
core/src/main/java/org/apache/accumulo/core/client/impl/ConditionalWriterImpl.java
core/src/main/java/org/apache/accumulo/core/iterators/IteratorUtil.java
server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/ScanDataSource.java
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/4eadac16
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/4eadac16
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/4eadac16
Branch: refs/heads/master
Commit: 4eadac16bb509f72696dfcf79afa84ecd1cc1adf
Parents: 97ce2a7 d4882a1
Author: Keith Turner <kt...@apache.org>
Authored: Wed Jan 27 20:41:17 2016 -0500
Committer: Keith Turner <kt...@apache.org>
Committed: Wed Jan 27 20:41:17 2016 -0500
----------------------------------------------------------------------
.../core/client/impl/CompressedIterators.java | 14 +-
.../core/client/impl/ConditionalWriterImpl.java | 46 +++++-
.../accumulo/core/iterators/IteratorUtil.java | 89 +++++-----
.../client/impl/ConditionalComparatorTest.java | 53 ++++++
.../tserver/ConditionCheckerContext.java | 164 +++++++++++++++++++
.../apache/accumulo/tserver/TabletServer.java | 79 +++------
.../accumulo/tserver/tablet/ScanDataSource.java | 19 ++-
.../apache/accumulo/tserver/tablet/Tablet.java | 18 ++
.../accumulo/test/ConditionalWriterIT.java | 139 +++++++++++++++-
9 files changed, 506 insertions(+), 115 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/4eadac16/core/src/main/java/org/apache/accumulo/core/client/impl/ConditionalWriterImpl.java
----------------------------------------------------------------------
diff --cc core/src/main/java/org/apache/accumulo/core/client/impl/ConditionalWriterImpl.java
index 546de9c,c7756ad..c9ef101
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/ConditionalWriterImpl.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/ConditionalWriterImpl.java
@@@ -17,6 -17,6 +17,7 @@@
package org.apache.accumulo.core.client.impl;
++import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
import static java.nio.charset.StandardCharsets.UTF_8;
import java.nio.ByteBuffer;
@@@ -74,6 -76,8 +77,7 @@@ import org.apache.accumulo.core.trace.T
import org.apache.accumulo.core.trace.thrift.TInfo;
import org.apache.accumulo.core.util.BadArgumentException;
import org.apache.accumulo.core.util.ByteBufferUtil;
+ import org.apache.accumulo.core.util.NamingThreadFactory;
-import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.accumulo.core.zookeeper.ZooUtil;
import org.apache.accumulo.fate.util.LoggingRunnable;
import org.apache.accumulo.fate.zookeeper.ZooCacheFactory;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/4eadac16/core/src/main/java/org/apache/accumulo/core/iterators/IteratorUtil.java
----------------------------------------------------------------------
diff --cc core/src/main/java/org/apache/accumulo/core/iterators/IteratorUtil.java
index 6523367,c739e56..6fee4e6
--- a/core/src/main/java/org/apache/accumulo/core/iterators/IteratorUtil.java
+++ b/core/src/main/java/org/apache/accumulo/core/iterators/IteratorUtil.java
@@@ -217,10 -205,20 +205,14 @@@ public class IteratorUtil
return loadIterators(scope, source, extent, conf, ssiList, ssio, env, true);
}
- public static <K extends WritableComparable<?>,V extends Writable> SortedKeyValueIterator<K,V> loadIterators(IteratorScope scope,
- SortedKeyValueIterator<K,V> source, KeyExtent extent, AccumuloConfiguration conf, List<IterInfo> ssiList, Map<String,Map<String,String>> ssio,
- IteratorEnvironment env, boolean useAccumuloClassLoader) throws IOException {
- List<IterInfo> iters = new ArrayList<IterInfo>(ssiList);
- Map<String,Map<String,String>> allOptions = new HashMap<String,Map<String,String>>();
-
+ private static void parseIteratorConfiguration(IteratorScope scope, List<IterInfo> iters, Map<String,Map<String,String>> ssio,
+ Map<String,Map<String,String>> allOptions, AccumuloConfiguration conf) {
parseIterConf(scope, iters, allOptions, conf);
+ mergeOptions(ssio, allOptions);
-
- return loadIterators(source, iters, allOptions, env, useAccumuloClassLoader, conf.get(Property.TABLE_CLASSPATH));
+ }
+
+ private static void mergeOptions(Map<String,Map<String,String>> ssio, Map<String,Map<String,String>> allOptions) {
for (Entry<String,Map<String,String>> entry : ssio.entrySet()) {
if (entry.getValue() == null)
continue;
@@@ -233,25 -231,6 +225,24 @@@
}
}
+ public static <K extends WritableComparable<?>,V extends Writable> SortedKeyValueIterator<K,V> loadIterators(IteratorScope scope,
+ SortedKeyValueIterator<K,V> source, KeyExtent extent, AccumuloConfiguration conf, List<IterInfo> ssiList, Map<String,Map<String,String>> ssio,
+ IteratorEnvironment env, boolean useAccumuloClassLoader) throws IOException {
+ List<IterInfo> iters = new ArrayList<IterInfo>(ssiList);
+ Map<String,Map<String,String>> allOptions = new HashMap<String,Map<String,String>>();
+ parseIteratorConfiguration(scope, iters, ssio, allOptions, conf);
+ return loadIterators(source, iters, allOptions, env, useAccumuloClassLoader, conf.get(Property.TABLE_CLASSPATH));
+ }
+
+ public static <K extends WritableComparable<?>,V extends Writable> SortedKeyValueIterator<K,V> loadIterators(IteratorScope scope,
+ SortedKeyValueIterator<K,V> source, KeyExtent extent, AccumuloConfiguration conf, List<IterInfo> ssiList, Map<String,Map<String,String>> ssio,
+ IteratorEnvironment env, boolean useAccumuloClassLoader, String classLoaderContext) throws IOException {
+ List<IterInfo> iters = new ArrayList<IterInfo>(ssiList);
+ Map<String,Map<String,String>> allOptions = new HashMap<String,Map<String,String>>();
+ parseIteratorConfiguration(scope, iters, ssio, allOptions, conf);
+ return loadIterators(source, iters, allOptions, env, useAccumuloClassLoader, classLoaderContext);
+ }
+
- @SuppressWarnings("unchecked")
public static <K extends WritableComparable<?>,V extends Writable> SortedKeyValueIterator<K,V> loadIterators(SortedKeyValueIterator<K,V> source,
Collection<IterInfo> iters, Map<String,Map<String,String>> iterOpts, IteratorEnvironment env, boolean useAccumuloClassLoader, String context)
throws IOException {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/4eadac16/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
----------------------------------------------------------------------
diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
index 42c6579,038d3e8..1b09841
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
@@@ -61,9 -59,7 +61,8 @@@ import org.apache.accumulo.core.client.
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.Durability;
import org.apache.accumulo.core.client.Instance;
+import org.apache.accumulo.core.client.SampleNotPresentException;
import org.apache.accumulo.core.client.impl.CompressedIterators;
- import org.apache.accumulo.core.client.impl.CompressedIterators.IterConfig;
import org.apache.accumulo.core.client.impl.DurabilityImpl;
import org.apache.accumulo.core.client.impl.ScannerImpl;
import org.apache.accumulo.core.client.impl.Tables;
@@@ -236,12 -226,10 +235,11 @@@ import org.apache.accumulo.tserver.tabl
import org.apache.accumulo.tserver.tablet.CompactionInfo;
import org.apache.accumulo.tserver.tablet.CompactionWatcher;
import org.apache.accumulo.tserver.tablet.Compactor;
+import org.apache.accumulo.tserver.tablet.KVEntry;
import org.apache.accumulo.tserver.tablet.ScanBatch;
- import org.apache.accumulo.tserver.tablet.Scanner;
-import org.apache.accumulo.tserver.tablet.SplitInfo;
import org.apache.accumulo.tserver.tablet.Tablet;
import org.apache.accumulo.tserver.tablet.TabletClosedException;
+import org.apache.accumulo.tserver.tablet.TabletData;
import org.apache.commons.collections.map.LRUMap;
import org.apache.hadoop.fs.FSError;
import org.apache.hadoop.fs.FileSystem;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/4eadac16/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/ScanDataSource.java
----------------------------------------------------------------------
diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/ScanDataSource.java
index 8f61b9d,f586e2e..6a4b6e4
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/ScanDataSource.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/ScanDataSource.java
@@@ -22,9 -23,9 +23,10 @@@ import java.util.Collections
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+ import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.accumulo.core.client.admin.SamplerConfiguration;
import org.apache.accumulo.core.data.Column;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Value;
@@@ -67,13 -65,17 +69,17 @@@ class ScanDataSource implements DataSou
private StatsIterator statsIterator;
private final ScanOptions options;
+ private final boolean loadIters;
+
+ private static final Set<Column> EMPTY_COLS = Collections.emptySet();
ScanDataSource(Tablet tablet, Authorizations authorizations, byte[] defaultLabels, HashSet<Column> columnSet, List<IterInfo> ssiList,
- Map<String,Map<String,String>> ssio, AtomicBoolean interruptFlag) {
+ Map<String,Map<String,String>> ssio, AtomicBoolean interruptFlag, SamplerConfiguration samplerConfig, long batchTimeOut, String context) {
this.tablet = tablet;
expectedDeletionCount = tablet.getDataSourceDeletions();
- this.options = new ScanOptions(-1, authorizations, defaultLabels, columnSet, ssiList, ssio, interruptFlag, false);
+ this.options = new ScanOptions(-1, authorizations, defaultLabels, columnSet, ssiList, ssio, interruptFlag, false, samplerConfig, batchTimeOut, context);
this.interruptFlag = interruptFlag;
+ this.loadIters = true;
}
ScanDataSource(Tablet tablet, ScanOptions options) {
@@@ -81,6 -83,15 +87,15 @@@
expectedDeletionCount = tablet.getDataSourceDeletions();
this.options = options;
this.interruptFlag = options.getInterruptFlag();
+ this.loadIters = true;
+ }
+
+ ScanDataSource(Tablet tablet, Authorizations authorizations, byte[] defaultLabels, AtomicBoolean iFlag) {
+ this.tablet = tablet;
+ expectedDeletionCount = tablet.getDataSourceDeletions();
- this.options = new ScanOptions(-1, authorizations, defaultLabels, EMPTY_COLS, null, null, iFlag, false);
++ this.options = new ScanOptions(-1, authorizations, defaultLabels, EMPTY_COLS, null, null, iFlag, false, null, -1, null);
+ this.interruptFlag = iFlag;
+ this.loadIters = false;
}
@Override
@@@ -178,12 -187,11 +193,14 @@@
VisibilityFilter visFilter = new VisibilityFilter(colFilter, options.getAuthorizations(), options.getDefaultLabels());
- if (null == options.getClassLoaderContext()) {
+ if (!loadIters) {
+ return visFilter;
- } else {
++ } else if (null == options.getClassLoaderContext()) {
return iterEnv.getTopLevelIterator(IteratorUtil.loadIterators(IteratorScope.scan, visFilter, tablet.getExtent(), tablet.getTableConfiguration(),
options.getSsiList(), options.getSsio(), iterEnv));
+ } else {
+ return iterEnv.getTopLevelIterator(IteratorUtil.loadIterators(IteratorScope.scan, visFilter, tablet.getExtent(), tablet.getTableConfiguration(),
+ options.getSsiList(), options.getSsio(), iterEnv, true, options.getClassLoaderContext()));
}
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/4eadac16/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
----------------------------------------------------------------------
diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
index de3bfd3,923b4a1..f48aec7
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java
@@@ -632,9 -803,25 +633,26 @@@ public class Tablet implements TabletCo
}
}
+ public void checkConditions(ConditionChecker checker, Authorizations authorizations, AtomicBoolean iFlag) throws IOException {
+
+ ScanDataSource dataSource = new ScanDataSource(this, authorizations, this.defaultSecurityLabel, iFlag);
+
+ try {
+ SortedKeyValueIterator<Key,Value> iter = new SourceSwitchingIterator(dataSource);
+ checker.check(iter);
+ } catch (IOException ioe) {
+ dataSource.close(true);
+ throw ioe;
+ } finally {
+ // code in finally block because always want
+ // to return mapfiles, even when exception is thrown
+ dataSource.close(false);
+ }
+ }
+
public LookupResult lookup(List<Range> ranges, HashSet<Column> columns, Authorizations authorizations, List<KVEntry> results, long maxResultSize,
- List<IterInfo> ssiList, Map<String,Map<String,String>> ssio, AtomicBoolean interruptFlag) throws IOException {
+ List<IterInfo> ssiList, Map<String,Map<String,String>> ssio, AtomicBoolean interruptFlag, SamplerConfiguration samplerConfig, long batchTimeOut,
+ String classLoaderContext) throws IOException {
if (ranges.size() == 0) {
return new LookupResult();
http://git-wip-us.apache.org/repos/asf/accumulo/blob/4eadac16/test/src/main/java/org/apache/accumulo/test/ConditionalWriterIT.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/ConditionalWriterIT.java
index 4f915d3,0000000..6485056
mode 100644,000000..100644
--- a/test/src/main/java/org/apache/accumulo/test/ConditionalWriterIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/ConditionalWriterIT.java
@@@ -1,1349 -1,0 +1,1484 @@@
+/*
+ * 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.accumulo.test;
+
++import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
+import static org.junit.Assert.assertTrue;
+
++import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
++import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Random;
+import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeSet;
+import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.accumulo.cluster.AccumuloCluster;
+import org.apache.accumulo.cluster.ClusterUser;
+import org.apache.accumulo.core.client.AccumuloException;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
+import org.apache.accumulo.core.client.ClientConfiguration;
+import org.apache.accumulo.core.client.ClientConfiguration.ClientProperty;
+import org.apache.accumulo.core.client.ConditionalWriter;
+import org.apache.accumulo.core.client.ConditionalWriter.Result;
+import org.apache.accumulo.core.client.ConditionalWriter.Status;
+import org.apache.accumulo.core.client.ConditionalWriterConfig;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.IsolatedScanner;
+import org.apache.accumulo.core.client.IteratorSetting;
+import org.apache.accumulo.core.client.RowIterator;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.TableDeletedException;
+import org.apache.accumulo.core.client.TableExistsException;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.client.TableOfflineException;
+import org.apache.accumulo.core.client.admin.NewTableConfiguration;
+import org.apache.accumulo.core.client.security.tokens.PasswordToken;
+import org.apache.accumulo.core.data.ArrayByteSequence;
+import org.apache.accumulo.core.data.ByteSequence;
+import org.apache.accumulo.core.data.Condition;
+import org.apache.accumulo.core.data.ConditionalMutation;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.Value;
++import org.apache.accumulo.core.iterators.IteratorEnvironment;
++import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
+import org.apache.accumulo.core.iterators.LongCombiner.Type;
++import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
++import org.apache.accumulo.core.iterators.WrappingIterator;
+import org.apache.accumulo.core.iterators.user.SummingCombiner;
+import org.apache.accumulo.core.iterators.user.VersioningIterator;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.security.ColumnVisibility;
+import org.apache.accumulo.core.security.SystemPermission;
+import org.apache.accumulo.core.security.TablePermission;
+import org.apache.accumulo.core.trace.DistributedTrace;
+import org.apache.accumulo.core.trace.Span;
+import org.apache.accumulo.core.trace.Trace;
+import org.apache.accumulo.core.util.FastFormat;
+import org.apache.accumulo.examples.simple.constraints.AlphaNumKeyConstraint;
+import org.apache.accumulo.harness.AccumuloClusterHarness;
+import org.apache.accumulo.minicluster.impl.MiniAccumuloClusterImpl;
+import org.apache.accumulo.test.functional.BadIterator;
+import org.apache.accumulo.test.functional.SlowIterator;
+import org.apache.accumulo.tracer.TraceDump;
+import org.apache.accumulo.tracer.TraceDump.Printer;
+import org.apache.accumulo.tracer.TraceServer;
+import org.apache.hadoop.io.Text;
+import org.junit.Assert;
+import org.junit.Assume;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
++import com.google.common.base.Charsets;
+import com.google.common.collect.Iterables;
- import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
+
+/**
+ *
+ */
+public class ConditionalWriterIT extends AccumuloClusterHarness {
+ private static final Logger log = LoggerFactory.getLogger(ConditionalWriterIT.class);
+
+ @Override
+ protected int defaultTimeoutSeconds() {
+ return 60;
+ }
+
+ public static long abs(long l) {
+ l = Math.abs(l); // abs(Long.MIN_VALUE) == Long.MIN_VALUE...
+ if (l < 0)
+ return 0;
+ return l;
+ }
+
+ @Before
+ public void deleteUsers() throws Exception {
+ Connector conn = getConnector();
+ Set<String> users = conn.securityOperations().listLocalUsers();
+ ClusterUser user = getUser(0);
+ if (users.contains(user.getPrincipal())) {
+ conn.securityOperations().dropLocalUser(user.getPrincipal());
+ }
+ }
+
+ @Test
+ public void testBasic() throws Exception {
+
+ Connector conn = getConnector();
+ String tableName = getUniqueNames(1)[0];
+
+ conn.tableOperations().create(tableName);
+
+ ConditionalWriter cw = conn.createConditionalWriter(tableName, new ConditionalWriterConfig());
+
+ // mutation conditional on column tx:seq not existing
+ ConditionalMutation cm0 = new ConditionalMutation("99006", new Condition("tx", "seq"));
+ cm0.put("name", "last", "doe");
+ cm0.put("name", "first", "john");
+ cm0.put("tx", "seq", "1");
+ Assert.assertEquals(Status.ACCEPTED, cw.write(cm0).getStatus());
+ Assert.assertEquals(Status.REJECTED, cw.write(cm0).getStatus());
+
+ // mutation conditional on column tx:seq being 1
+ ConditionalMutation cm1 = new ConditionalMutation("99006", new Condition("tx", "seq").setValue("1"));
+ cm1.put("name", "last", "Doe");
+ cm1.put("tx", "seq", "2");
+ Assert.assertEquals(Status.ACCEPTED, cw.write(cm1).getStatus());
+
+ // test condition where value differs
+ ConditionalMutation cm2 = new ConditionalMutation("99006", new Condition("tx", "seq").setValue("1"));
+ cm2.put("name", "last", "DOE");
+ cm2.put("tx", "seq", "2");
+ Assert.assertEquals(Status.REJECTED, cw.write(cm2).getStatus());
+
+ // test condition where column does not exists
+ ConditionalMutation cm3 = new ConditionalMutation("99006", new Condition("txtypo", "seq").setValue("1"));
+ cm3.put("name", "last", "deo");
+ cm3.put("tx", "seq", "2");
+ Assert.assertEquals(Status.REJECTED, cw.write(cm3).getStatus());
+
+ // test two conditions, where one should fail
+ ConditionalMutation cm4 = new ConditionalMutation("99006", new Condition("tx", "seq").setValue("2"), new Condition("name", "last").setValue("doe"));
+ cm4.put("name", "last", "deo");
+ cm4.put("tx", "seq", "3");
+ Assert.assertEquals(Status.REJECTED, cw.write(cm4).getStatus());
+
+ // test two conditions, where one should fail
+ ConditionalMutation cm5 = new ConditionalMutation("99006", new Condition("tx", "seq").setValue("1"), new Condition("name", "last").setValue("Doe"));
+ cm5.put("name", "last", "deo");
+ cm5.put("tx", "seq", "3");
+ Assert.assertEquals(Status.REJECTED, cw.write(cm5).getStatus());
+
+ // ensure rejected mutations did not write
+ Scanner scanner = conn.createScanner(tableName, Authorizations.EMPTY);
+ scanner.fetchColumn(new Text("name"), new Text("last"));
+ scanner.setRange(new Range("99006"));
+ Entry<Key,Value> entry = Iterables.getOnlyElement(scanner);
+ Assert.assertEquals("Doe", entry.getValue().toString());
+
+ // test w/ two conditions that are met
+ ConditionalMutation cm6 = new ConditionalMutation("99006", new Condition("tx", "seq").setValue("2"), new Condition("name", "last").setValue("Doe"));
+ cm6.put("name", "last", "DOE");
+ cm6.put("tx", "seq", "3");
+ Assert.assertEquals(Status.ACCEPTED, cw.write(cm6).getStatus());
+
+ entry = Iterables.getOnlyElement(scanner);
+ Assert.assertEquals("DOE", entry.getValue().toString());
+
+ // test a conditional mutation that deletes
+ ConditionalMutation cm7 = new ConditionalMutation("99006", new Condition("tx", "seq").setValue("3"));
+ cm7.putDelete("name", "last");
+ cm7.putDelete("name", "first");
+ cm7.putDelete("tx", "seq");
+ Assert.assertEquals(Status.ACCEPTED, cw.write(cm7).getStatus());
+
+ Assert.assertFalse("Did not expect to find any results", scanner.iterator().hasNext());
+
+ // add the row back
+ Assert.assertEquals(Status.ACCEPTED, cw.write(cm0).getStatus());
+ Assert.assertEquals(Status.REJECTED, cw.write(cm0).getStatus());
+
+ entry = Iterables.getOnlyElement(scanner);
+ Assert.assertEquals("doe", entry.getValue().toString());
+ }
+
+ @Test
+ public void testFields() throws Exception {
+
+ Connector conn = getConnector();
+ String tableName = getUniqueNames(1)[0];
+
+ String user = null;
+ ClientConfiguration clientConf = cluster.getClientConfig();
+ final boolean saslEnabled = clientConf.getBoolean(ClientProperty.INSTANCE_RPC_SASL_ENABLED.getKey(), false);
+
+ ClusterUser user1 = getUser(0);
+ user = user1.getPrincipal();
+ if (saslEnabled) {
+ // The token is pointless for kerberos
+ conn.securityOperations().createLocalUser(user, null);
+ } else {
+ conn.securityOperations().createLocalUser(user, new PasswordToken(user1.getPassword()));
+ }
+
+ Authorizations auths = new Authorizations("A", "B");
+
+ conn.securityOperations().changeUserAuthorizations(user, auths);
+ conn.securityOperations().grantSystemPermission(user, SystemPermission.CREATE_TABLE);
+
+ conn = conn.getInstance().getConnector(user, user1.getToken());
+
+ conn.tableOperations().create(tableName);
+
+ ConditionalWriter cw = conn.createConditionalWriter(tableName, new ConditionalWriterConfig().setAuthorizations(auths));
+
+ ColumnVisibility cva = new ColumnVisibility("A");
+ ColumnVisibility cvb = new ColumnVisibility("B");
+
+ ConditionalMutation cm0 = new ConditionalMutation("99006", new Condition("tx", "seq").setVisibility(cva));
+ cm0.put("name", "last", cva, "doe");
+ cm0.put("name", "first", cva, "john");
+ cm0.put("tx", "seq", cva, "1");
+ Assert.assertEquals(Status.ACCEPTED, cw.write(cm0).getStatus());
+
+ Scanner scanner = conn.createScanner(tableName, auths);
+ scanner.setRange(new Range("99006"));
+ // TODO verify all columns
+ scanner.fetchColumn(new Text("tx"), new Text("seq"));
+ Entry<Key,Value> entry = Iterables.getOnlyElement(scanner);
+ Assert.assertEquals("1", entry.getValue().toString());
+ long ts = entry.getKey().getTimestamp();
+
+ // test wrong colf
+ ConditionalMutation cm1 = new ConditionalMutation("99006", new Condition("txA", "seq").setVisibility(cva).setValue("1"));
+ cm1.put("name", "last", cva, "Doe");
+ cm1.put("name", "first", cva, "John");
+ cm1.put("tx", "seq", cva, "2");
+ Assert.assertEquals(Status.REJECTED, cw.write(cm1).getStatus());
+
+ // test wrong colq
+ ConditionalMutation cm2 = new ConditionalMutation("99006", new Condition("tx", "seqA").setVisibility(cva).setValue("1"));
+ cm2.put("name", "last", cva, "Doe");
+ cm2.put("name", "first", cva, "John");
+ cm2.put("tx", "seq", cva, "2");
+ Assert.assertEquals(Status.REJECTED, cw.write(cm2).getStatus());
+
+ // test wrong colv
+ ConditionalMutation cm3 = new ConditionalMutation("99006", new Condition("tx", "seq").setVisibility(cvb).setValue("1"));
+ cm3.put("name", "last", cva, "Doe");
+ cm3.put("name", "first", cva, "John");
+ cm3.put("tx", "seq", cva, "2");
+ Assert.assertEquals(Status.REJECTED, cw.write(cm3).getStatus());
+
+ // test wrong timestamp
+ ConditionalMutation cm4 = new ConditionalMutation("99006", new Condition("tx", "seq").setVisibility(cva).setTimestamp(ts + 1).setValue("1"));
+ cm4.put("name", "last", cva, "Doe");
+ cm4.put("name", "first", cva, "John");
+ cm4.put("tx", "seq", cva, "2");
+ Assert.assertEquals(Status.REJECTED, cw.write(cm4).getStatus());
+
+ // test wrong timestamp
+ ConditionalMutation cm5 = new ConditionalMutation("99006", new Condition("tx", "seq").setVisibility(cva).setTimestamp(ts - 1).setValue("1"));
+ cm5.put("name", "last", cva, "Doe");
+ cm5.put("name", "first", cva, "John");
+ cm5.put("tx", "seq", cva, "2");
+ Assert.assertEquals(Status.REJECTED, cw.write(cm5).getStatus());
+
+ // ensure no updates were made
+ entry = Iterables.getOnlyElement(scanner);
+ Assert.assertEquals("1", entry.getValue().toString());
+
+ // set all columns correctly
+ ConditionalMutation cm6 = new ConditionalMutation("99006", new Condition("tx", "seq").setVisibility(cva).setTimestamp(ts).setValue("1"));
+ cm6.put("name", "last", cva, "Doe");
+ cm6.put("name", "first", cva, "John");
+ cm6.put("tx", "seq", cva, "2");
+ Assert.assertEquals(Status.ACCEPTED, cw.write(cm6).getStatus());
+
+ entry = Iterables.getOnlyElement(scanner);
+ Assert.assertEquals("2", entry.getValue().toString());
+
+ }
+
+ @Test
+ public void testBadColVis() throws Exception {
+ // test when a user sets a col vis in a condition that can never be seen
+
+ Connector conn = getConnector();
+ String tableName = getUniqueNames(1)[0];
+
+ conn.tableOperations().create(tableName);
+
+ Authorizations auths = new Authorizations("A", "B");
+
+ conn.securityOperations().changeUserAuthorizations(getAdminPrincipal(), auths);
+
+ Authorizations filteredAuths = new Authorizations("A");
+
+ ConditionalWriter cw = conn.createConditionalWriter(tableName, new ConditionalWriterConfig().setAuthorizations(filteredAuths));
+
+ ColumnVisibility cva = new ColumnVisibility("A");
+ ColumnVisibility cvb = new ColumnVisibility("B");
+ ColumnVisibility cvc = new ColumnVisibility("C");
+
+ // User has authorization, but didn't include it in the writer
+ ConditionalMutation cm0 = new ConditionalMutation("99006", new Condition("tx", "seq").setVisibility(cvb));
+ cm0.put("name", "last", cva, "doe");
+ cm0.put("name", "first", cva, "john");
+ cm0.put("tx", "seq", cva, "1");
+ Assert.assertEquals(Status.INVISIBLE_VISIBILITY, cw.write(cm0).getStatus());
+
+ ConditionalMutation cm1 = new ConditionalMutation("99006", new Condition("tx", "seq").setVisibility(cvb).setValue("1"));
+ cm1.put("name", "last", cva, "doe");
+ cm1.put("name", "first", cva, "john");
+ cm1.put("tx", "seq", cva, "1");
+ Assert.assertEquals(Status.INVISIBLE_VISIBILITY, cw.write(cm1).getStatus());
+
+ // User does not have the authorization
+ ConditionalMutation cm2 = new ConditionalMutation("99006", new Condition("tx", "seq").setVisibility(cvc));
+ cm2.put("name", "last", cva, "doe");
+ cm2.put("name", "first", cva, "john");
+ cm2.put("tx", "seq", cva, "1");
+ Assert.assertEquals(Status.INVISIBLE_VISIBILITY, cw.write(cm2).getStatus());
+
+ ConditionalMutation cm3 = new ConditionalMutation("99006", new Condition("tx", "seq").setVisibility(cvc).setValue("1"));
+ cm3.put("name", "last", cva, "doe");
+ cm3.put("name", "first", cva, "john");
+ cm3.put("tx", "seq", cva, "1");
+ Assert.assertEquals(Status.INVISIBLE_VISIBILITY, cw.write(cm3).getStatus());
+
+ // if any visibility is bad, good visibilities don't override
+ ConditionalMutation cm4 = new ConditionalMutation("99006", new Condition("tx", "seq").setVisibility(cvb), new Condition("tx", "seq").setVisibility(cva));
+
+ cm4.put("name", "last", cva, "doe");
+ cm4.put("name", "first", cva, "john");
+ cm4.put("tx", "seq", cva, "1");
+ Assert.assertEquals(Status.INVISIBLE_VISIBILITY, cw.write(cm4).getStatus());
+
+ ConditionalMutation cm5 = new ConditionalMutation("99006", new Condition("tx", "seq").setVisibility(cvb).setValue("1"), new Condition("tx", "seq")
+ .setVisibility(cva).setValue("1"));
+ cm5.put("name", "last", cva, "doe");
+ cm5.put("name", "first", cva, "john");
+ cm5.put("tx", "seq", cva, "1");
+ Assert.assertEquals(Status.INVISIBLE_VISIBILITY, cw.write(cm5).getStatus());
+
+ ConditionalMutation cm6 = new ConditionalMutation("99006", new Condition("tx", "seq").setVisibility(cvb).setValue("1"),
+ new Condition("tx", "seq").setVisibility(cva));
+ cm6.put("name", "last", cva, "doe");
+ cm6.put("name", "first", cva, "john");
+ cm6.put("tx", "seq", cva, "1");
+ Assert.assertEquals(Status.INVISIBLE_VISIBILITY, cw.write(cm6).getStatus());
+
+ ConditionalMutation cm7 = new ConditionalMutation("99006", new Condition("tx", "seq").setVisibility(cvb), new Condition("tx", "seq").setVisibility(cva)
+ .setValue("1"));
+ cm7.put("name", "last", cva, "doe");
+ cm7.put("name", "first", cva, "john");
+ cm7.put("tx", "seq", cva, "1");
+ Assert.assertEquals(Status.INVISIBLE_VISIBILITY, cw.write(cm7).getStatus());
+
+ cw.close();
+
+ // test passing auths that exceed users configured auths
+
+ Authorizations exceedingAuths = new Authorizations("A", "B", "D");
+ ConditionalWriter cw2 = conn.createConditionalWriter(tableName, new ConditionalWriterConfig().setAuthorizations(exceedingAuths));
+
+ ConditionalMutation cm8 = new ConditionalMutation("99006", new Condition("tx", "seq").setVisibility(cvb), new Condition("tx", "seq").setVisibility(cva)
+ .setValue("1"));
+ cm8.put("name", "last", cva, "doe");
+ cm8.put("name", "first", cva, "john");
+ cm8.put("tx", "seq", cva, "1");
+
+ try {
+ Status status = cw2.write(cm8).getStatus();
+ Assert.fail("Writing mutation with Authorizations the user doesn't have should fail. Got status: " + status);
+ } catch (AccumuloSecurityException ase) {
+ // expected, check specific failure?
+ } finally {
+ cw2.close();
+ }
+ }
+
+ @Test
+ public void testConstraints() throws Exception {
+ // ensure constraint violations are properly reported
+
+ Connector conn = getConnector();
+ String tableName = getUniqueNames(1)[0];
+
+ conn.tableOperations().create(tableName);
+ conn.tableOperations().addConstraint(tableName, AlphaNumKeyConstraint.class.getName());
+ conn.tableOperations().clone(tableName, tableName + "_clone", true, new HashMap<String,String>(), new HashSet<String>());
+
+ Scanner scanner = conn.createScanner(tableName + "_clone", new Authorizations());
+
+ ConditionalWriter cw = conn.createConditionalWriter(tableName + "_clone", new ConditionalWriterConfig());
+
+ ConditionalMutation cm0 = new ConditionalMutation("99006+", new Condition("tx", "seq"));
+ cm0.put("tx", "seq", "1");
+
+ Assert.assertEquals(Status.VIOLATED, cw.write(cm0).getStatus());
+ Assert.assertFalse("Should find no results in the table is mutation result was violated", scanner.iterator().hasNext());
+
+ ConditionalMutation cm1 = new ConditionalMutation("99006", new Condition("tx", "seq"));
+ cm1.put("tx", "seq", "1");
+
+ Assert.assertEquals(Status.ACCEPTED, cw.write(cm1).getStatus());
+ Assert.assertTrue("Accepted result should be returned when reading table", scanner.iterator().hasNext());
+
+ cw.close();
+ }
+
+ @Test
+ public void testIterators() throws Exception {
+
+ Connector conn = getConnector();
+ String tableName = getUniqueNames(1)[0];
+
+ conn.tableOperations().create(tableName, new NewTableConfiguration().withoutDefaultIterators());
+
+ BatchWriter bw = conn.createBatchWriter(tableName, new BatchWriterConfig());
+
+ Mutation m = new Mutation("ACCUMULO-1000");
+ m.put("count", "comments", "1");
+ bw.addMutation(m);
+ bw.addMutation(m);
+ bw.addMutation(m);
+
+ m = new Mutation("ACCUMULO-1001");
+ m.put("count2", "comments", "1");
+ bw.addMutation(m);
+ bw.addMutation(m);
+
+ m = new Mutation("ACCUMULO-1002");
+ m.put("count2", "comments", "1");
+ bw.addMutation(m);
+ bw.addMutation(m);
+
+ bw.close();
+
+ IteratorSetting iterConfig = new IteratorSetting(10, SummingCombiner.class);
+ SummingCombiner.setEncodingType(iterConfig, Type.STRING);
+ SummingCombiner.setColumns(iterConfig, Collections.singletonList(new IteratorSetting.Column("count")));
+
+ IteratorSetting iterConfig2 = new IteratorSetting(10, SummingCombiner.class);
+ SummingCombiner.setEncodingType(iterConfig2, Type.STRING);
+ SummingCombiner.setColumns(iterConfig2, Collections.singletonList(new IteratorSetting.Column("count2", "comments")));
+
+ IteratorSetting iterConfig3 = new IteratorSetting(5, VersioningIterator.class);
+ VersioningIterator.setMaxVersions(iterConfig3, 1);
+
+ Scanner scanner = conn.createScanner(tableName, new Authorizations());
+ scanner.addScanIterator(iterConfig);
+ scanner.setRange(new Range("ACCUMULO-1000"));
+ scanner.fetchColumn(new Text("count"), new Text("comments"));
+
+ Entry<Key,Value> entry = Iterables.getOnlyElement(scanner);
+ Assert.assertEquals("3", entry.getValue().toString());
+
+ ConditionalWriter cw = conn.createConditionalWriter(tableName, new ConditionalWriterConfig());
+
+ ConditionalMutation cm0 = new ConditionalMutation("ACCUMULO-1000", new Condition("count", "comments").setValue("3"));
+ cm0.put("count", "comments", "1");
+ Assert.assertEquals(Status.REJECTED, cw.write(cm0).getStatus());
+ entry = Iterables.getOnlyElement(scanner);
+ Assert.assertEquals("3", entry.getValue().toString());
+
+ ConditionalMutation cm1 = new ConditionalMutation("ACCUMULO-1000", new Condition("count", "comments").setIterators(iterConfig).setValue("3"));
+ cm1.put("count", "comments", "1");
+ Assert.assertEquals(Status.ACCEPTED, cw.write(cm1).getStatus());
+ entry = Iterables.getOnlyElement(scanner);
+ Assert.assertEquals("4", entry.getValue().toString());
+
+ ConditionalMutation cm2 = new ConditionalMutation("ACCUMULO-1000", new Condition("count", "comments").setValue("4"));
+ cm2.put("count", "comments", "1");
+ Assert.assertEquals(Status.REJECTED, cw.write(cm1).getStatus());
+ entry = Iterables.getOnlyElement(scanner);
+ Assert.assertEquals("4", entry.getValue().toString());
+
+ // run test with multiple iterators passed in same batch and condition with two iterators
+
+ ConditionalMutation cm3 = new ConditionalMutation("ACCUMULO-1000", new Condition("count", "comments").setIterators(iterConfig).setValue("4"));
+ cm3.put("count", "comments", "1");
+
+ ConditionalMutation cm4 = new ConditionalMutation("ACCUMULO-1001", new Condition("count2", "comments").setIterators(iterConfig2).setValue("2"));
+ cm4.put("count2", "comments", "1");
+
+ ConditionalMutation cm5 = new ConditionalMutation("ACCUMULO-1002", new Condition("count2", "comments").setIterators(iterConfig2, iterConfig3).setValue("2"));
+ cm5.put("count2", "comments", "1");
+
+ Iterator<Result> results = cw.write(Arrays.asList(cm3, cm4, cm5).iterator());
+ Map<String,Status> actual = new HashMap<String,Status>();
+
+ while (results.hasNext()) {
+ Result result = results.next();
+ String k = new String(result.getMutation().getRow());
+ Assert.assertFalse("Did not expect to see multiple resultus for the row: " + k, actual.containsKey(k));
+ actual.put(k, result.getStatus());
+ }
+
+ Map<String,Status> expected = new HashMap<String,Status>();
+ expected.put("ACCUMULO-1000", Status.ACCEPTED);
+ expected.put("ACCUMULO-1001", Status.ACCEPTED);
+ expected.put("ACCUMULO-1002", Status.REJECTED);
+
+ Assert.assertEquals(expected, actual);
+
- // TODO test w/ table that has iterators configured
++ cw.close();
++ }
++
++ public static class AddingIterator extends WrappingIterator {
++ long amount = 0;
++
++ @Override
++ public Value getTopValue() {
++ Value val = super.getTopValue();
++ long l = Long.parseLong(val.toString());
++ String newVal = (l + amount) + "";
++ return new Value(newVal.getBytes(Charsets.UTF_8));
++ }
++
++ @Override
++ public void init(SortedKeyValueIterator<Key,Value> source, Map<String,String> options, IteratorEnvironment env) throws IOException {
++ this.setSource(source);
++ amount = Long.parseLong(options.get("amount"));
++ }
++ }
++
++ public static class MultiplyingIterator extends WrappingIterator {
++ long amount = 0;
++
++ @Override
++ public Value getTopValue() {
++ Value val = super.getTopValue();
++ long l = Long.parseLong(val.toString());
++ String newVal = l * amount + "";
++ return new Value(newVal.getBytes(Charsets.UTF_8));
++ }
++
++ @Override
++ public void init(SortedKeyValueIterator<Key,Value> source, Map<String,String> options, IteratorEnvironment env) throws IOException {
++ this.setSource(source);
++ amount = Long.parseLong(options.get("amount"));
++ }
++ }
++
++ @Test
++ public void testTableAndConditionIterators() throws Exception {
++
++ // test w/ table that has iterators configured
++ Connector conn = getConnector();
++ String tableName = getUniqueNames(1)[0];
++
++ IteratorSetting aiConfig1 = new IteratorSetting(30, "AI1", AddingIterator.class);
++ aiConfig1.addOption("amount", "2");
++ IteratorSetting aiConfig2 = new IteratorSetting(35, "MI1", MultiplyingIterator.class);
++ aiConfig2.addOption("amount", "3");
++ IteratorSetting aiConfig3 = new IteratorSetting(40, "AI2", AddingIterator.class);
++ aiConfig3.addOption("amount", "5");
++
++ conn.tableOperations().create(tableName);
++
++ BatchWriter bw = conn.createBatchWriter(tableName, new BatchWriterConfig());
++
++ Mutation m = new Mutation("ACCUMULO-1000");
++ m.put("count", "comments", "6");
++ bw.addMutation(m);
++
++ m = new Mutation("ACCUMULO-1001");
++ m.put("count", "comments", "7");
++ bw.addMutation(m);
++
++ m = new Mutation("ACCUMULO-1002");
++ m.put("count", "comments", "8");
++ bw.addMutation(m);
++
++ bw.close();
++
++ conn.tableOperations().attachIterator(tableName, aiConfig1, EnumSet.of(IteratorScope.scan));
++ conn.tableOperations().offline(tableName, true);
++ conn.tableOperations().online(tableName, true);
++
++ ConditionalWriter cw = conn.createConditionalWriter(tableName, new ConditionalWriterConfig());
++
++ ConditionalMutation cm6 = new ConditionalMutation("ACCUMULO-1000", new Condition("count", "comments").setValue("8"));
++ cm6.put("count", "comments", "7");
++ Assert.assertEquals(Status.ACCEPTED, cw.write(cm6).getStatus());
++
++ Scanner scanner = conn.createScanner(tableName, new Authorizations());
++ scanner.setRange(new Range("ACCUMULO-1000"));
++ scanner.fetchColumn(new Text("count"), new Text("comments"));
++
++ Entry<Key,Value> entry = Iterables.getOnlyElement(scanner);
++ Assert.assertEquals("9", entry.getValue().toString());
++
++ ConditionalMutation cm7 = new ConditionalMutation("ACCUMULO-1000", new Condition("count", "comments").setIterators(aiConfig2).setValue("27"));
++ cm7.put("count", "comments", "8");
++ Assert.assertEquals(Status.ACCEPTED, cw.write(cm7).getStatus());
++
++ entry = Iterables.getOnlyElement(scanner);
++ Assert.assertEquals("10", entry.getValue().toString());
++
++ ConditionalMutation cm8 = new ConditionalMutation("ACCUMULO-1000", new Condition("count", "comments").setIterators(aiConfig2, aiConfig3).setValue("35"));
++ cm8.put("count", "comments", "9");
++ Assert.assertEquals(Status.ACCEPTED, cw.write(cm8).getStatus());
++
++ entry = Iterables.getOnlyElement(scanner);
++ Assert.assertEquals("11", entry.getValue().toString());
++
++ ConditionalMutation cm3 = new ConditionalMutation("ACCUMULO-1000", new Condition("count", "comments").setIterators(aiConfig2).setValue("33"));
++ cm3.put("count", "comments", "3");
++
++ ConditionalMutation cm4 = new ConditionalMutation("ACCUMULO-1001", new Condition("count", "comments").setIterators(aiConfig3).setValue("14"));
++ cm4.put("count", "comments", "3");
++
++ ConditionalMutation cm5 = new ConditionalMutation("ACCUMULO-1002", new Condition("count", "comments").setIterators(aiConfig3).setValue("10"));
++ cm5.put("count", "comments", "3");
++
++ Iterator<Result> results = cw.write(Arrays.asList(cm3, cm4, cm5).iterator());
++ Map<String,Status> actual = new HashMap<String,Status>();
++
++ while (results.hasNext()) {
++ Result result = results.next();
++ String k = new String(result.getMutation().getRow());
++ Assert.assertFalse("Did not expect to see multiple resultus for the row: " + k, actual.containsKey(k));
++ actual.put(k, result.getStatus());
++ }
++
++ cw.close();
++
++ Map<String,Status> expected = new HashMap<String,Status>();
++ expected.put("ACCUMULO-1000", Status.ACCEPTED);
++ expected.put("ACCUMULO-1001", Status.ACCEPTED);
++ expected.put("ACCUMULO-1002", Status.REJECTED);
++
++ Assert.assertEquals(expected, actual);
+
+ cw.close();
+ }
+
+ @Test
+ public void testBatch() throws Exception {
+
+ Connector conn = getConnector();
+ String tableName = getUniqueNames(1)[0];
+
+ conn.tableOperations().create(tableName);
+
+ conn.securityOperations().changeUserAuthorizations(getAdminPrincipal(), new Authorizations("A", "B"));
+
+ ColumnVisibility cvab = new ColumnVisibility("A|B");
+
+ ArrayList<ConditionalMutation> mutations = new ArrayList<ConditionalMutation>();
+
+ ConditionalMutation cm0 = new ConditionalMutation("99006", new Condition("tx", "seq").setVisibility(cvab));
+ cm0.put("name", "last", cvab, "doe");
+ cm0.put("name", "first", cvab, "john");
+ cm0.put("tx", "seq", cvab, "1");
+ mutations.add(cm0);
+
+ ConditionalMutation cm1 = new ConditionalMutation("59056", new Condition("tx", "seq").setVisibility(cvab));
+ cm1.put("name", "last", cvab, "doe");
+ cm1.put("name", "first", cvab, "jane");
+ cm1.put("tx", "seq", cvab, "1");
+ mutations.add(cm1);
+
+ ConditionalMutation cm2 = new ConditionalMutation("19059", new Condition("tx", "seq").setVisibility(cvab));
+ cm2.put("name", "last", cvab, "doe");
+ cm2.put("name", "first", cvab, "jack");
+ cm2.put("tx", "seq", cvab, "1");
+ mutations.add(cm2);
+
+ ConditionalWriter cw = conn.createConditionalWriter(tableName, new ConditionalWriterConfig().setAuthorizations(new Authorizations("A")));
+ Iterator<Result> results = cw.write(mutations.iterator());
+ int count = 0;
+ while (results.hasNext()) {
+ Result result = results.next();
+ Assert.assertEquals(Status.ACCEPTED, result.getStatus());
+ count++;
+ }
+
+ Assert.assertEquals(3, count);
+
+ Scanner scanner = conn.createScanner(tableName, new Authorizations("A"));
+ scanner.fetchColumn(new Text("tx"), new Text("seq"));
+
+ for (String row : new String[] {"99006", "59056", "19059"}) {
+ scanner.setRange(new Range(row));
+ Entry<Key,Value> entry = Iterables.getOnlyElement(scanner);
+ Assert.assertEquals("1", entry.getValue().toString());
+ }
+
+ TreeSet<Text> splits = new TreeSet<Text>();
+ splits.add(new Text("7"));
+ splits.add(new Text("3"));
+ conn.tableOperations().addSplits(tableName, splits);
+
+ mutations.clear();
+
+ ConditionalMutation cm3 = new ConditionalMutation("99006", new Condition("tx", "seq").setVisibility(cvab).setValue("1"));
+ cm3.put("name", "last", cvab, "Doe");
+ cm3.put("tx", "seq", cvab, "2");
+ mutations.add(cm3);
+
+ ConditionalMutation cm4 = new ConditionalMutation("59056", new Condition("tx", "seq").setVisibility(cvab));
+ cm4.put("name", "last", cvab, "Doe");
+ cm4.put("tx", "seq", cvab, "1");
+ mutations.add(cm4);
+
+ ConditionalMutation cm5 = new ConditionalMutation("19059", new Condition("tx", "seq").setVisibility(cvab).setValue("2"));
+ cm5.put("name", "last", cvab, "Doe");
+ cm5.put("tx", "seq", cvab, "3");
+ mutations.add(cm5);
+
+ results = cw.write(mutations.iterator());
+ int accepted = 0;
+ int rejected = 0;
+ while (results.hasNext()) {
+ Result result = results.next();
+ if (new String(result.getMutation().getRow()).equals("99006")) {
+ Assert.assertEquals(Status.ACCEPTED, result.getStatus());
+ accepted++;
+ } else {
+ Assert.assertEquals(Status.REJECTED, result.getStatus());
+ rejected++;
+ }
+ }
+
+ Assert.assertEquals("Expected only one accepted conditional mutation", 1, accepted);
+ Assert.assertEquals("Expected two rejected conditional mutations", 2, rejected);
+
+ for (String row : new String[] {"59056", "19059"}) {
+ scanner.setRange(new Range(row));
+ Entry<Key,Value> entry = Iterables.getOnlyElement(scanner);
+ Assert.assertEquals("1", entry.getValue().toString());
+ }
+
+ scanner.setRange(new Range("99006"));
+ Entry<Key,Value> entry = Iterables.getOnlyElement(scanner);
+ Assert.assertEquals("2", entry.getValue().toString());
+
+ scanner.clearColumns();
+ scanner.fetchColumn(new Text("name"), new Text("last"));
+ entry = Iterables.getOnlyElement(scanner);
+ Assert.assertEquals("Doe", entry.getValue().toString());
+
+ cw.close();
+ }
+
+ @Test
+ public void testBigBatch() throws Exception {
+
+ Connector conn = getConnector();
+ String tableName = getUniqueNames(1)[0];
+
+ conn.tableOperations().create(tableName);
+ conn.tableOperations().addSplits(tableName, nss("2", "4", "6"));
+
+ sleepUninterruptibly(2, TimeUnit.SECONDS);
+
+ int num = 100;
+
+ ArrayList<byte[]> rows = new ArrayList<byte[]>(num);
+ ArrayList<ConditionalMutation> cml = new ArrayList<ConditionalMutation>(num);
+
+ Random r = new Random();
+ byte[] e = new byte[0];
+
+ for (int i = 0; i < num; i++) {
+ rows.add(FastFormat.toZeroPaddedString(abs(r.nextLong()), 16, 16, e));
+ }
+
+ for (int i = 0; i < num; i++) {
+ ConditionalMutation cm = new ConditionalMutation(rows.get(i), new Condition("meta", "seq"));
+
+ cm.put("meta", "seq", "1");
+ cm.put("meta", "tx", UUID.randomUUID().toString());
+
+ cml.add(cm);
+ }
+
+ ConditionalWriter cw = conn.createConditionalWriter(tableName, new ConditionalWriterConfig());
+
+ Iterator<Result> results = cw.write(cml.iterator());
+
+ int count = 0;
+
+ // TODO check got each row back
+ while (results.hasNext()) {
+ Result result = results.next();
+ Assert.assertEquals(Status.ACCEPTED, result.getStatus());
+ count++;
+ }
+
+ Assert.assertEquals("Did not receive the expected number of results", num, count);
+
+ ArrayList<ConditionalMutation> cml2 = new ArrayList<ConditionalMutation>(num);
+
+ for (int i = 0; i < num; i++) {
+ ConditionalMutation cm = new ConditionalMutation(rows.get(i), new Condition("meta", "seq").setValue("1"));
+
+ cm.put("meta", "seq", "2");
+ cm.put("meta", "tx", UUID.randomUUID().toString());
+
+ cml2.add(cm);
+ }
+
+ count = 0;
+
+ results = cw.write(cml2.iterator());
+
+ while (results.hasNext()) {
+ Result result = results.next();
+ Assert.assertEquals(Status.ACCEPTED, result.getStatus());
+ count++;
+ }
+
+ Assert.assertEquals("Did not receive the expected number of results", num, count);
+
+ cw.close();
+ }
+
+ @Test
+ public void testBatchErrors() throws Exception {
+
+ Connector conn = getConnector();
+ String tableName = getUniqueNames(1)[0];
+
+ conn.tableOperations().create(tableName);
+ conn.tableOperations().addConstraint(tableName, AlphaNumKeyConstraint.class.getName());
+ conn.tableOperations().clone(tableName, tableName + "_clone", true, new HashMap<String,String>(), new HashSet<String>());
+
+ conn.securityOperations().changeUserAuthorizations(getAdminPrincipal(), new Authorizations("A", "B"));
+
+ ColumnVisibility cvaob = new ColumnVisibility("A|B");
+ ColumnVisibility cvaab = new ColumnVisibility("A&B");
+
+ switch ((new Random()).nextInt(3)) {
+ case 1:
+ conn.tableOperations().addSplits(tableName, nss("6"));
+ break;
+ case 2:
+ conn.tableOperations().addSplits(tableName, nss("2", "95"));
+ break;
+ }
+
+ ArrayList<ConditionalMutation> mutations = new ArrayList<ConditionalMutation>();
+
+ ConditionalMutation cm0 = new ConditionalMutation("99006", new Condition("tx", "seq").setVisibility(cvaob));
+ cm0.put("name+", "last", cvaob, "doe");
+ cm0.put("name", "first", cvaob, "john");
+ cm0.put("tx", "seq", cvaob, "1");
+ mutations.add(cm0);
+
+ ConditionalMutation cm1 = new ConditionalMutation("59056", new Condition("tx", "seq").setVisibility(cvaab));
+ cm1.put("name", "last", cvaab, "doe");
+ cm1.put("name", "first", cvaab, "jane");
+ cm1.put("tx", "seq", cvaab, "1");
+ mutations.add(cm1);
+
+ ConditionalMutation cm2 = new ConditionalMutation("19059", new Condition("tx", "seq").setVisibility(cvaob));
+ cm2.put("name", "last", cvaob, "doe");
+ cm2.put("name", "first", cvaob, "jack");
+ cm2.put("tx", "seq", cvaob, "1");
+ mutations.add(cm2);
+
+ ConditionalMutation cm3 = new ConditionalMutation("90909", new Condition("tx", "seq").setVisibility(cvaob).setValue("1"));
+ cm3.put("name", "last", cvaob, "doe");
+ cm3.put("name", "first", cvaob, "john");
+ cm3.put("tx", "seq", cvaob, "2");
+ mutations.add(cm3);
+
+ ConditionalWriter cw = conn.createConditionalWriter(tableName, new ConditionalWriterConfig().setAuthorizations(new Authorizations("A")));
+ Iterator<Result> results = cw.write(mutations.iterator());
+ HashSet<String> rows = new HashSet<String>();
+ while (results.hasNext()) {
+ Result result = results.next();
+ String row = new String(result.getMutation().getRow());
+ if (row.equals("19059")) {
+ Assert.assertEquals(Status.ACCEPTED, result.getStatus());
+ } else if (row.equals("59056")) {
+ Assert.assertEquals(Status.INVISIBLE_VISIBILITY, result.getStatus());
+ } else if (row.equals("99006")) {
+ Assert.assertEquals(Status.VIOLATED, result.getStatus());
+ } else if (row.equals("90909")) {
+ Assert.assertEquals(Status.REJECTED, result.getStatus());
+ }
+ rows.add(row);
+ }
+
+ Assert.assertEquals(4, rows.size());
+
+ Scanner scanner = conn.createScanner(tableName, new Authorizations("A"));
+ scanner.fetchColumn(new Text("tx"), new Text("seq"));
+
+ Entry<Key,Value> entry = Iterables.getOnlyElement(scanner);
+ Assert.assertEquals("1", entry.getValue().toString());
+
+ cw.close();
+ }
+
+ @Test
+ public void testSameRow() throws Exception {
+ // test multiple mutations for same row in same batch
+
+ Connector conn = getConnector();
+ String tableName = getUniqueNames(1)[0];
+
+ conn.tableOperations().create(tableName);
+
+ ConditionalWriter cw = conn.createConditionalWriter(tableName, new ConditionalWriterConfig());
+
+ ConditionalMutation cm1 = new ConditionalMutation("r1", new Condition("tx", "seq"));
+ cm1.put("tx", "seq", "1");
+ cm1.put("data", "x", "a");
+
+ Assert.assertEquals(Status.ACCEPTED, cw.write(cm1).getStatus());
+
+ ConditionalMutation cm2 = new ConditionalMutation("r1", new Condition("tx", "seq").setValue("1"));
+ cm2.put("tx", "seq", "2");
+ cm2.put("data", "x", "b");
+
+ ConditionalMutation cm3 = new ConditionalMutation("r1", new Condition("tx", "seq").setValue("1"));
+ cm3.put("tx", "seq", "2");
+ cm3.put("data", "x", "c");
+
+ ConditionalMutation cm4 = new ConditionalMutation("r1", new Condition("tx", "seq").setValue("1"));
+ cm4.put("tx", "seq", "2");
+ cm4.put("data", "x", "d");
+
+ Iterator<Result> results = cw.write(Arrays.asList(cm2, cm3, cm4).iterator());
+
+ int accepted = 0;
+ int rejected = 0;
+ int total = 0;
+
+ while (results.hasNext()) {
+ Status status = results.next().getStatus();
+ if (status == Status.ACCEPTED)
+ accepted++;
+ if (status == Status.REJECTED)
+ rejected++;
+ total++;
+ }
+
+ Assert.assertEquals("Expected one accepted result", 1, accepted);
+ Assert.assertEquals("Expected two rejected results", 2, rejected);
+ Assert.assertEquals("Expected three total results", 3, total);
+
+ cw.close();
+ }
+
+ private static class Stats {
+
+ ByteSequence row = null;
+ int seq;
+ long sum;
+ int data[] = new int[10];
+
+ public Stats(Iterator<Entry<Key,Value>> iterator) {
+ while (iterator.hasNext()) {
+ Entry<Key,Value> entry = iterator.next();
+
+ if (row == null)
+ row = entry.getKey().getRowData();
+
+ String cf = entry.getKey().getColumnFamilyData().toString();
+ String cq = entry.getKey().getColumnQualifierData().toString();
+
+ if (cf.equals("data")) {
+ data[Integer.parseInt(cq)] = Integer.parseInt(entry.getValue().toString());
+ } else if (cf.equals("meta")) {
+ if (cq.equals("sum")) {
+ sum = Long.parseLong(entry.getValue().toString());
+ } else if (cq.equals("seq")) {
+ seq = Integer.parseInt(entry.getValue().toString());
+ }
+ }
+ }
+
+ long sum2 = 0;
+
+ for (int datum : data) {
+ sum2 += datum;
+ }
+
+ Assert.assertEquals(sum2, sum);
+ }
+
+ public Stats(ByteSequence row) {
+ this.row = row;
+ for (int i = 0; i < data.length; i++) {
+ this.data[i] = 0;
+ }
+ this.seq = -1;
+ this.sum = 0;
+ }
+
+ void set(int index, int value) {
+ sum -= data[index];
+ sum += value;
+ data[index] = value;
+ }
+
+ ConditionalMutation toMutation() {
+ Condition cond = new Condition("meta", "seq");
+ if (seq >= 0)
+ cond.setValue(seq + "");
+
+ ConditionalMutation cm = new ConditionalMutation(row, cond);
+
+ cm.put("meta", "seq", (seq + 1) + "");
+ cm.put("meta", "sum", (sum) + "");
+
+ for (int i = 0; i < data.length; i++) {
+ cm.put("data", i + "", data[i] + "");
+ }
+
+ return cm;
+ }
+
+ @Override
+ public String toString() {
+ return row + " " + seq + " " + sum;
+ }
+ }
+
+ private static class MutatorTask implements Runnable {
+ String table;
+ ArrayList<ByteSequence> rows;
+ ConditionalWriter cw;
+ Connector conn;
+ AtomicBoolean failed;
+
+ public MutatorTask(String table, Connector conn, ArrayList<ByteSequence> rows, ConditionalWriter cw, AtomicBoolean failed) {
+ this.table = table;
+ this.rows = rows;
+ this.conn = conn;
+ this.cw = cw;
+ this.failed = failed;
+ }
+
+ @Override
+ public void run() {
+ try {
+ Random rand = new Random();
+
+ Scanner scanner = new IsolatedScanner(conn.createScanner(table, Authorizations.EMPTY));
+
+ for (int i = 0; i < 20; i++) {
+ int numRows = rand.nextInt(10) + 1;
+
+ ArrayList<ByteSequence> changes = new ArrayList<ByteSequence>(numRows);
+ ArrayList<ConditionalMutation> mutations = new ArrayList<ConditionalMutation>();
+
+ for (int j = 0; j < numRows; j++)
+ changes.add(rows.get(rand.nextInt(rows.size())));
+
+ for (ByteSequence row : changes) {
+ scanner.setRange(new Range(row.toString()));
+ Stats stats = new Stats(scanner.iterator());
+ stats.set(rand.nextInt(10), rand.nextInt(Integer.MAX_VALUE));
+ mutations.add(stats.toMutation());
+ }
+
+ ArrayList<ByteSequence> changed = new ArrayList<ByteSequence>(numRows);
+ Iterator<Result> results = cw.write(mutations.iterator());
+ while (results.hasNext()) {
+ Result result = results.next();
+ changed.add(new ArrayByteSequence(result.getMutation().getRow()));
+ }
+
+ Collections.sort(changes);
+ Collections.sort(changed);
+
+ Assert.assertEquals(changes, changed);
+
+ }
+
+ } catch (Exception e) {
+ log.error("{}", e.getMessage(), e);
+ failed.set(true);
+ }
+ }
+ }
+
+ @Test
+ public void testThreads() throws Exception {
+ // test multiple threads using a single conditional writer
+
+ String table = getUniqueNames(1)[0];
+ Connector conn = getConnector();
+
+ conn.tableOperations().create(table);
+
+ Random rand = new Random();
+
+ switch (rand.nextInt(3)) {
+ case 1:
+ conn.tableOperations().addSplits(table, nss("4"));
+ break;
+ case 2:
+ conn.tableOperations().addSplits(table, nss("3", "5"));
+ break;
+ }
+
+ ConditionalWriter cw = conn.createConditionalWriter(table, new ConditionalWriterConfig());
+
+ ArrayList<ByteSequence> rows = new ArrayList<ByteSequence>();
+
+ for (int i = 0; i < 1000; i++) {
+ rows.add(new ArrayByteSequence(FastFormat.toZeroPaddedString(abs(rand.nextLong()), 16, 16, new byte[0])));
+ }
+
+ ArrayList<ConditionalMutation> mutations = new ArrayList<ConditionalMutation>();
+
+ for (ByteSequence row : rows)
+ mutations.add(new Stats(row).toMutation());
+
+ ArrayList<ByteSequence> rows2 = new ArrayList<ByteSequence>();
+ Iterator<Result> results = cw.write(mutations.iterator());
+ while (results.hasNext()) {
+ Result result = results.next();
+ Assert.assertEquals(Status.ACCEPTED, result.getStatus());
+ rows2.add(new ArrayByteSequence(result.getMutation().getRow()));
+ }
+
+ Collections.sort(rows);
+ Collections.sort(rows2);
+
+ Assert.assertEquals(rows, rows2);
+
+ AtomicBoolean failed = new AtomicBoolean(false);
+
+ ExecutorService tp = Executors.newFixedThreadPool(5);
+ for (int i = 0; i < 5; i++) {
+ tp.submit(new MutatorTask(table, conn, rows, cw, failed));
+ }
+
+ tp.shutdown();
+
+ while (!tp.isTerminated()) {
+ tp.awaitTermination(1, TimeUnit.MINUTES);
+ }
+
+ Assert.assertFalse("A MutatorTask failed with an exception", failed.get());
+
+ Scanner scanner = conn.createScanner(table, Authorizations.EMPTY);
+
+ RowIterator rowIter = new RowIterator(scanner);
+
+ while (rowIter.hasNext()) {
+ Iterator<Entry<Key,Value>> row = rowIter.next();
+ new Stats(row);
+ }
+ }
+
+ private SortedSet<Text> nss(String... splits) {
+ TreeSet<Text> ret = new TreeSet<Text>();
+ for (String split : splits)
+ ret.add(new Text(split));
+
+ return ret;
+ }
+
+ @Test
+ public void testSecurity() throws Exception {
+ // test against table user does not have read and/or write permissions for
+ Connector conn = getConnector();
+ String user = null;
+ ClientConfiguration clientConf = cluster.getClientConfig();
+ final boolean saslEnabled = clientConf.getBoolean(ClientProperty.INSTANCE_RPC_SASL_ENABLED.getKey(), false);
+
+ // Create a new user
+ ClusterUser user1 = getUser(0);
+ user = user1.getPrincipal();
+ if (saslEnabled) {
+ conn.securityOperations().createLocalUser(user, null);
+ } else {
+ conn.securityOperations().createLocalUser(user, new PasswordToken(user1.getPassword()));
+ }
+
+ String[] tables = getUniqueNames(3);
+ String table1 = tables[0], table2 = tables[1], table3 = tables[2];
+
+ // Create three tables
+ conn.tableOperations().create(table1);
+ conn.tableOperations().create(table2);
+ conn.tableOperations().create(table3);
+
+ // Grant R on table1, W on table2, R/W on table3
+ conn.securityOperations().grantTablePermission(user, table1, TablePermission.READ);
+ conn.securityOperations().grantTablePermission(user, table2, TablePermission.WRITE);
+ conn.securityOperations().grantTablePermission(user, table3, TablePermission.READ);
+ conn.securityOperations().grantTablePermission(user, table3, TablePermission.WRITE);
+
+ // Login as the user
+ Connector conn2 = conn.getInstance().getConnector(user, user1.getToken());
+
+ ConditionalMutation cm1 = new ConditionalMutation("r1", new Condition("tx", "seq"));
+ cm1.put("tx", "seq", "1");
+ cm1.put("data", "x", "a");
+
+ ConditionalWriter cw1 = conn2.createConditionalWriter(table1, new ConditionalWriterConfig());
+ ConditionalWriter cw2 = conn2.createConditionalWriter(table2, new ConditionalWriterConfig());
+ ConditionalWriter cw3 = conn2.createConditionalWriter(table3, new ConditionalWriterConfig());
+
+ // Should be able to conditional-update a table we have R/W on
+ Assert.assertEquals(Status.ACCEPTED, cw3.write(cm1).getStatus());
+
+ // Conditional-update to a table we only have read on should fail
+ try {
+ Status status = cw1.write(cm1).getStatus();
+ Assert.fail("Expected exception writing conditional mutation to table the user doesn't have write access to, Got status: " + status);
+ } catch (AccumuloSecurityException ase) {
+
+ }
+
+ // Conditional-update to a table we only have writer on should fail
+ try {
+ Status status = cw2.write(cm1).getStatus();
+ Assert.fail("Expected exception writing conditional mutation to table the user doesn't have read access to. Got status: " + status);
+ } catch (AccumuloSecurityException ase) {
+
+ }
+ }
+
+ @Test
+ public void testTimeout() throws Exception {
+ Connector conn = getConnector();
+
+ String table = getUniqueNames(1)[0];
+
+ conn.tableOperations().create(table);
+
+ ConditionalWriter cw = conn.createConditionalWriter(table, new ConditionalWriterConfig().setTimeout(3, TimeUnit.SECONDS));
+
+ ConditionalMutation cm1 = new ConditionalMutation("r1", new Condition("tx", "seq"));
+ cm1.put("tx", "seq", "1");
+ cm1.put("data", "x", "a");
+
+ Assert.assertEquals(cw.write(cm1).getStatus(), Status.ACCEPTED);
+
+ IteratorSetting is = new IteratorSetting(5, SlowIterator.class);
+ SlowIterator.setSeekSleepTime(is, 5000);
+
+ ConditionalMutation cm2 = new ConditionalMutation("r1", new Condition("tx", "seq").setValue("1").setIterators(is));
+ cm2.put("tx", "seq", "2");
+ cm2.put("data", "x", "b");
+
+ Assert.assertEquals(cw.write(cm2).getStatus(), Status.UNKNOWN);
+
+ Scanner scanner = conn.createScanner(table, Authorizations.EMPTY);
+
+ for (Entry<Key,Value> entry : scanner) {
+ String cf = entry.getKey().getColumnFamilyData().toString();
+ String cq = entry.getKey().getColumnQualifierData().toString();
+ String val = entry.getValue().toString();
+
+ if (cf.equals("tx") && cq.equals("seq"))
+ Assert.assertEquals("Unexpected value in tx:seq", "1", val);
+ else if (cf.equals("data") && cq.equals("x"))
+ Assert.assertEquals("Unexpected value in data:x", "a", val);
+ else
+ Assert.fail("Saw unexpected column family and qualifier: " + entry);
+ }
+
+ ConditionalMutation cm3 = new ConditionalMutation("r1", new Condition("tx", "seq").setValue("1"));
+ cm3.put("tx", "seq", "2");
+ cm3.put("data", "x", "b");
+
+ Assert.assertEquals(cw.write(cm3).getStatus(), Status.ACCEPTED);
+
+ cw.close();
+ }
+
+ @Test
+ public void testDeleteTable() throws Exception {
+ String table = getUniqueNames(1)[0];
+ Connector conn = getConnector();
+
+ try {
+ conn.createConditionalWriter(table, new ConditionalWriterConfig());
+ Assert.fail("Creating conditional writer for table that doesn't exist should fail");
+ } catch (TableNotFoundException e) {}
+
+ conn.tableOperations().create(table);
+
+ ConditionalWriter cw = conn.createConditionalWriter(table, new ConditionalWriterConfig());
+
+ conn.tableOperations().delete(table);
+
+ ConditionalMutation cm1 = new ConditionalMutation("r1", new Condition("tx", "seq"));
+ cm1.put("tx", "seq", "1");
+ cm1.put("data", "x", "a");
+
+ Result result = cw.write(cm1);
+
+ try {
+ Status status = result.getStatus();
+ Assert.fail("Expected exception writing conditional mutation to deleted table. Got status: " + status);
+ } catch (AccumuloException ae) {
+ Assert.assertEquals(TableDeletedException.class, ae.getCause().getClass());
+ }
+ }
+
+ @Test
+ public void testOffline() throws Exception {
+ String table = getUniqueNames(1)[0];
+ Connector conn = getConnector();
+
+ conn.tableOperations().create(table);
+
+ ConditionalWriter cw = conn.createConditionalWriter(table, new ConditionalWriterConfig());
+
+ conn.tableOperations().offline(table, true);
+
+ ConditionalMutation cm1 = new ConditionalMutation("r1", new Condition("tx", "seq"));
+ cm1.put("tx", "seq", "1");
+ cm1.put("data", "x", "a");
+
+ Result result = cw.write(cm1);
+
+ try {
+ Status status = result.getStatus();
+ Assert.fail("Expected exception writing conditional mutation to offline table. Got status: " + status);
+ } catch (AccumuloException ae) {
+ Assert.assertEquals(TableOfflineException.class, ae.getCause().getClass());
+ }
+
+ cw.close();
+
+ try {
+ conn.createConditionalWriter(table, new ConditionalWriterConfig());
+ Assert.fail("Expected exception creating conditional writer to offline table");
+ } catch (TableOfflineException e) {}
+ }
+
+ @Test
+ public void testError() throws Exception {
+ String table = getUniqueNames(1)[0];
+ Connector conn = getConnector();
+
+ conn.tableOperations().create(table);
+
+ ConditionalWriter cw = conn.createConditionalWriter(table, new ConditionalWriterConfig());
+
+ IteratorSetting iterSetting = new IteratorSetting(5, BadIterator.class);
+
+ ConditionalMutation cm1 = new ConditionalMutation("r1", new Condition("tx", "seq").setIterators(iterSetting));
+ cm1.put("tx", "seq", "1");
+ cm1.put("data", "x", "a");
+
+ Result result = cw.write(cm1);
+
+ try {
+ Status status = result.getStatus();
+ Assert.fail("Expected exception using iterator which throws an error, Got status: " + status);
+ } catch (AccumuloException ae) {
+
+ }
+
+ cw.close();
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testNoConditions() throws AccumuloException, AccumuloSecurityException, TableExistsException, TableNotFoundException {
+ String table = getUniqueNames(1)[0];
+ Connector conn = getConnector();
+
+ conn.tableOperations().create(table);
+
+ ConditionalWriter cw = conn.createConditionalWriter(table, new ConditionalWriterConfig());
+
+ ConditionalMutation cm1 = new ConditionalMutation("r1");
+ cm1.put("tx", "seq", "1");
+ cm1.put("data", "x", "a");
+
+ cw.write(cm1);
+ }
+
+ @Test
+ public void testTrace() throws Exception {
+ // Need to add a getClientConfig() to AccumuloCluster
+ Assume.assumeTrue(getClusterType() == ClusterType.MINI);
+ Process tracer = null;
+ Connector conn = getConnector();
+ AccumuloCluster cluster = getCluster();
+ MiniAccumuloClusterImpl mac = (MiniAccumuloClusterImpl) cluster;
+ if (!conn.tableOperations().exists("trace")) {
+ tracer = mac.exec(TraceServer.class);
+ while (!conn.tableOperations().exists("trace")) {
+ sleepUninterruptibly(1, TimeUnit.SECONDS);
+ }
+ }
+
+ String tableName = getUniqueNames(1)[0];
+ conn.tableOperations().create(tableName);
+
+ DistributedTrace.enable("localhost", "testTrace", mac.getClientConfig());
+ sleepUninterruptibly(1, TimeUnit.SECONDS);
+ Span root = Trace.on("traceTest");
+ ConditionalWriter cw = conn.createConditionalWriter(tableName, new ConditionalWriterConfig());
+
+ // mutation conditional on column tx:seq not exiting
+ ConditionalMutation cm0 = new ConditionalMutation("99006", new Condition("tx", "seq"));
+ cm0.put("name", "last", "doe");
+ cm0.put("name", "first", "john");
+ cm0.put("tx", "seq", "1");
+ Assert.assertEquals(Status.ACCEPTED, cw.write(cm0).getStatus());
+ root.stop();
+
+ final Scanner scanner = conn.createScanner("trace", Authorizations.EMPTY);
+ scanner.setRange(new Range(new Text(Long.toHexString(root.traceId()))));
+ loop: while (true) {
+ final StringBuffer finalBuffer = new StringBuffer();
+ int traceCount = TraceDump.printTrace(scanner, new Printer() {
+ @Override
+ public void print(final String line) {
+ try {
+ finalBuffer.append(line).append("\n");
+ } catch (Exception ex) {
+ throw new RuntimeException(ex);
+ }
+ }
+ });
+ String traceOutput = finalBuffer.toString();
+ log.info("Trace output:" + traceOutput);
+ if (traceCount > 0) {
+ int lastPos = 0;
+ for (String part : "traceTest, startScan,startConditionalUpdate,conditionalUpdate,Check conditions,apply conditional mutations".split(",")) {
+ log.info("Looking in trace output for '" + part + "'");
+ int pos = traceOutput.indexOf(part);
+ if (-1 == pos) {
+ log.info("Trace output doesn't contain '" + part + "'");
+ Thread.sleep(1000);
+ break loop;
+ }
+ assertTrue("Did not find '" + part + "' in output", pos > 0);
+ assertTrue("'" + part + "' occurred earlier than the previous element unexpectedly", pos > lastPos);
+ lastPos = pos;
+ }
+ break;
+ } else {
+ log.info("Ignoring trace output as traceCount not greater than zero: " + traceCount);
+ Thread.sleep(1000);
+ }
+ }
+ if (tracer != null) {
+ tracer.destroy();
+ }
+ }
+}