You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by md...@apache.org on 2014/03/04 15:17:27 UTC
[01/16] git commit: [maven-release-plugin] prepare release 1.5.1
Repository: accumulo
Updated Branches:
refs/heads/1.4.5-SNAPSHOT a8bbb916c -> 759582b78
refs/heads/1.5.2-SNAPSHOT bee78fa69 -> bd283aec0
refs/heads/1.6.0-SNAPSHOT e0571816a -> 9abf9424d
refs/heads/master 11d724e87 -> cf61a3c2e
[maven-release-plugin] prepare release 1.5.1
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/3478f71a
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/3478f71a
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/3478f71a
Branch: refs/heads/master
Commit: 3478f71ae888f8d73aaa93837319a6dbb4ba0c8a
Parents: 3145cce
Author: Josh Elser <el...@apache.org>
Authored: Mon Feb 24 13:13:25 2014 -0500
Committer: Josh Elser <el...@apache.org>
Committed: Mon Feb 24 13:13:25 2014 -0500
----------------------------------------------------------------------
assemble/pom.xml | 2 +-
core/pom.xml | 2 +-
examples/pom.xml | 2 +-
examples/simple/pom.xml | 2 +-
fate/pom.xml | 2 +-
minicluster/pom.xml | 2 +-
pom.xml | 4 ++--
proxy/pom.xml | 2 +-
server/pom.xml | 2 +-
start/pom.xml | 2 +-
test/pom.xml | 2 +-
trace/pom.xml | 2 +-
12 files changed, 13 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/3478f71a/assemble/pom.xml
----------------------------------------------------------------------
diff --git a/assemble/pom.xml b/assemble/pom.xml
index 40b77cc..21a31c8 100644
--- a/assemble/pom.xml
+++ b/assemble/pom.xml
@@ -20,7 +20,7 @@
<parent>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-project</artifactId>
- <version>1.5.1-SNAPSHOT</version>
+ <version>1.5.1</version>
</parent>
<artifactId>accumulo</artifactId>
<packaging>pom</packaging>
http://git-wip-us.apache.org/repos/asf/accumulo/blob/3478f71a/core/pom.xml
----------------------------------------------------------------------
diff --git a/core/pom.xml b/core/pom.xml
index d02a3cd..538ec4a 100644
--- a/core/pom.xml
+++ b/core/pom.xml
@@ -20,7 +20,7 @@
<parent>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-project</artifactId>
- <version>1.5.1-SNAPSHOT</version>
+ <version>1.5.1</version>
</parent>
<artifactId>accumulo-core</artifactId>
<name>Core</name>
http://git-wip-us.apache.org/repos/asf/accumulo/blob/3478f71a/examples/pom.xml
----------------------------------------------------------------------
diff --git a/examples/pom.xml b/examples/pom.xml
index ec39e30..56d4311 100644
--- a/examples/pom.xml
+++ b/examples/pom.xml
@@ -20,7 +20,7 @@
<parent>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-project</artifactId>
- <version>1.5.1-SNAPSHOT</version>
+ <version>1.5.1</version>
</parent>
<artifactId>accumulo-examples</artifactId>
<packaging>pom</packaging>
http://git-wip-us.apache.org/repos/asf/accumulo/blob/3478f71a/examples/simple/pom.xml
----------------------------------------------------------------------
diff --git a/examples/simple/pom.xml b/examples/simple/pom.xml
index d513e68..7ca5ed3 100644
--- a/examples/simple/pom.xml
+++ b/examples/simple/pom.xml
@@ -20,7 +20,7 @@
<parent>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-examples</artifactId>
- <version>1.5.1-SNAPSHOT</version>
+ <version>1.5.1</version>
</parent>
<artifactId>accumulo-examples-simple</artifactId>
<name>Simple Examples</name>
http://git-wip-us.apache.org/repos/asf/accumulo/blob/3478f71a/fate/pom.xml
----------------------------------------------------------------------
diff --git a/fate/pom.xml b/fate/pom.xml
index 30c4973..b023acd 100644
--- a/fate/pom.xml
+++ b/fate/pom.xml
@@ -20,7 +20,7 @@
<parent>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-project</artifactId>
- <version>1.5.1-SNAPSHOT</version>
+ <version>1.5.1</version>
</parent>
<artifactId>accumulo-fate</artifactId>
<name>Fate</name>
http://git-wip-us.apache.org/repos/asf/accumulo/blob/3478f71a/minicluster/pom.xml
----------------------------------------------------------------------
diff --git a/minicluster/pom.xml b/minicluster/pom.xml
index 32848d8..877d375 100644
--- a/minicluster/pom.xml
+++ b/minicluster/pom.xml
@@ -20,7 +20,7 @@
<parent>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-project</artifactId>
- <version>1.5.1-SNAPSHOT</version>
+ <version>1.5.1</version>
</parent>
<artifactId>accumulo-minicluster</artifactId>
<name>MiniCluster</name>
http://git-wip-us.apache.org/repos/asf/accumulo/blob/3478f71a/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 8ae58ed..d9d53ee 100644
--- a/pom.xml
+++ b/pom.xml
@@ -24,7 +24,7 @@
</parent>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-project</artifactId>
- <version>1.5.1-SNAPSHOT</version>
+ <version>1.5.1</version>
<packaging>pom</packaging>
<name>Apache Accumulo Project</name>
<description>Apache Accumulo is a sorted, distributed key/value store based on Google's BigTable design. It is built on top of Apache Hadoop, Zookeeper, and Thrift. It features a few novel improvements on the BigTable design in the form of cell-level access labels and a server-side programming mechanism that can modify key/value pairs at various points in the data management process.</description>
@@ -86,7 +86,7 @@
<scm>
<connection>scm:git:git://git.apache.org/accumulo.git</connection>
<developerConnection>scm:git:https://git-wip-us.apache.org/repos/asf/accumulo.git</developerConnection>
- <tag>${project.version}</tag>
+ <tag>1.5.1</tag>
<url>https://git-wip-us.apache.org/repos/asf?p=accumulo.git</url>
</scm>
<issueManagement>
http://git-wip-us.apache.org/repos/asf/accumulo/blob/3478f71a/proxy/pom.xml
----------------------------------------------------------------------
diff --git a/proxy/pom.xml b/proxy/pom.xml
index 1b2bfa2..626c611 100644
--- a/proxy/pom.xml
+++ b/proxy/pom.xml
@@ -20,7 +20,7 @@
<parent>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-project</artifactId>
- <version>1.5.1-SNAPSHOT</version>
+ <version>1.5.1</version>
</parent>
<artifactId>accumulo-proxy</artifactId>
<name>Proxy</name>
http://git-wip-us.apache.org/repos/asf/accumulo/blob/3478f71a/server/pom.xml
----------------------------------------------------------------------
diff --git a/server/pom.xml b/server/pom.xml
index 52b2f1e..b800c09 100644
--- a/server/pom.xml
+++ b/server/pom.xml
@@ -20,7 +20,7 @@
<parent>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-project</artifactId>
- <version>1.5.1-SNAPSHOT</version>
+ <version>1.5.1</version>
</parent>
<artifactId>accumulo-server</artifactId>
<name>Server</name>
http://git-wip-us.apache.org/repos/asf/accumulo/blob/3478f71a/start/pom.xml
----------------------------------------------------------------------
diff --git a/start/pom.xml b/start/pom.xml
index 5c2c09b..e61fedd 100644
--- a/start/pom.xml
+++ b/start/pom.xml
@@ -20,7 +20,7 @@
<parent>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-project</artifactId>
- <version>1.5.1-SNAPSHOT</version>
+ <version>1.5.1</version>
</parent>
<artifactId>accumulo-start</artifactId>
<name>Start</name>
http://git-wip-us.apache.org/repos/asf/accumulo/blob/3478f71a/test/pom.xml
----------------------------------------------------------------------
diff --git a/test/pom.xml b/test/pom.xml
index 9579581..69c0a72 100644
--- a/test/pom.xml
+++ b/test/pom.xml
@@ -20,7 +20,7 @@
<parent>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-project</artifactId>
- <version>1.5.1-SNAPSHOT</version>
+ <version>1.5.1</version>
</parent>
<artifactId>accumulo-test</artifactId>
<name>Testing</name>
http://git-wip-us.apache.org/repos/asf/accumulo/blob/3478f71a/trace/pom.xml
----------------------------------------------------------------------
diff --git a/trace/pom.xml b/trace/pom.xml
index 650ec1a..de32df7 100644
--- a/trace/pom.xml
+++ b/trace/pom.xml
@@ -20,7 +20,7 @@
<parent>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-project</artifactId>
- <version>1.5.1-SNAPSHOT</version>
+ <version>1.5.1</version>
</parent>
<artifactId>accumulo-trace</artifactId>
<name>Trace</name>
[03/16] git commit: Merge branch '1.4.5-SNAPSHOT' into 1.5.1-SNAPSHOT
(-sours)
Posted by md...@apache.org.
Merge branch '1.4.5-SNAPSHOT' into 1.5.1-SNAPSHOT (-sours)
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/a3c2f61f
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/a3c2f61f
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/a3c2f61f
Branch: refs/heads/master
Commit: a3c2f61f450f21f37c7d71f17d3ff020efc1e823
Parents: 7eeff02 a8bbb91
Author: Mike Drob <md...@cloudera.com>
Authored: Mon Mar 3 12:09:28 2014 -0500
Committer: Mike Drob <md...@cloudera.com>
Committed: Mon Mar 3 12:09:28 2014 -0500
----------------------------------------------------------------------
----------------------------------------------------------------------
[15/16] git commit: Merge branch '1.5.2-SNAPSHOT' into 1.6.0-SNAPSHOT
Posted by md...@apache.org.
Merge branch '1.5.2-SNAPSHOT' into 1.6.0-SNAPSHOT
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/9abf9424
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/9abf9424
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/9abf9424
Branch: refs/heads/1.6.0-SNAPSHOT
Commit: 9abf9424d8d172d10d2fea8d26f257c2a55208c6
Parents: e057181 bd283ae
Author: Mike Drob <md...@cloudera.com>
Authored: Tue Mar 4 09:15:29 2014 -0500
Committer: Mike Drob <md...@cloudera.com>
Committed: Tue Mar 4 09:15:29 2014 -0500
----------------------------------------------------------------------
.../test/continuous/ContinuousBatchWalker.java | 2 +-
.../test/continuous/ContinuousIngest.java | 9 ++--
.../test/continuous/ContinuousQuery.java | 2 +-
.../test/continuous/ContinuousScanner.java | 2 +-
.../test/continuous/ContinuousUtil.java | 49 ++++++++++++++++++++
.../test/continuous/ContinuousWalk.java | 2 +-
6 files changed, 57 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9abf9424/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousBatchWalker.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9abf9424/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousWalk.java
----------------------------------------------------------------------
[11/16] git commit: Merge branch '1.4.5-SNAPSHOT' into 1.5.2-SNAPSHOT
Posted by md...@apache.org.
Merge branch '1.4.5-SNAPSHOT' into 1.5.2-SNAPSHOT
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/bd283aec
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/bd283aec
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/bd283aec
Branch: refs/heads/1.6.0-SNAPSHOT
Commit: bd283aec0aa10d4c9deba8b5df097d48a424780b
Parents: bee78fa 759582b
Author: Mike Drob <md...@cloudera.com>
Authored: Tue Mar 4 09:15:16 2014 -0500
Committer: Mike Drob <md...@cloudera.com>
Committed: Tue Mar 4 09:15:16 2014 -0500
----------------------------------------------------------------------
.../test/continuous/ContinuousBatchWalker.java | 2 +-
.../test/continuous/ContinuousIngest.java | 9 ++--
.../test/continuous/ContinuousQuery.java | 2 +-
.../test/continuous/ContinuousScanner.java | 2 +-
.../test/continuous/ContinuousUtil.java | 49 ++++++++++++++++++++
.../test/continuous/ContinuousWalk.java | 6 +--
6 files changed, 59 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/bd283aec/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousBatchWalker.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/continuous/ContinuousBatchWalker.java
index d021164,0000000..3304d24
mode 100644,000000..100644
--- a/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousBatchWalker.java
+++ b/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousBatchWalker.java
@@@ -1,182 -1,0 +1,182 @@@
+/*
+ * 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.continuous;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.accumulo.core.cli.BatchScannerOpts;
+import org.apache.accumulo.core.cli.ScannerOpts;
+import org.apache.accumulo.core.client.BatchScanner;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Scanner;
+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.security.Authorizations;
+import org.apache.accumulo.core.util.UtilWaitThread;
+import org.apache.hadoop.io.Text;
+
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.validators.PositiveInteger;
+
+public class ContinuousBatchWalker {
+
+ static class Opts extends ContinuousWalk.Opts {
+ @Parameter(names="--numToScan", description="Number rows to scan between sleeps", required=true, validateWith=PositiveInteger.class)
+ long numToScan = 0;
+ }
+
+ public static void main(String[] args) throws Exception {
+
+ Opts opts = new Opts();
+ ScannerOpts scanOpts = new ScannerOpts();
+ BatchScannerOpts bsOpts = new BatchScannerOpts();
+ opts.parseArgs(ContinuousBatchWalker.class.getName(), args, scanOpts, bsOpts);
+
+ Random r = new Random();
+ Authorizations auths = opts.randomAuths.getAuths(r);
+
+ Connector conn = opts.getConnector();
- Scanner scanner = conn.createScanner(opts.getTableName(), auths);
++ Scanner scanner = ContinuousUtil.createScanner(conn, opts.getTableName(), auths);
+ scanner.setBatchSize(scanOpts.scanBatchSize);
+
+ BatchScanner bs = conn.createBatchScanner(opts.getTableName(), auths, bsOpts.scanThreads);
+ bs.setTimeout(bsOpts.scanTimeout, TimeUnit.MILLISECONDS);
+
+ while (true) {
+ Set<Text> batch = getBatch(scanner, opts.min, opts.max, scanOpts.scanBatchSize, r);
+ List<Range> ranges = new ArrayList<Range>(batch.size());
+
+ for (Text row : batch) {
+ ranges.add(new Range(row));
+ }
+
+ runBatchScan(scanOpts.scanBatchSize, bs, batch, ranges);
+
+ UtilWaitThread.sleep(opts.sleepTime);
+ }
+
+ }
+
+ /*
+ * private static void runSequentialScan(Scanner scanner, List<Range> ranges) { Set<Text> srowsSeen = new HashSet<Text>(); long st1 =
+ * System.currentTimeMillis(); int scount = 0; for (Range range : ranges) { scanner.setRange(range);
+ *
+ * for (Entry<Key,Value> entry : scanner) { srowsSeen.add(entry.getKey().getRow()); scount++; } }
+ *
+ *
+ * long st2 = System.currentTimeMillis(); System.out.println("SRQ "+(st2 - st1)+" "+srowsSeen.size() +" "+scount); }
+ */
+
+ private static void runBatchScan(int batchSize, BatchScanner bs, Set<Text> batch, List<Range> ranges) {
+ bs.setRanges(ranges);
+
+ Set<Text> rowsSeen = new HashSet<Text>();
+
+ int count = 0;
+
+ long t1 = System.currentTimeMillis();
+
+ for (Entry<Key,Value> entry : bs) {
+ ContinuousWalk.validate(entry.getKey(), entry.getValue());
+
+ rowsSeen.add(entry.getKey().getRow());
+
+ addRow(batchSize, entry.getValue());
+
+ count++;
+ }
+
+ long t2 = System.currentTimeMillis();
+
+ if (!rowsSeen.equals(batch)) {
+ HashSet<Text> copy1 = new HashSet<Text>(rowsSeen);
+ HashSet<Text> copy2 = new HashSet<Text>(batch);
+
+ copy1.removeAll(batch);
+ copy2.removeAll(rowsSeen);
+
+ System.out.printf("DIF %d %d %d%n", t1, copy1.size(), copy2.size());
+ System.err.printf("DIF %d %d %d%n", t1, copy1.size(), copy2.size());
+ System.err.println("Extra seen : " + copy1);
+ System.err.println("Not seen : " + copy2);
+ } else {
+ System.out.printf("BRQ %d %d %d %d %d%n", t1, (t2 - t1), rowsSeen.size(), count, (int) (rowsSeen.size() / ((t2 - t1) / 1000.0)));
+ }
+
+ }
+
+ private static void addRow(int batchSize, Value v) {
+ byte[] val = v.get();
+
+ int offset = ContinuousWalk.getPrevRowOffset(val);
+ if (offset > 1) {
+ Text prevRow = new Text();
+ prevRow.set(val, offset, 16);
+ if (rowsToQuery.size() < 3 * batchSize) {
+ rowsToQuery.add(prevRow);
+ }
+ }
+ }
+
+ private static HashSet<Text> rowsToQuery = new HashSet<Text>();
+
+ private static Set<Text> getBatch(Scanner scanner, long min, long max, int batchSize, Random r) {
+
+ while (rowsToQuery.size() < batchSize) {
+ byte[] scanStart = ContinuousIngest.genRow(min, max, r);
+ scanner.setRange(new Range(new Text(scanStart), null));
+
+ int count = 0;
+
+ long t1 = System.currentTimeMillis();
+
+ Iterator<Entry<Key,Value>> iter = scanner.iterator();
+ while (iter.hasNext() && rowsToQuery.size() < 3 * batchSize) {
+ Entry<Key,Value> entry = iter.next();
+ ContinuousWalk.validate(entry.getKey(), entry.getValue());
+ addRow(batchSize, entry.getValue());
+ count++;
+ }
+
+ long t2 = System.currentTimeMillis();
+
+ System.out.println("FSB " + t1 + " " + (t2 - t1) + " " + count);
+
+ UtilWaitThread.sleep(100);
+ }
+
+ HashSet<Text> ret = new HashSet<Text>();
+
+ Iterator<Text> iter = rowsToQuery.iterator();
+
+ for (int i = 0; i < batchSize; i++) {
+ ret.add(iter.next());
+ iter.remove();
+ }
+
+ return ret;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/bd283aec/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousIngest.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/continuous/ContinuousIngest.java
index 23cf15d,0000000..e3f0485
mode 100644,000000..100644
--- a/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousIngest.java
+++ b/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousIngest.java
@@@ -1,312 -1,0 +1,311 @@@
+/*
+ * 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.continuous;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+import java.util.UUID;
+import java.util.zip.CRC32;
+import java.util.zip.Checksum;
+
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.cli.BatchWriterOpts;
+import org.apache.accumulo.core.cli.ClientOnDefaultTable;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.MutationsRejectedException;
- import org.apache.accumulo.core.client.TableExistsException;
++import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.security.ColumnVisibility;
+import org.apache.accumulo.core.util.FastFormat;
+import org.apache.accumulo.trace.instrument.CountSampler;
+import org.apache.accumulo.trace.instrument.Trace;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.log4j.FileAppender;
+import org.apache.log4j.Level;
+import org.apache.log4j.Logger;
+import org.apache.log4j.PatternLayout;
+
+import com.beust.jcommander.IStringConverter;
+import com.beust.jcommander.Parameter;
+
+
+public class ContinuousIngest {
+
+ static public class BaseOpts extends ClientOnDefaultTable {
+ public class DebugConverter implements IStringConverter<String> {
+ @Override
+ public String convert(String debugLog) {
+ Logger logger = Logger.getLogger(Constants.CORE_PACKAGE_NAME);
+ logger.setLevel(Level.TRACE);
+ logger.setAdditivity(false);
+ try {
+ logger.addAppender(new FileAppender(new PatternLayout("%d{dd HH:mm:ss,SSS} [%-8c{2}] %-5p: %m%n"), debugLog, true));
+ } catch (IOException ex) {
+ throw new RuntimeException(ex);
+ }
+ return debugLog;
+ }
+ }
+
+ @Parameter(names="--min", description="lowest random row number to use")
+ long min = 0;
+
+ @Parameter(names="--max", description="maximum random row number to use")
+ long max = Long.MAX_VALUE;
+
+ @Parameter(names="--debugLog", description="file to write debugging output", converter=DebugConverter.class)
+ String debugLog = null;
+
+ BaseOpts() { super("ci"); }
+ }
+
+ public static class ShortConverter implements IStringConverter<Short> {
+ @Override
+ public Short convert(String value) {
+ return Short.valueOf(value);
+ }
+ }
+
+ static public class Opts extends BaseOpts {
+ @Parameter(names="--num", description="the number of entries to ingest")
+ long num = Long.MAX_VALUE;
+
+ @Parameter(names="--maxColF", description="maximum column family value to use", converter=ShortConverter.class)
+ short maxColF = Short.MAX_VALUE;
+
+ @Parameter(names="--maxColQ", description="maximum column qualifier value to use", converter=ShortConverter.class)
+ short maxColQ = Short.MAX_VALUE;
+
+ @Parameter(names="--addCheckSum", description="turn on checksums")
+ boolean checksum = false;
+
+ @Parameter(names="--visibilities", description="read the visibilities to ingest with from a file")
+ String visFile = null;
+ }
+
+ private static final byte[] EMPTY_BYTES = new byte[0];
+
+ private static List<ColumnVisibility> visibilities;
+
+ private static void initVisibilities(Opts opts) throws Exception {
+ if (opts.visFile == null) {
+ visibilities = Collections.singletonList(new ColumnVisibility());
+ return;
+ }
+
+ visibilities = new ArrayList<ColumnVisibility>();
+
+ FileSystem fs = FileSystem.get(new Configuration());
+ BufferedReader in = new BufferedReader(new InputStreamReader(fs.open(new Path(opts.visFile)), Constants.UTF8));
+
+ String line;
+
+ while ((line = in.readLine()) != null) {
+ visibilities.add(new ColumnVisibility(line));
+ }
+
+ in.close();
+ }
+
+ private static ColumnVisibility getVisibility(Random rand) {
+ return visibilities.get(rand.nextInt(visibilities.size()));
+ }
+
+ public static void main(String[] args) throws Exception {
+
+ Opts opts = new Opts();
+ BatchWriterOpts bwOpts = new BatchWriterOpts();
+ opts.parseArgs(ContinuousIngest.class.getName(), args, bwOpts);
+
+ initVisibilities(opts);
+
+ if (opts.min < 0 || opts.max < 0 || opts.max <= opts.min) {
+ throw new IllegalArgumentException("bad min and max");
+ }
+ Connector conn = opts.getConnector();
+
- if (!conn.tableOperations().exists(opts.getTableName()))
- try {
- conn.tableOperations().create(opts.getTableName());
- } catch (TableExistsException tee) {}
++ if (!conn.tableOperations().exists(opts.getTableName())) {
++ throw new TableNotFoundException(null, opts.getTableName(), "Consult the README and create the table before starting ingest.");
++ }
+
+ BatchWriter bw = conn.createBatchWriter(opts.getTableName(), bwOpts.getBatchWriterConfig());
+ bw = Trace.wrapAll(bw, new CountSampler(1024));
+
+ Random r = new Random();
+
+ byte[] ingestInstanceId = UUID.randomUUID().toString().getBytes(Constants.UTF8);
+
+ System.out.printf("UUID %d %s%n", System.currentTimeMillis(), new String(ingestInstanceId, Constants.UTF8));
+
+ long count = 0;
+ final int flushInterval = 1000000;
+ final int maxDepth = 25;
+
+ // always want to point back to flushed data. This way the previous item should
+ // always exist in accumulo when verifying data. To do this make insert N point
+ // back to the row from insert (N - flushInterval). The array below is used to keep
+ // track of this.
+ long prevRows[] = new long[flushInterval];
+ long firstRows[] = new long[flushInterval];
+ int firstColFams[] = new int[flushInterval];
+ int firstColQuals[] = new int[flushInterval];
+
+ long lastFlushTime = System.currentTimeMillis();
+
+ out: while (true) {
+ // generate first set of nodes
+ ColumnVisibility cv = getVisibility(r);
+
+ for (int index = 0; index < flushInterval; index++) {
+ long rowLong = genLong(opts.min, opts.max, r);
+ prevRows[index] = rowLong;
+ firstRows[index] = rowLong;
+
+ int cf = r.nextInt(opts.maxColF);
+ int cq = r.nextInt(opts.maxColQ);
+
+ firstColFams[index] = cf;
+ firstColQuals[index] = cq;
+
+ Mutation m = genMutation(rowLong, cf, cq, cv, ingestInstanceId, count, null, r, opts.checksum);
+ count++;
+ bw.addMutation(m);
+ }
+
+ lastFlushTime = flush(bw, count, flushInterval, lastFlushTime);
+ if (count >= opts.num)
+ break out;
+
+ // generate subsequent sets of nodes that link to previous set of nodes
+ for (int depth = 1; depth < maxDepth; depth++) {
+ for (int index = 0; index < flushInterval; index++) {
+ long rowLong = genLong(opts.min, opts.max, r);
+ byte[] prevRow = genRow(prevRows[index]);
+ prevRows[index] = rowLong;
+ Mutation m = genMutation(rowLong, r.nextInt(opts.maxColF), r.nextInt(opts.maxColQ), cv, ingestInstanceId, count, prevRow, r, opts.checksum);
+ count++;
+ bw.addMutation(m);
+ }
+
+ lastFlushTime = flush(bw, count, flushInterval, lastFlushTime);
+ if (count >= opts.num)
+ break out;
+ }
+
+ // create one big linked list, this makes all of the first inserts
+ // point to something
+ for (int index = 0; index < flushInterval - 1; index++) {
+ Mutation m = genMutation(firstRows[index], firstColFams[index], firstColQuals[index], cv, ingestInstanceId, count, genRow(prevRows[index + 1]), r,
+ opts.checksum);
+ count++;
+ bw.addMutation(m);
+ }
+ lastFlushTime = flush(bw, count, flushInterval, lastFlushTime);
+ if (count >= opts.num)
+ break out;
+ }
+
+ bw.close();
+ opts.stopTracing();
+ }
+
+ private static long flush(BatchWriter bw, long count, final int flushInterval, long lastFlushTime) throws MutationsRejectedException {
+ long t1 = System.currentTimeMillis();
+ bw.flush();
+ long t2 = System.currentTimeMillis();
+ System.out.printf("FLUSH %d %d %d %d %d%n", t2, (t2 - lastFlushTime), (t2 - t1), count, flushInterval);
+ lastFlushTime = t2;
+ return lastFlushTime;
+ }
+
+ public static Mutation genMutation(long rowLong, int cfInt, int cqInt, ColumnVisibility cv, byte[] ingestInstanceId, long count, byte[] prevRow, Random r,
+ boolean checksum) {
+ // Adler32 is supposed to be faster, but according to wikipedia is not good for small data.... so used CRC32 instead
+ CRC32 cksum = null;
+
+ byte[] rowString = genRow(rowLong);
+
+ byte[] cfString = FastFormat.toZeroPaddedString(cfInt, 4, 16, EMPTY_BYTES);
+ byte[] cqString = FastFormat.toZeroPaddedString(cqInt, 4, 16, EMPTY_BYTES);
+
+ if (checksum) {
+ cksum = new CRC32();
+ cksum.update(rowString);
+ cksum.update(cfString);
+ cksum.update(cqString);
+ cksum.update(cv.getExpression());
+ }
+
+ Mutation m = new Mutation(new Text(rowString));
+
+ m.put(new Text(cfString), new Text(cqString), cv, createValue(ingestInstanceId, count, prevRow, cksum));
+ return m;
+ }
+
+ public static final long genLong(long min, long max, Random r) {
+ return ((r.nextLong() & 0x7fffffffffffffffl) % (max - min)) + min;
+ }
+
+ static final byte[] genRow(long min, long max, Random r) {
+ return genRow(genLong(min, max, r));
+ }
+
+ static final byte[] genRow(long rowLong) {
+ return FastFormat.toZeroPaddedString(rowLong, 16, 16, EMPTY_BYTES);
+ }
+
+ private static Value createValue(byte[] ingestInstanceId, long count, byte[] prevRow, Checksum cksum) {
+ int dataLen = ingestInstanceId.length + 16 + (prevRow == null ? 0 : prevRow.length) + 3;
+ if (cksum != null)
+ dataLen += 8;
+ byte val[] = new byte[dataLen];
+ System.arraycopy(ingestInstanceId, 0, val, 0, ingestInstanceId.length);
+ int index = ingestInstanceId.length;
+ val[index++] = ':';
+ int added = FastFormat.toZeroPaddedString(val, index, count, 16, 16, EMPTY_BYTES);
+ if (added != 16)
+ throw new RuntimeException(" " + added);
+ index += 16;
+ val[index++] = ':';
+ if (prevRow != null) {
+ System.arraycopy(prevRow, 0, val, index, prevRow.length);
+ index += prevRow.length;
+ }
+
+ val[index++] = ':';
+
+ if (cksum != null) {
+ cksum.update(val, 0, index);
+ cksum.getValue();
+ FastFormat.toZeroPaddedString(val, index, cksum.getValue(), 8, 16, EMPTY_BYTES);
+ }
+
+ // System.out.println("val "+new String(val));
+
+ return new Value(val);
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/bd283aec/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousQuery.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/continuous/ContinuousQuery.java
index 117c136,0000000..4bbc85f
mode 100644,000000..100644
--- a/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousQuery.java
+++ b/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousQuery.java
@@@ -1,71 -1,0 +1,71 @@@
+/*
+ * 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.continuous;
+
+import java.util.Map.Entry;
+import java.util.Random;
+
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.cli.ScannerOpts;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Scanner;
+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.test.continuous.ContinuousIngest.BaseOpts;
+import org.apache.hadoop.io.Text;
+
+import com.beust.jcommander.Parameter;
+
+public class ContinuousQuery {
+
+ public static class Opts extends BaseOpts {
+ @Parameter(names="--sleep", description="the time to wait between queries", converter=TimeConverter.class)
+ long sleepTime = 100;
+ }
+
+ public static void main(String[] args) throws Exception {
+ Opts opts = new Opts();
+ ScannerOpts scanOpts = new ScannerOpts();
+ opts.parseArgs(ContinuousQuery.class.getName(), args, scanOpts);
+
+ Connector conn = opts.getConnector();
- Scanner scanner = conn.createScanner(opts.getTableName(), opts.auths);
++ Scanner scanner = ContinuousUtil.createScanner(conn, opts.getTableName(), opts.auths);
+ scanner.setBatchSize(scanOpts.scanBatchSize);
+
+ Random r = new Random();
+
+ while (true) {
+ byte[] row = ContinuousIngest.genRow(opts.min, opts.max, r);
+
+ int count = 0;
+
+ long t1 = System.currentTimeMillis();
+ scanner.setRange(new Range(new Text(row)));
+ for (Entry<Key,Value> entry : scanner) {
+ ContinuousWalk.validate(entry.getKey(), entry.getValue());
+ count++;
+ }
+ long t2 = System.currentTimeMillis();
+
+ System.out.printf("SRQ %d %s %d %d%n", t1, new String(row, Constants.UTF8), (t2 - t1), count);
+
+ if (opts.sleepTime > 0)
+ Thread.sleep(opts.sleepTime);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/bd283aec/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousScanner.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/continuous/ContinuousScanner.java
index c331bab,0000000..fcc8fec
mode 100644,000000..100644
--- a/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousScanner.java
+++ b/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousScanner.java
@@@ -1,104 -1,0 +1,104 @@@
+/*
+ * 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.continuous;
+
+import java.util.Iterator;
+import java.util.Map.Entry;
+import java.util.Random;
+
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.cli.ScannerOpts;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Scanner;
+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.security.Authorizations;
+import org.apache.accumulo.core.util.UtilWaitThread;
+import org.apache.hadoop.io.Text;
+
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.validators.PositiveInteger;
+
+public class ContinuousScanner {
+
+ static class Opts extends ContinuousWalk.Opts {
+ @Parameter(names="--numToScan", description="Number rows to scan between sleeps", required=true, validateWith=PositiveInteger.class)
+ long numToScan = 0;
+ }
+
+ public static void main(String[] args) throws Exception {
+ Opts opts = new Opts();
+ ScannerOpts scanOpts = new ScannerOpts();
+ opts.parseArgs(ContinuousScanner.class.getName(), args, scanOpts);
+
+ Random r = new Random();
+
+ long distance = 1000000000000l;
+
+ Connector conn = opts.getConnector();
+ Authorizations auths = opts.randomAuths.getAuths(r);
- Scanner scanner = conn.createScanner(opts.getTableName(), auths);
++ Scanner scanner = ContinuousUtil.createScanner(conn, opts.getTableName(), auths);
+ scanner.setBatchSize(scanOpts.scanBatchSize);
+
+ double delta = Math.min(.05, .05 / (opts.numToScan / 1000.0));
+
+ while (true) {
+ long startRow = ContinuousIngest.genLong(opts.min, opts.max - distance, r);
+ byte[] scanStart = ContinuousIngest.genRow(startRow);
+ byte[] scanStop = ContinuousIngest.genRow(startRow + distance);
+
+ scanner.setRange(new Range(new Text(scanStart), new Text(scanStop)));
+
+ int count = 0;
+ Iterator<Entry<Key,Value>> iter = scanner.iterator();
+
+ long t1 = System.currentTimeMillis();
+
+ while (iter.hasNext()) {
+ Entry<Key,Value> entry = iter.next();
+ ContinuousWalk.validate(entry.getKey(), entry.getValue());
+ count++;
+ }
+
+ long t2 = System.currentTimeMillis();
+
+ // System.out.println("P1 " +count +" "+((1-delta) * numToScan)+" "+((1+delta) * numToScan)+" "+numToScan);
+
+ if (count < (1 - delta) * opts.numToScan || count > (1 + delta) * opts.numToScan) {
+ if (count == 0) {
+ distance = distance * 10;
+ if (distance < 0)
+ distance = 1000000000000l;
+ } else {
+ double ratio = (double) opts.numToScan / count;
+ // move ratio closer to 1 to make change slower
+ ratio = ratio - (ratio - 1.0) * (2.0 / 3.0);
+ distance = (long) (ratio * distance);
+ }
+
+ // System.out.println("P2 "+delta +" "+numToScan+" "+distance+" "+((double)numToScan/count ));
+ }
+
+ System.out.printf("SCN %d %s %d %d%n", t1, new String(scanStart, Constants.UTF8), (t2 - t1), count);
+
+ if (opts.sleepTime > 0)
+ UtilWaitThread.sleep(opts.sleepTime);
+ }
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/bd283aec/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousUtil.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/continuous/ContinuousUtil.java
index 0000000,0000000..a8b2930
new file mode 100644
--- /dev/null
+++ b/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousUtil.java
@@@ -1,0 -1,0 +1,49 @@@
++/*
++ * 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.continuous;
++
++import org.apache.accumulo.core.client.Connector;
++import org.apache.accumulo.core.client.Scanner;
++import org.apache.accumulo.core.client.TableNotFoundException;
++import org.apache.accumulo.core.security.Authorizations;
++
++/**
++ * Useful utility methods common to the Continuous test suite.
++ */
++final class ContinuousUtil {
++ private ContinuousUtil() {}
++
++ /**
++ * Attempt to create a table scanner, or fail if the table does not exist.
++ *
++ * @param connector
++ * A populated connector object
++ * @param table
++ * The table name to scan over
++ * @param auths
++ * The authorizations to use for the scanner
++ * @return a scanner for the requested table
++ * @throws TableNotFoundException
++ * If the table does not exist
++ */
++ static Scanner createScanner(Connector connector, String table, Authorizations auths) throws TableNotFoundException {
++ if (!connector.tableOperations().exists(table)) {
++ throw new TableNotFoundException(null, table, "Consult the README and create the table before starting test processes.");
++ }
++ return connector.createScanner(table, auths);
++ }
++}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/bd283aec/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousWalk.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/continuous/ContinuousWalk.java
index 4032dfa,0000000..34a5e9b
mode 100644,000000..100644
--- a/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousWalk.java
+++ b/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousWalk.java
@@@ -1,237 -1,0 +1,237 @@@
+/*
+ * 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.continuous;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Random;
+import java.util.zip.CRC32;
+
- import org.apache.accumulo.trace.instrument.Span;
- import org.apache.accumulo.trace.instrument.Trace;
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Scanner;
+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.security.Authorizations;
++import org.apache.accumulo.trace.instrument.Span;
++import org.apache.accumulo.trace.instrument.Trace;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+
+import com.beust.jcommander.IStringConverter;
+import com.beust.jcommander.Parameter;
+
+
+public class ContinuousWalk {
+
+ static public class Opts extends ContinuousQuery.Opts {
+ class RandomAuthsConverter implements IStringConverter<RandomAuths> {
+ @Override
+ public RandomAuths convert(String value) {
+ try {
+ return new RandomAuths(value);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+ @Parameter(names="--authsFile", description="read the authorities to use from a file")
+ RandomAuths randomAuths = new RandomAuths();
+ }
+
+ static class BadChecksumException extends RuntimeException {
+ private static final long serialVersionUID = 1L;
+
+ public BadChecksumException(String msg) {
+ super(msg);
+ }
+
+ }
+
+ static class RandomAuths {
+ private List<Authorizations> auths;
+
+ RandomAuths() {
+ auths = Collections.singletonList(Constants.NO_AUTHS);
+ }
+
+ RandomAuths(String file) throws IOException {
+ if (file == null) {
+ auths = Collections.singletonList(Constants.NO_AUTHS);
+ return;
+ }
+
+ auths = new ArrayList<Authorizations>();
+
+ FileSystem fs = FileSystem.get(new Configuration());
+ BufferedReader in = new BufferedReader(new InputStreamReader(fs.open(new Path(file)), Constants.UTF8));
+ try {
+ String line;
+ while ((line = in.readLine()) != null) {
+ auths.add(new Authorizations(line.split(",")));
+ }
+ } finally {
+ in.close();
+ }
+ }
+
+ Authorizations getAuths(Random r) {
+ return auths.get(r.nextInt(auths.size()));
+ }
+ }
+
+ public static void main(String[] args) throws Exception {
+ Opts opts = new Opts();
+ opts.parseArgs(ContinuousWalk.class.getName(), args);
+
+ Connector conn = opts.getConnector();
+
+ Random r = new Random();
+
+ ArrayList<Value> values = new ArrayList<Value>();
+
+ while (true) {
- Scanner scanner = conn.createScanner(opts.getTableName(), opts.randomAuths.getAuths(r));
++ Scanner scanner = ContinuousUtil.createScanner(conn, opts.getTableName(), opts.randomAuths.getAuths(r));
+ String row = findAStartRow(opts.min, opts.max, scanner, r);
+
+ while (row != null) {
+
+ values.clear();
+
+ long t1 = System.currentTimeMillis();
+ Span span = Trace.on("walk");
+ try {
+ scanner.setRange(new Range(new Text(row)));
+ for (Entry<Key,Value> entry : scanner) {
+ validate(entry.getKey(), entry.getValue());
+ values.add(entry.getValue());
+ }
+ } finally {
+ span.stop();
+ }
+ long t2 = System.currentTimeMillis();
+
+ System.out.printf("SRQ %d %s %d %d%n", t1, row, (t2 - t1), values.size());
+
+ if (values.size() > 0) {
+ row = getPrevRow(values.get(r.nextInt(values.size())));
+ } else {
+ System.out.printf("MIS %d %s%n", t1, row);
+ System.err.printf("MIS %d %s%n", t1, row);
+ row = null;
+ }
+
+ if (opts.sleepTime > 0)
+ Thread.sleep(opts.sleepTime);
+ }
+
+ if (opts.sleepTime > 0)
+ Thread.sleep(opts.sleepTime);
+ }
+ }
+
+ private static String findAStartRow(long min, long max, Scanner scanner, Random r) {
+
+ byte[] scanStart = ContinuousIngest.genRow(min, max, r);
+ scanner.setRange(new Range(new Text(scanStart), null));
+ scanner.setBatchSize(100);
+
+ int count = 0;
+ String pr = null;
+
+ long t1 = System.currentTimeMillis();
+
+ for (Entry<Key,Value> entry : scanner) {
+ validate(entry.getKey(), entry.getValue());
+ pr = getPrevRow(entry.getValue());
+ count++;
+ if (pr != null)
+ break;
+ }
+
+ long t2 = System.currentTimeMillis();
+
+ System.out.printf("FSR %d %s %d %d%n", t1, new String(scanStart, Constants.UTF8), (t2 - t1), count);
+
+ return pr;
+ }
+
+ static int getPrevRowOffset(byte val[]) {
+ if (val.length == 0)
+ throw new IllegalArgumentException();
+ if (val[53] != ':')
+ throw new IllegalArgumentException(new String(val, Constants.UTF8));
+
+ // prev row starts at 54
+ if (val[54] != ':') {
+ if (val[54 + 16] != ':')
+ throw new IllegalArgumentException(new String(val, Constants.UTF8));
+ return 54;
+ }
+
+ return -1;
+ }
+
+ static String getPrevRow(Value value) {
+
+ byte[] val = value.get();
+ int offset = getPrevRowOffset(val);
+ if (offset > 0) {
+ return new String(val, offset, 16, Constants.UTF8);
+ }
+
+ return null;
+ }
+
+ static int getChecksumOffset(byte val[]) {
+ if (val[val.length - 1] != ':') {
+ if (val[val.length - 9] != ':')
+ throw new IllegalArgumentException(new String(val, Constants.UTF8));
+ return val.length - 8;
+ }
+
+ return -1;
+ }
+
+ static void validate(Key key, Value value) throws BadChecksumException {
+ int ckOff = getChecksumOffset(value.get());
+ if (ckOff < 0)
+ return;
+
+ long storedCksum = Long.parseLong(new String(value.get(), ckOff, 8, Constants.UTF8), 16);
+
+ CRC32 cksum = new CRC32();
+
+ cksum.update(key.getRowData().toArray());
+ cksum.update(key.getColumnFamilyData().toArray());
+ cksum.update(key.getColumnQualifierData().toArray());
+ cksum.update(key.getColumnVisibilityData().toArray());
+ cksum.update(value.get(), 0, ckOff);
+
+ if (cksum.getValue() != storedCksum) {
+ throw new BadChecksumException("Checksum invalid " + key + " " + value);
+ }
+ }
+}
[16/16] git commit: Merge branch '1.6.0-SNAPSHOT'
Posted by md...@apache.org.
Merge branch '1.6.0-SNAPSHOT'
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/cf61a3c2
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/cf61a3c2
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/cf61a3c2
Branch: refs/heads/master
Commit: cf61a3c2e3e43f31975536e2ceb5b5023dc32148
Parents: 11d724e 9abf942
Author: Mike Drob <md...@cloudera.com>
Authored: Tue Mar 4 09:15:41 2014 -0500
Committer: Mike Drob <md...@cloudera.com>
Committed: Tue Mar 4 09:15:41 2014 -0500
----------------------------------------------------------------------
.../test/continuous/ContinuousBatchWalker.java | 2 +-
.../test/continuous/ContinuousIngest.java | 9 ++--
.../test/continuous/ContinuousQuery.java | 2 +-
.../test/continuous/ContinuousScanner.java | 2 +-
.../test/continuous/ContinuousUtil.java | 49 ++++++++++++++++++++
.../test/continuous/ContinuousWalk.java | 2 +-
6 files changed, 57 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
[14/16] git commit: Merge branch '1.5.2-SNAPSHOT' into 1.6.0-SNAPSHOT
Posted by md...@apache.org.
Merge branch '1.5.2-SNAPSHOT' into 1.6.0-SNAPSHOT
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/9abf9424
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/9abf9424
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/9abf9424
Branch: refs/heads/master
Commit: 9abf9424d8d172d10d2fea8d26f257c2a55208c6
Parents: e057181 bd283ae
Author: Mike Drob <md...@cloudera.com>
Authored: Tue Mar 4 09:15:29 2014 -0500
Committer: Mike Drob <md...@cloudera.com>
Committed: Tue Mar 4 09:15:29 2014 -0500
----------------------------------------------------------------------
.../test/continuous/ContinuousBatchWalker.java | 2 +-
.../test/continuous/ContinuousIngest.java | 9 ++--
.../test/continuous/ContinuousQuery.java | 2 +-
.../test/continuous/ContinuousScanner.java | 2 +-
.../test/continuous/ContinuousUtil.java | 49 ++++++++++++++++++++
.../test/continuous/ContinuousWalk.java | 2 +-
6 files changed, 57 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9abf9424/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousBatchWalker.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9abf9424/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousWalk.java
----------------------------------------------------------------------
[04/16] git commit: ACCUMULO-2170 updating snapshot to 1.5.2-SNAPSHOT
Posted by md...@apache.org.
ACCUMULO-2170 updating snapshot to 1.5.2-SNAPSHOT
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/d095b21f
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/d095b21f
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/d095b21f
Branch: refs/heads/master
Commit: d095b21f965b8f45f6df45ceb7deb0b0beaac8e8
Parents: a3c2f61
Author: John Vines <vi...@apache.org>
Authored: Mon Mar 3 14:36:32 2014 -0500
Committer: John Vines <vi...@apache.org>
Committed: Mon Mar 3 14:36:32 2014 -0500
----------------------------------------------------------------------
assemble/pom.xml | 2 +-
core/pom.xml | 2 +-
examples/pom.xml | 2 +-
examples/simple/pom.xml | 2 +-
fate/pom.xml | 2 +-
minicluster/pom.xml | 2 +-
pom.xml | 2 +-
proxy/pom.xml | 2 +-
server/pom.xml | 2 +-
start/pom.xml | 2 +-
test/pom.xml | 2 +-
trace/pom.xml | 2 +-
12 files changed, 12 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/d095b21f/assemble/pom.xml
----------------------------------------------------------------------
diff --git a/assemble/pom.xml b/assemble/pom.xml
index 40b77cc..6a20967 100644
--- a/assemble/pom.xml
+++ b/assemble/pom.xml
@@ -20,7 +20,7 @@
<parent>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-project</artifactId>
- <version>1.5.1-SNAPSHOT</version>
+ <version>1.5.2-SNAPSHOT</version>
</parent>
<artifactId>accumulo</artifactId>
<packaging>pom</packaging>
http://git-wip-us.apache.org/repos/asf/accumulo/blob/d095b21f/core/pom.xml
----------------------------------------------------------------------
diff --git a/core/pom.xml b/core/pom.xml
index d02a3cd..6e3d917 100644
--- a/core/pom.xml
+++ b/core/pom.xml
@@ -20,7 +20,7 @@
<parent>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-project</artifactId>
- <version>1.5.1-SNAPSHOT</version>
+ <version>1.5.2-SNAPSHOT</version>
</parent>
<artifactId>accumulo-core</artifactId>
<name>Core</name>
http://git-wip-us.apache.org/repos/asf/accumulo/blob/d095b21f/examples/pom.xml
----------------------------------------------------------------------
diff --git a/examples/pom.xml b/examples/pom.xml
index ec39e30..fef685d 100644
--- a/examples/pom.xml
+++ b/examples/pom.xml
@@ -20,7 +20,7 @@
<parent>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-project</artifactId>
- <version>1.5.1-SNAPSHOT</version>
+ <version>1.5.2-SNAPSHOT</version>
</parent>
<artifactId>accumulo-examples</artifactId>
<packaging>pom</packaging>
http://git-wip-us.apache.org/repos/asf/accumulo/blob/d095b21f/examples/simple/pom.xml
----------------------------------------------------------------------
diff --git a/examples/simple/pom.xml b/examples/simple/pom.xml
index d513e68..2b6ab69 100644
--- a/examples/simple/pom.xml
+++ b/examples/simple/pom.xml
@@ -20,7 +20,7 @@
<parent>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-examples</artifactId>
- <version>1.5.1-SNAPSHOT</version>
+ <version>1.5.2-SNAPSHOT</version>
</parent>
<artifactId>accumulo-examples-simple</artifactId>
<name>Simple Examples</name>
http://git-wip-us.apache.org/repos/asf/accumulo/blob/d095b21f/fate/pom.xml
----------------------------------------------------------------------
diff --git a/fate/pom.xml b/fate/pom.xml
index 30c4973..433e61a 100644
--- a/fate/pom.xml
+++ b/fate/pom.xml
@@ -20,7 +20,7 @@
<parent>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-project</artifactId>
- <version>1.5.1-SNAPSHOT</version>
+ <version>1.5.2-SNAPSHOT</version>
</parent>
<artifactId>accumulo-fate</artifactId>
<name>Fate</name>
http://git-wip-us.apache.org/repos/asf/accumulo/blob/d095b21f/minicluster/pom.xml
----------------------------------------------------------------------
diff --git a/minicluster/pom.xml b/minicluster/pom.xml
index 32848d8..9d2b7cb 100644
--- a/minicluster/pom.xml
+++ b/minicluster/pom.xml
@@ -20,7 +20,7 @@
<parent>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-project</artifactId>
- <version>1.5.1-SNAPSHOT</version>
+ <version>1.5.2-SNAPSHOT</version>
</parent>
<artifactId>accumulo-minicluster</artifactId>
<name>MiniCluster</name>
http://git-wip-us.apache.org/repos/asf/accumulo/blob/d095b21f/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 8ae58ed..9fd5bfa 100644
--- a/pom.xml
+++ b/pom.xml
@@ -24,7 +24,7 @@
</parent>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-project</artifactId>
- <version>1.5.1-SNAPSHOT</version>
+ <version>1.5.2-SNAPSHOT</version>
<packaging>pom</packaging>
<name>Apache Accumulo Project</name>
<description>Apache Accumulo is a sorted, distributed key/value store based on Google's BigTable design. It is built on top of Apache Hadoop, Zookeeper, and Thrift. It features a few novel improvements on the BigTable design in the form of cell-level access labels and a server-side programming mechanism that can modify key/value pairs at various points in the data management process.</description>
http://git-wip-us.apache.org/repos/asf/accumulo/blob/d095b21f/proxy/pom.xml
----------------------------------------------------------------------
diff --git a/proxy/pom.xml b/proxy/pom.xml
index 1b2bfa2..9661104 100644
--- a/proxy/pom.xml
+++ b/proxy/pom.xml
@@ -20,7 +20,7 @@
<parent>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-project</artifactId>
- <version>1.5.1-SNAPSHOT</version>
+ <version>1.5.2-SNAPSHOT</version>
</parent>
<artifactId>accumulo-proxy</artifactId>
<name>Proxy</name>
http://git-wip-us.apache.org/repos/asf/accumulo/blob/d095b21f/server/pom.xml
----------------------------------------------------------------------
diff --git a/server/pom.xml b/server/pom.xml
index 52b2f1e..667b864 100644
--- a/server/pom.xml
+++ b/server/pom.xml
@@ -20,7 +20,7 @@
<parent>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-project</artifactId>
- <version>1.5.1-SNAPSHOT</version>
+ <version>1.5.2-SNAPSHOT</version>
</parent>
<artifactId>accumulo-server</artifactId>
<name>Server</name>
http://git-wip-us.apache.org/repos/asf/accumulo/blob/d095b21f/start/pom.xml
----------------------------------------------------------------------
diff --git a/start/pom.xml b/start/pom.xml
index 5c2c09b..5fc9346 100644
--- a/start/pom.xml
+++ b/start/pom.xml
@@ -20,7 +20,7 @@
<parent>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-project</artifactId>
- <version>1.5.1-SNAPSHOT</version>
+ <version>1.5.2-SNAPSHOT</version>
</parent>
<artifactId>accumulo-start</artifactId>
<name>Start</name>
http://git-wip-us.apache.org/repos/asf/accumulo/blob/d095b21f/test/pom.xml
----------------------------------------------------------------------
diff --git a/test/pom.xml b/test/pom.xml
index 9579581..bf0a3f8 100644
--- a/test/pom.xml
+++ b/test/pom.xml
@@ -20,7 +20,7 @@
<parent>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-project</artifactId>
- <version>1.5.1-SNAPSHOT</version>
+ <version>1.5.2-SNAPSHOT</version>
</parent>
<artifactId>accumulo-test</artifactId>
<name>Testing</name>
http://git-wip-us.apache.org/repos/asf/accumulo/blob/d095b21f/trace/pom.xml
----------------------------------------------------------------------
diff --git a/trace/pom.xml b/trace/pom.xml
index 650ec1a..8e7b083 100644
--- a/trace/pom.xml
+++ b/trace/pom.xml
@@ -20,7 +20,7 @@
<parent>
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-project</artifactId>
- <version>1.5.1-SNAPSHOT</version>
+ <version>1.5.2-SNAPSHOT</version>
</parent>
<artifactId>accumulo-trace</artifactId>
<name>Trace</name>
[08/16] git commit: ACCUMULO-2399 Alert user CI table should exist
Posted by md...@apache.org.
ACCUMULO-2399 Alert user CI table should exist
Extract common table checking functionality for continuous clients and
fail fast when the table does not exist. No longer create the table in
ingest to make the requirement explicit.
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/759582b7
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/759582b7
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/759582b7
Branch: refs/heads/1.5.2-SNAPSHOT
Commit: 759582b78d5d72870a4a8c359ef6134c4dd97993
Parents: a8bbb91
Author: Mike Drob <md...@cloudera.com>
Authored: Tue Feb 25 15:40:12 2014 -0500
Committer: Mike Drob <md...@cloudera.com>
Committed: Tue Mar 4 08:36:13 2014 -0500
----------------------------------------------------------------------
.../test/continuous/ContinuousBatchWalker.java | 2 +-
.../test/continuous/ContinuousIngest.java | 9 ++--
.../server/test/continuous/ContinuousQuery.java | 2 +-
.../test/continuous/ContinuousScanner.java | 2 +-
.../server/test/continuous/ContinuousUtil.java | 49 ++++++++++++++++++++
.../server/test/continuous/ContinuousWalk.java | 2 +-
6 files changed, 57 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/759582b7/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousBatchWalker.java
----------------------------------------------------------------------
diff --git a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousBatchWalker.java b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousBatchWalker.java
index 4659aff..fe738b6 100644
--- a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousBatchWalker.java
+++ b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousBatchWalker.java
@@ -98,7 +98,7 @@ public class ContinuousBatchWalker {
Authorizations auths = randomAuths.getAuths(r);
Connector conn = new ZooKeeperInstance(instanceName, zooKeepers).getConnector(user, password.getBytes());
- Scanner scanner = conn.createScanner(table, auths);
+ Scanner scanner = ContinuousUtil.createScanner(conn, table, auths);
BatchScanner bs = conn.createBatchScanner(table, auths, numQueryThreads);
while (true) {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/759582b7/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousIngest.java
----------------------------------------------------------------------
diff --git a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousIngest.java b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousIngest.java
index b63efbc..1699ab4 100644
--- a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousIngest.java
+++ b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousIngest.java
@@ -36,7 +36,7 @@ import org.apache.accumulo.core.client.BatchWriter;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.MutationsRejectedException;
-import org.apache.accumulo.core.client.TableExistsException;
+import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.client.ZooKeeperInstance;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Value;
@@ -150,10 +150,9 @@ public class ContinuousIngest {
String path = ZooUtil.getRoot(instance) + Constants.ZTRACERS;
Tracer.getInstance().addReceiver(new ZooSpanClient(zooKeepers, path, localhost, "cingest", 1000));
- if (!conn.tableOperations().exists(table))
- try {
- conn.tableOperations().create(table);
- } catch (TableExistsException tee) {}
+ if (!conn.tableOperations().exists(table)) {
+ throw new TableNotFoundException(null, table, "Consult the README and create the table before starting ingest.");
+ }
BatchWriter bw = conn.createBatchWriter(table, maxMemory, maxLatency, maxWriteThreads);
bw = Trace.wrapAll(bw, new CountSampler(1024));
http://git-wip-us.apache.org/repos/asf/accumulo/blob/759582b7/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousQuery.java
----------------------------------------------------------------------
diff --git a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousQuery.java b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousQuery.java
index c8ae6ec..accd867 100644
--- a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousQuery.java
+++ b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousQuery.java
@@ -50,7 +50,7 @@ public class ContinuousQuery {
long sleepTime = Long.parseLong(args[7]);
Connector conn = new ZooKeeperInstance(instanceName, zooKeepers).getConnector(user, password.getBytes());
- Scanner scanner = conn.createScanner(table, new Authorizations());
+ Scanner scanner = ContinuousUtil.createScanner(conn, table, new Authorizations());
Random r = new Random();
http://git-wip-us.apache.org/repos/asf/accumulo/blob/759582b7/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousScanner.java
----------------------------------------------------------------------
diff --git a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousScanner.java b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousScanner.java
index 0ac3df6..13a91db 100644
--- a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousScanner.java
+++ b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousScanner.java
@@ -97,7 +97,7 @@ public class ContinuousScanner {
Instance instance = new ZooKeeperInstance(instanceName, zooKeepers);
Connector conn = instance.getConnector(user, password.getBytes());
Authorizations auths = randomAuths.getAuths(r);
- Scanner scanner = conn.createScanner(table, auths);
+ Scanner scanner = ContinuousUtil.createScanner(conn, table, auths);
double delta = Math.min(.05, .05 / (numToScan / 1000.0));
// System.out.println("Delta "+delta);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/759582b7/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousUtil.java
----------------------------------------------------------------------
diff --git a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousUtil.java b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousUtil.java
new file mode 100644
index 0000000..a788b59
--- /dev/null
+++ b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousUtil.java
@@ -0,0 +1,49 @@
+/*
+ * 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.server.test.continuous;
+
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.security.Authorizations;
+
+/**
+ * Useful utility methods common to the Continuous test suite.
+ */
+final class ContinuousUtil {
+ private ContinuousUtil() {}
+
+ /**
+ * Attempt to create a table scanner, or fail if the table does not exist.
+ *
+ * @param connector
+ * A populated connector object
+ * @param table
+ * The table name to scan over
+ * @param auths
+ * The authorizations to use for the scanner
+ * @return a scanner for the requested table
+ * @throws TableNotFoundException
+ * If the table does not exist
+ */
+ static Scanner createScanner(Connector connector, String table, Authorizations auths) throws TableNotFoundException {
+ if (!connector.tableOperations().exists(table)) {
+ throw new TableNotFoundException(null, table, "Consult the README and create the table before starting test processes.");
+ }
+ return connector.createScanner(table, auths);
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/759582b7/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousWalk.java
----------------------------------------------------------------------
diff --git a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousWalk.java b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousWalk.java
index 7d1e7f9..15b6090 100644
--- a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousWalk.java
+++ b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousWalk.java
@@ -154,7 +154,7 @@ public class ContinuousWalk {
ArrayList<Value> values = new ArrayList<Value>();
while (true) {
- Scanner scanner = conn.createScanner(table, randomAuths.getAuths(r));
+ Scanner scanner = ContinuousUtil.createScanner(conn, table, randomAuths.getAuths(r));
String row = findAStartRow(min, max, scanner, r);
while (row != null) {
[07/16] git commit: ACCUMULO-2399 Alert user CI table should exist
Posted by md...@apache.org.
ACCUMULO-2399 Alert user CI table should exist
Extract common table checking functionality for continuous clients and
fail fast when the table does not exist. No longer create the table in
ingest to make the requirement explicit.
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/759582b7
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/759582b7
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/759582b7
Branch: refs/heads/1.6.0-SNAPSHOT
Commit: 759582b78d5d72870a4a8c359ef6134c4dd97993
Parents: a8bbb91
Author: Mike Drob <md...@cloudera.com>
Authored: Tue Feb 25 15:40:12 2014 -0500
Committer: Mike Drob <md...@cloudera.com>
Committed: Tue Mar 4 08:36:13 2014 -0500
----------------------------------------------------------------------
.../test/continuous/ContinuousBatchWalker.java | 2 +-
.../test/continuous/ContinuousIngest.java | 9 ++--
.../server/test/continuous/ContinuousQuery.java | 2 +-
.../test/continuous/ContinuousScanner.java | 2 +-
.../server/test/continuous/ContinuousUtil.java | 49 ++++++++++++++++++++
.../server/test/continuous/ContinuousWalk.java | 2 +-
6 files changed, 57 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/759582b7/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousBatchWalker.java
----------------------------------------------------------------------
diff --git a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousBatchWalker.java b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousBatchWalker.java
index 4659aff..fe738b6 100644
--- a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousBatchWalker.java
+++ b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousBatchWalker.java
@@ -98,7 +98,7 @@ public class ContinuousBatchWalker {
Authorizations auths = randomAuths.getAuths(r);
Connector conn = new ZooKeeperInstance(instanceName, zooKeepers).getConnector(user, password.getBytes());
- Scanner scanner = conn.createScanner(table, auths);
+ Scanner scanner = ContinuousUtil.createScanner(conn, table, auths);
BatchScanner bs = conn.createBatchScanner(table, auths, numQueryThreads);
while (true) {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/759582b7/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousIngest.java
----------------------------------------------------------------------
diff --git a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousIngest.java b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousIngest.java
index b63efbc..1699ab4 100644
--- a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousIngest.java
+++ b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousIngest.java
@@ -36,7 +36,7 @@ import org.apache.accumulo.core.client.BatchWriter;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.MutationsRejectedException;
-import org.apache.accumulo.core.client.TableExistsException;
+import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.client.ZooKeeperInstance;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Value;
@@ -150,10 +150,9 @@ public class ContinuousIngest {
String path = ZooUtil.getRoot(instance) + Constants.ZTRACERS;
Tracer.getInstance().addReceiver(new ZooSpanClient(zooKeepers, path, localhost, "cingest", 1000));
- if (!conn.tableOperations().exists(table))
- try {
- conn.tableOperations().create(table);
- } catch (TableExistsException tee) {}
+ if (!conn.tableOperations().exists(table)) {
+ throw new TableNotFoundException(null, table, "Consult the README and create the table before starting ingest.");
+ }
BatchWriter bw = conn.createBatchWriter(table, maxMemory, maxLatency, maxWriteThreads);
bw = Trace.wrapAll(bw, new CountSampler(1024));
http://git-wip-us.apache.org/repos/asf/accumulo/blob/759582b7/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousQuery.java
----------------------------------------------------------------------
diff --git a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousQuery.java b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousQuery.java
index c8ae6ec..accd867 100644
--- a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousQuery.java
+++ b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousQuery.java
@@ -50,7 +50,7 @@ public class ContinuousQuery {
long sleepTime = Long.parseLong(args[7]);
Connector conn = new ZooKeeperInstance(instanceName, zooKeepers).getConnector(user, password.getBytes());
- Scanner scanner = conn.createScanner(table, new Authorizations());
+ Scanner scanner = ContinuousUtil.createScanner(conn, table, new Authorizations());
Random r = new Random();
http://git-wip-us.apache.org/repos/asf/accumulo/blob/759582b7/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousScanner.java
----------------------------------------------------------------------
diff --git a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousScanner.java b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousScanner.java
index 0ac3df6..13a91db 100644
--- a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousScanner.java
+++ b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousScanner.java
@@ -97,7 +97,7 @@ public class ContinuousScanner {
Instance instance = new ZooKeeperInstance(instanceName, zooKeepers);
Connector conn = instance.getConnector(user, password.getBytes());
Authorizations auths = randomAuths.getAuths(r);
- Scanner scanner = conn.createScanner(table, auths);
+ Scanner scanner = ContinuousUtil.createScanner(conn, table, auths);
double delta = Math.min(.05, .05 / (numToScan / 1000.0));
// System.out.println("Delta "+delta);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/759582b7/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousUtil.java
----------------------------------------------------------------------
diff --git a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousUtil.java b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousUtil.java
new file mode 100644
index 0000000..a788b59
--- /dev/null
+++ b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousUtil.java
@@ -0,0 +1,49 @@
+/*
+ * 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.server.test.continuous;
+
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.security.Authorizations;
+
+/**
+ * Useful utility methods common to the Continuous test suite.
+ */
+final class ContinuousUtil {
+ private ContinuousUtil() {}
+
+ /**
+ * Attempt to create a table scanner, or fail if the table does not exist.
+ *
+ * @param connector
+ * A populated connector object
+ * @param table
+ * The table name to scan over
+ * @param auths
+ * The authorizations to use for the scanner
+ * @return a scanner for the requested table
+ * @throws TableNotFoundException
+ * If the table does not exist
+ */
+ static Scanner createScanner(Connector connector, String table, Authorizations auths) throws TableNotFoundException {
+ if (!connector.tableOperations().exists(table)) {
+ throw new TableNotFoundException(null, table, "Consult the README and create the table before starting test processes.");
+ }
+ return connector.createScanner(table, auths);
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/759582b7/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousWalk.java
----------------------------------------------------------------------
diff --git a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousWalk.java b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousWalk.java
index 7d1e7f9..15b6090 100644
--- a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousWalk.java
+++ b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousWalk.java
@@ -154,7 +154,7 @@ public class ContinuousWalk {
ArrayList<Value> values = new ArrayList<Value>();
while (true) {
- Scanner scanner = conn.createScanner(table, randomAuths.getAuths(r));
+ Scanner scanner = ContinuousUtil.createScanner(conn, table, randomAuths.getAuths(r));
String row = findAStartRow(min, max, scanner, r);
while (row != null) {
[05/16] git commit: Merge tag '1.5.1' into 1.5.2-SNAPSHOT
Posted by md...@apache.org.
Merge tag '1.5.1' into 1.5.2-SNAPSHOT
Signing 1.5.1 release
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/bee78fa6
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/bee78fa6
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/bee78fa6
Branch: refs/heads/master
Commit: bee78fa69f03cf1910d3b227e474d6b486f3c0a9
Parents: d095b21 3478f71
Author: John Vines <vi...@apache.org>
Authored: Mon Mar 3 14:39:17 2014 -0500
Committer: John Vines <vi...@apache.org>
Committed: Mon Mar 3 14:39:17 2014 -0500
----------------------------------------------------------------------
----------------------------------------------------------------------
[02/16] git commit: ACCUMULO-2281 copy dependencies to correct dir
Posted by md...@apache.org.
ACCUMULO-2281 copy dependencies to correct dir
Use project.parent.basedir instead of relative paths for copying
dependencies for packaging and assembly.
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/a8bbb916
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/a8bbb916
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/a8bbb916
Branch: refs/heads/master
Commit: a8bbb916c2796d7f2bbc64ef82291034bbb3810d
Parents: 57b2e5c
Author: Mike Drob <md...@cloudera.com>
Authored: Mon Mar 3 12:05:55 2014 -0500
Committer: Mike Drob <md...@cloudera.com>
Committed: Mon Mar 3 12:05:55 2014 -0500
----------------------------------------------------------------------
pom.xml | 2 +-
src/examples/simple/pom.xml | 16 ++++++++++++++++
2 files changed, 17 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/a8bbb916/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 92d8106..4c8e5bf 100644
--- a/pom.xml
+++ b/pom.xml
@@ -126,7 +126,7 @@
<goal>copy-dependencies</goal>
</goals>
<configuration>
- <outputDirectory>../../lib</outputDirectory>
+ <outputDirectory>${project.parent.basedir}/lib</outputDirectory>
<!-- just grab the non-provided runtime dependencies -->
<includeArtifactIds>commons-collections,commons-configuration,commons-io,commons-lang,jline,log4j,libthrift,commons-jci-core,commons-jci-fam,commons-logging,commons-logging-api,guava</includeArtifactIds>
<excludeTransitive>true</excludeTransitive>
http://git-wip-us.apache.org/repos/asf/accumulo/blob/a8bbb916/src/examples/simple/pom.xml
----------------------------------------------------------------------
diff --git a/src/examples/simple/pom.xml b/src/examples/simple/pom.xml
index 6ef7741..44f6b88 100644
--- a/src/examples/simple/pom.xml
+++ b/src/examples/simple/pom.xml
@@ -124,6 +124,22 @@
</configuration>
</plugin>
<plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-dependency-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>copy-dependencies</id>
+ <phase>process-resources</phase>
+ <goals>
+ <goal>copy-dependencies</goal>
+ </goals>
+ <configuration>
+ <outputDirectory>${project.parent.parent.basedir}/lib</outputDirectory>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
<artifactId>maven-source-plugin</artifactId>
<configuration>
<outputDirectory>../../../lib</outputDirectory>
[12/16] git commit: Merge branch '1.4.5-SNAPSHOT' into 1.5.2-SNAPSHOT
Posted by md...@apache.org.
Merge branch '1.4.5-SNAPSHOT' into 1.5.2-SNAPSHOT
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/bd283aec
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/bd283aec
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/bd283aec
Branch: refs/heads/1.5.2-SNAPSHOT
Commit: bd283aec0aa10d4c9deba8b5df097d48a424780b
Parents: bee78fa 759582b
Author: Mike Drob <md...@cloudera.com>
Authored: Tue Mar 4 09:15:16 2014 -0500
Committer: Mike Drob <md...@cloudera.com>
Committed: Tue Mar 4 09:15:16 2014 -0500
----------------------------------------------------------------------
.../test/continuous/ContinuousBatchWalker.java | 2 +-
.../test/continuous/ContinuousIngest.java | 9 ++--
.../test/continuous/ContinuousQuery.java | 2 +-
.../test/continuous/ContinuousScanner.java | 2 +-
.../test/continuous/ContinuousUtil.java | 49 ++++++++++++++++++++
.../test/continuous/ContinuousWalk.java | 6 +--
6 files changed, 59 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/bd283aec/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousBatchWalker.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/continuous/ContinuousBatchWalker.java
index d021164,0000000..3304d24
mode 100644,000000..100644
--- a/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousBatchWalker.java
+++ b/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousBatchWalker.java
@@@ -1,182 -1,0 +1,182 @@@
+/*
+ * 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.continuous;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.accumulo.core.cli.BatchScannerOpts;
+import org.apache.accumulo.core.cli.ScannerOpts;
+import org.apache.accumulo.core.client.BatchScanner;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Scanner;
+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.security.Authorizations;
+import org.apache.accumulo.core.util.UtilWaitThread;
+import org.apache.hadoop.io.Text;
+
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.validators.PositiveInteger;
+
+public class ContinuousBatchWalker {
+
+ static class Opts extends ContinuousWalk.Opts {
+ @Parameter(names="--numToScan", description="Number rows to scan between sleeps", required=true, validateWith=PositiveInteger.class)
+ long numToScan = 0;
+ }
+
+ public static void main(String[] args) throws Exception {
+
+ Opts opts = new Opts();
+ ScannerOpts scanOpts = new ScannerOpts();
+ BatchScannerOpts bsOpts = new BatchScannerOpts();
+ opts.parseArgs(ContinuousBatchWalker.class.getName(), args, scanOpts, bsOpts);
+
+ Random r = new Random();
+ Authorizations auths = opts.randomAuths.getAuths(r);
+
+ Connector conn = opts.getConnector();
- Scanner scanner = conn.createScanner(opts.getTableName(), auths);
++ Scanner scanner = ContinuousUtil.createScanner(conn, opts.getTableName(), auths);
+ scanner.setBatchSize(scanOpts.scanBatchSize);
+
+ BatchScanner bs = conn.createBatchScanner(opts.getTableName(), auths, bsOpts.scanThreads);
+ bs.setTimeout(bsOpts.scanTimeout, TimeUnit.MILLISECONDS);
+
+ while (true) {
+ Set<Text> batch = getBatch(scanner, opts.min, opts.max, scanOpts.scanBatchSize, r);
+ List<Range> ranges = new ArrayList<Range>(batch.size());
+
+ for (Text row : batch) {
+ ranges.add(new Range(row));
+ }
+
+ runBatchScan(scanOpts.scanBatchSize, bs, batch, ranges);
+
+ UtilWaitThread.sleep(opts.sleepTime);
+ }
+
+ }
+
+ /*
+ * private static void runSequentialScan(Scanner scanner, List<Range> ranges) { Set<Text> srowsSeen = new HashSet<Text>(); long st1 =
+ * System.currentTimeMillis(); int scount = 0; for (Range range : ranges) { scanner.setRange(range);
+ *
+ * for (Entry<Key,Value> entry : scanner) { srowsSeen.add(entry.getKey().getRow()); scount++; } }
+ *
+ *
+ * long st2 = System.currentTimeMillis(); System.out.println("SRQ "+(st2 - st1)+" "+srowsSeen.size() +" "+scount); }
+ */
+
+ private static void runBatchScan(int batchSize, BatchScanner bs, Set<Text> batch, List<Range> ranges) {
+ bs.setRanges(ranges);
+
+ Set<Text> rowsSeen = new HashSet<Text>();
+
+ int count = 0;
+
+ long t1 = System.currentTimeMillis();
+
+ for (Entry<Key,Value> entry : bs) {
+ ContinuousWalk.validate(entry.getKey(), entry.getValue());
+
+ rowsSeen.add(entry.getKey().getRow());
+
+ addRow(batchSize, entry.getValue());
+
+ count++;
+ }
+
+ long t2 = System.currentTimeMillis();
+
+ if (!rowsSeen.equals(batch)) {
+ HashSet<Text> copy1 = new HashSet<Text>(rowsSeen);
+ HashSet<Text> copy2 = new HashSet<Text>(batch);
+
+ copy1.removeAll(batch);
+ copy2.removeAll(rowsSeen);
+
+ System.out.printf("DIF %d %d %d%n", t1, copy1.size(), copy2.size());
+ System.err.printf("DIF %d %d %d%n", t1, copy1.size(), copy2.size());
+ System.err.println("Extra seen : " + copy1);
+ System.err.println("Not seen : " + copy2);
+ } else {
+ System.out.printf("BRQ %d %d %d %d %d%n", t1, (t2 - t1), rowsSeen.size(), count, (int) (rowsSeen.size() / ((t2 - t1) / 1000.0)));
+ }
+
+ }
+
+ private static void addRow(int batchSize, Value v) {
+ byte[] val = v.get();
+
+ int offset = ContinuousWalk.getPrevRowOffset(val);
+ if (offset > 1) {
+ Text prevRow = new Text();
+ prevRow.set(val, offset, 16);
+ if (rowsToQuery.size() < 3 * batchSize) {
+ rowsToQuery.add(prevRow);
+ }
+ }
+ }
+
+ private static HashSet<Text> rowsToQuery = new HashSet<Text>();
+
+ private static Set<Text> getBatch(Scanner scanner, long min, long max, int batchSize, Random r) {
+
+ while (rowsToQuery.size() < batchSize) {
+ byte[] scanStart = ContinuousIngest.genRow(min, max, r);
+ scanner.setRange(new Range(new Text(scanStart), null));
+
+ int count = 0;
+
+ long t1 = System.currentTimeMillis();
+
+ Iterator<Entry<Key,Value>> iter = scanner.iterator();
+ while (iter.hasNext() && rowsToQuery.size() < 3 * batchSize) {
+ Entry<Key,Value> entry = iter.next();
+ ContinuousWalk.validate(entry.getKey(), entry.getValue());
+ addRow(batchSize, entry.getValue());
+ count++;
+ }
+
+ long t2 = System.currentTimeMillis();
+
+ System.out.println("FSB " + t1 + " " + (t2 - t1) + " " + count);
+
+ UtilWaitThread.sleep(100);
+ }
+
+ HashSet<Text> ret = new HashSet<Text>();
+
+ Iterator<Text> iter = rowsToQuery.iterator();
+
+ for (int i = 0; i < batchSize; i++) {
+ ret.add(iter.next());
+ iter.remove();
+ }
+
+ return ret;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/bd283aec/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousIngest.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/continuous/ContinuousIngest.java
index 23cf15d,0000000..e3f0485
mode 100644,000000..100644
--- a/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousIngest.java
+++ b/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousIngest.java
@@@ -1,312 -1,0 +1,311 @@@
+/*
+ * 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.continuous;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+import java.util.UUID;
+import java.util.zip.CRC32;
+import java.util.zip.Checksum;
+
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.cli.BatchWriterOpts;
+import org.apache.accumulo.core.cli.ClientOnDefaultTable;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.MutationsRejectedException;
- import org.apache.accumulo.core.client.TableExistsException;
++import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.security.ColumnVisibility;
+import org.apache.accumulo.core.util.FastFormat;
+import org.apache.accumulo.trace.instrument.CountSampler;
+import org.apache.accumulo.trace.instrument.Trace;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.log4j.FileAppender;
+import org.apache.log4j.Level;
+import org.apache.log4j.Logger;
+import org.apache.log4j.PatternLayout;
+
+import com.beust.jcommander.IStringConverter;
+import com.beust.jcommander.Parameter;
+
+
+public class ContinuousIngest {
+
+ static public class BaseOpts extends ClientOnDefaultTable {
+ public class DebugConverter implements IStringConverter<String> {
+ @Override
+ public String convert(String debugLog) {
+ Logger logger = Logger.getLogger(Constants.CORE_PACKAGE_NAME);
+ logger.setLevel(Level.TRACE);
+ logger.setAdditivity(false);
+ try {
+ logger.addAppender(new FileAppender(new PatternLayout("%d{dd HH:mm:ss,SSS} [%-8c{2}] %-5p: %m%n"), debugLog, true));
+ } catch (IOException ex) {
+ throw new RuntimeException(ex);
+ }
+ return debugLog;
+ }
+ }
+
+ @Parameter(names="--min", description="lowest random row number to use")
+ long min = 0;
+
+ @Parameter(names="--max", description="maximum random row number to use")
+ long max = Long.MAX_VALUE;
+
+ @Parameter(names="--debugLog", description="file to write debugging output", converter=DebugConverter.class)
+ String debugLog = null;
+
+ BaseOpts() { super("ci"); }
+ }
+
+ public static class ShortConverter implements IStringConverter<Short> {
+ @Override
+ public Short convert(String value) {
+ return Short.valueOf(value);
+ }
+ }
+
+ static public class Opts extends BaseOpts {
+ @Parameter(names="--num", description="the number of entries to ingest")
+ long num = Long.MAX_VALUE;
+
+ @Parameter(names="--maxColF", description="maximum column family value to use", converter=ShortConverter.class)
+ short maxColF = Short.MAX_VALUE;
+
+ @Parameter(names="--maxColQ", description="maximum column qualifier value to use", converter=ShortConverter.class)
+ short maxColQ = Short.MAX_VALUE;
+
+ @Parameter(names="--addCheckSum", description="turn on checksums")
+ boolean checksum = false;
+
+ @Parameter(names="--visibilities", description="read the visibilities to ingest with from a file")
+ String visFile = null;
+ }
+
+ private static final byte[] EMPTY_BYTES = new byte[0];
+
+ private static List<ColumnVisibility> visibilities;
+
+ private static void initVisibilities(Opts opts) throws Exception {
+ if (opts.visFile == null) {
+ visibilities = Collections.singletonList(new ColumnVisibility());
+ return;
+ }
+
+ visibilities = new ArrayList<ColumnVisibility>();
+
+ FileSystem fs = FileSystem.get(new Configuration());
+ BufferedReader in = new BufferedReader(new InputStreamReader(fs.open(new Path(opts.visFile)), Constants.UTF8));
+
+ String line;
+
+ while ((line = in.readLine()) != null) {
+ visibilities.add(new ColumnVisibility(line));
+ }
+
+ in.close();
+ }
+
+ private static ColumnVisibility getVisibility(Random rand) {
+ return visibilities.get(rand.nextInt(visibilities.size()));
+ }
+
+ public static void main(String[] args) throws Exception {
+
+ Opts opts = new Opts();
+ BatchWriterOpts bwOpts = new BatchWriterOpts();
+ opts.parseArgs(ContinuousIngest.class.getName(), args, bwOpts);
+
+ initVisibilities(opts);
+
+ if (opts.min < 0 || opts.max < 0 || opts.max <= opts.min) {
+ throw new IllegalArgumentException("bad min and max");
+ }
+ Connector conn = opts.getConnector();
+
- if (!conn.tableOperations().exists(opts.getTableName()))
- try {
- conn.tableOperations().create(opts.getTableName());
- } catch (TableExistsException tee) {}
++ if (!conn.tableOperations().exists(opts.getTableName())) {
++ throw new TableNotFoundException(null, opts.getTableName(), "Consult the README and create the table before starting ingest.");
++ }
+
+ BatchWriter bw = conn.createBatchWriter(opts.getTableName(), bwOpts.getBatchWriterConfig());
+ bw = Trace.wrapAll(bw, new CountSampler(1024));
+
+ Random r = new Random();
+
+ byte[] ingestInstanceId = UUID.randomUUID().toString().getBytes(Constants.UTF8);
+
+ System.out.printf("UUID %d %s%n", System.currentTimeMillis(), new String(ingestInstanceId, Constants.UTF8));
+
+ long count = 0;
+ final int flushInterval = 1000000;
+ final int maxDepth = 25;
+
+ // always want to point back to flushed data. This way the previous item should
+ // always exist in accumulo when verifying data. To do this make insert N point
+ // back to the row from insert (N - flushInterval). The array below is used to keep
+ // track of this.
+ long prevRows[] = new long[flushInterval];
+ long firstRows[] = new long[flushInterval];
+ int firstColFams[] = new int[flushInterval];
+ int firstColQuals[] = new int[flushInterval];
+
+ long lastFlushTime = System.currentTimeMillis();
+
+ out: while (true) {
+ // generate first set of nodes
+ ColumnVisibility cv = getVisibility(r);
+
+ for (int index = 0; index < flushInterval; index++) {
+ long rowLong = genLong(opts.min, opts.max, r);
+ prevRows[index] = rowLong;
+ firstRows[index] = rowLong;
+
+ int cf = r.nextInt(opts.maxColF);
+ int cq = r.nextInt(opts.maxColQ);
+
+ firstColFams[index] = cf;
+ firstColQuals[index] = cq;
+
+ Mutation m = genMutation(rowLong, cf, cq, cv, ingestInstanceId, count, null, r, opts.checksum);
+ count++;
+ bw.addMutation(m);
+ }
+
+ lastFlushTime = flush(bw, count, flushInterval, lastFlushTime);
+ if (count >= opts.num)
+ break out;
+
+ // generate subsequent sets of nodes that link to previous set of nodes
+ for (int depth = 1; depth < maxDepth; depth++) {
+ for (int index = 0; index < flushInterval; index++) {
+ long rowLong = genLong(opts.min, opts.max, r);
+ byte[] prevRow = genRow(prevRows[index]);
+ prevRows[index] = rowLong;
+ Mutation m = genMutation(rowLong, r.nextInt(opts.maxColF), r.nextInt(opts.maxColQ), cv, ingestInstanceId, count, prevRow, r, opts.checksum);
+ count++;
+ bw.addMutation(m);
+ }
+
+ lastFlushTime = flush(bw, count, flushInterval, lastFlushTime);
+ if (count >= opts.num)
+ break out;
+ }
+
+ // create one big linked list, this makes all of the first inserts
+ // point to something
+ for (int index = 0; index < flushInterval - 1; index++) {
+ Mutation m = genMutation(firstRows[index], firstColFams[index], firstColQuals[index], cv, ingestInstanceId, count, genRow(prevRows[index + 1]), r,
+ opts.checksum);
+ count++;
+ bw.addMutation(m);
+ }
+ lastFlushTime = flush(bw, count, flushInterval, lastFlushTime);
+ if (count >= opts.num)
+ break out;
+ }
+
+ bw.close();
+ opts.stopTracing();
+ }
+
+ private static long flush(BatchWriter bw, long count, final int flushInterval, long lastFlushTime) throws MutationsRejectedException {
+ long t1 = System.currentTimeMillis();
+ bw.flush();
+ long t2 = System.currentTimeMillis();
+ System.out.printf("FLUSH %d %d %d %d %d%n", t2, (t2 - lastFlushTime), (t2 - t1), count, flushInterval);
+ lastFlushTime = t2;
+ return lastFlushTime;
+ }
+
+ public static Mutation genMutation(long rowLong, int cfInt, int cqInt, ColumnVisibility cv, byte[] ingestInstanceId, long count, byte[] prevRow, Random r,
+ boolean checksum) {
+ // Adler32 is supposed to be faster, but according to wikipedia is not good for small data.... so used CRC32 instead
+ CRC32 cksum = null;
+
+ byte[] rowString = genRow(rowLong);
+
+ byte[] cfString = FastFormat.toZeroPaddedString(cfInt, 4, 16, EMPTY_BYTES);
+ byte[] cqString = FastFormat.toZeroPaddedString(cqInt, 4, 16, EMPTY_BYTES);
+
+ if (checksum) {
+ cksum = new CRC32();
+ cksum.update(rowString);
+ cksum.update(cfString);
+ cksum.update(cqString);
+ cksum.update(cv.getExpression());
+ }
+
+ Mutation m = new Mutation(new Text(rowString));
+
+ m.put(new Text(cfString), new Text(cqString), cv, createValue(ingestInstanceId, count, prevRow, cksum));
+ return m;
+ }
+
+ public static final long genLong(long min, long max, Random r) {
+ return ((r.nextLong() & 0x7fffffffffffffffl) % (max - min)) + min;
+ }
+
+ static final byte[] genRow(long min, long max, Random r) {
+ return genRow(genLong(min, max, r));
+ }
+
+ static final byte[] genRow(long rowLong) {
+ return FastFormat.toZeroPaddedString(rowLong, 16, 16, EMPTY_BYTES);
+ }
+
+ private static Value createValue(byte[] ingestInstanceId, long count, byte[] prevRow, Checksum cksum) {
+ int dataLen = ingestInstanceId.length + 16 + (prevRow == null ? 0 : prevRow.length) + 3;
+ if (cksum != null)
+ dataLen += 8;
+ byte val[] = new byte[dataLen];
+ System.arraycopy(ingestInstanceId, 0, val, 0, ingestInstanceId.length);
+ int index = ingestInstanceId.length;
+ val[index++] = ':';
+ int added = FastFormat.toZeroPaddedString(val, index, count, 16, 16, EMPTY_BYTES);
+ if (added != 16)
+ throw new RuntimeException(" " + added);
+ index += 16;
+ val[index++] = ':';
+ if (prevRow != null) {
+ System.arraycopy(prevRow, 0, val, index, prevRow.length);
+ index += prevRow.length;
+ }
+
+ val[index++] = ':';
+
+ if (cksum != null) {
+ cksum.update(val, 0, index);
+ cksum.getValue();
+ FastFormat.toZeroPaddedString(val, index, cksum.getValue(), 8, 16, EMPTY_BYTES);
+ }
+
+ // System.out.println("val "+new String(val));
+
+ return new Value(val);
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/bd283aec/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousQuery.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/continuous/ContinuousQuery.java
index 117c136,0000000..4bbc85f
mode 100644,000000..100644
--- a/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousQuery.java
+++ b/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousQuery.java
@@@ -1,71 -1,0 +1,71 @@@
+/*
+ * 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.continuous;
+
+import java.util.Map.Entry;
+import java.util.Random;
+
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.cli.ScannerOpts;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Scanner;
+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.test.continuous.ContinuousIngest.BaseOpts;
+import org.apache.hadoop.io.Text;
+
+import com.beust.jcommander.Parameter;
+
+public class ContinuousQuery {
+
+ public static class Opts extends BaseOpts {
+ @Parameter(names="--sleep", description="the time to wait between queries", converter=TimeConverter.class)
+ long sleepTime = 100;
+ }
+
+ public static void main(String[] args) throws Exception {
+ Opts opts = new Opts();
+ ScannerOpts scanOpts = new ScannerOpts();
+ opts.parseArgs(ContinuousQuery.class.getName(), args, scanOpts);
+
+ Connector conn = opts.getConnector();
- Scanner scanner = conn.createScanner(opts.getTableName(), opts.auths);
++ Scanner scanner = ContinuousUtil.createScanner(conn, opts.getTableName(), opts.auths);
+ scanner.setBatchSize(scanOpts.scanBatchSize);
+
+ Random r = new Random();
+
+ while (true) {
+ byte[] row = ContinuousIngest.genRow(opts.min, opts.max, r);
+
+ int count = 0;
+
+ long t1 = System.currentTimeMillis();
+ scanner.setRange(new Range(new Text(row)));
+ for (Entry<Key,Value> entry : scanner) {
+ ContinuousWalk.validate(entry.getKey(), entry.getValue());
+ count++;
+ }
+ long t2 = System.currentTimeMillis();
+
+ System.out.printf("SRQ %d %s %d %d%n", t1, new String(row, Constants.UTF8), (t2 - t1), count);
+
+ if (opts.sleepTime > 0)
+ Thread.sleep(opts.sleepTime);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/bd283aec/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousScanner.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/continuous/ContinuousScanner.java
index c331bab,0000000..fcc8fec
mode 100644,000000..100644
--- a/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousScanner.java
+++ b/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousScanner.java
@@@ -1,104 -1,0 +1,104 @@@
+/*
+ * 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.continuous;
+
+import java.util.Iterator;
+import java.util.Map.Entry;
+import java.util.Random;
+
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.cli.ScannerOpts;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Scanner;
+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.security.Authorizations;
+import org.apache.accumulo.core.util.UtilWaitThread;
+import org.apache.hadoop.io.Text;
+
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.validators.PositiveInteger;
+
+public class ContinuousScanner {
+
+ static class Opts extends ContinuousWalk.Opts {
+ @Parameter(names="--numToScan", description="Number rows to scan between sleeps", required=true, validateWith=PositiveInteger.class)
+ long numToScan = 0;
+ }
+
+ public static void main(String[] args) throws Exception {
+ Opts opts = new Opts();
+ ScannerOpts scanOpts = new ScannerOpts();
+ opts.parseArgs(ContinuousScanner.class.getName(), args, scanOpts);
+
+ Random r = new Random();
+
+ long distance = 1000000000000l;
+
+ Connector conn = opts.getConnector();
+ Authorizations auths = opts.randomAuths.getAuths(r);
- Scanner scanner = conn.createScanner(opts.getTableName(), auths);
++ Scanner scanner = ContinuousUtil.createScanner(conn, opts.getTableName(), auths);
+ scanner.setBatchSize(scanOpts.scanBatchSize);
+
+ double delta = Math.min(.05, .05 / (opts.numToScan / 1000.0));
+
+ while (true) {
+ long startRow = ContinuousIngest.genLong(opts.min, opts.max - distance, r);
+ byte[] scanStart = ContinuousIngest.genRow(startRow);
+ byte[] scanStop = ContinuousIngest.genRow(startRow + distance);
+
+ scanner.setRange(new Range(new Text(scanStart), new Text(scanStop)));
+
+ int count = 0;
+ Iterator<Entry<Key,Value>> iter = scanner.iterator();
+
+ long t1 = System.currentTimeMillis();
+
+ while (iter.hasNext()) {
+ Entry<Key,Value> entry = iter.next();
+ ContinuousWalk.validate(entry.getKey(), entry.getValue());
+ count++;
+ }
+
+ long t2 = System.currentTimeMillis();
+
+ // System.out.println("P1 " +count +" "+((1-delta) * numToScan)+" "+((1+delta) * numToScan)+" "+numToScan);
+
+ if (count < (1 - delta) * opts.numToScan || count > (1 + delta) * opts.numToScan) {
+ if (count == 0) {
+ distance = distance * 10;
+ if (distance < 0)
+ distance = 1000000000000l;
+ } else {
+ double ratio = (double) opts.numToScan / count;
+ // move ratio closer to 1 to make change slower
+ ratio = ratio - (ratio - 1.0) * (2.0 / 3.0);
+ distance = (long) (ratio * distance);
+ }
+
+ // System.out.println("P2 "+delta +" "+numToScan+" "+distance+" "+((double)numToScan/count ));
+ }
+
+ System.out.printf("SCN %d %s %d %d%n", t1, new String(scanStart, Constants.UTF8), (t2 - t1), count);
+
+ if (opts.sleepTime > 0)
+ UtilWaitThread.sleep(opts.sleepTime);
+ }
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/bd283aec/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousUtil.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/continuous/ContinuousUtil.java
index 0000000,0000000..a8b2930
new file mode 100644
--- /dev/null
+++ b/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousUtil.java
@@@ -1,0 -1,0 +1,49 @@@
++/*
++ * 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.continuous;
++
++import org.apache.accumulo.core.client.Connector;
++import org.apache.accumulo.core.client.Scanner;
++import org.apache.accumulo.core.client.TableNotFoundException;
++import org.apache.accumulo.core.security.Authorizations;
++
++/**
++ * Useful utility methods common to the Continuous test suite.
++ */
++final class ContinuousUtil {
++ private ContinuousUtil() {}
++
++ /**
++ * Attempt to create a table scanner, or fail if the table does not exist.
++ *
++ * @param connector
++ * A populated connector object
++ * @param table
++ * The table name to scan over
++ * @param auths
++ * The authorizations to use for the scanner
++ * @return a scanner for the requested table
++ * @throws TableNotFoundException
++ * If the table does not exist
++ */
++ static Scanner createScanner(Connector connector, String table, Authorizations auths) throws TableNotFoundException {
++ if (!connector.tableOperations().exists(table)) {
++ throw new TableNotFoundException(null, table, "Consult the README and create the table before starting test processes.");
++ }
++ return connector.createScanner(table, auths);
++ }
++}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/bd283aec/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousWalk.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/continuous/ContinuousWalk.java
index 4032dfa,0000000..34a5e9b
mode 100644,000000..100644
--- a/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousWalk.java
+++ b/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousWalk.java
@@@ -1,237 -1,0 +1,237 @@@
+/*
+ * 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.continuous;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Random;
+import java.util.zip.CRC32;
+
- import org.apache.accumulo.trace.instrument.Span;
- import org.apache.accumulo.trace.instrument.Trace;
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Scanner;
+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.security.Authorizations;
++import org.apache.accumulo.trace.instrument.Span;
++import org.apache.accumulo.trace.instrument.Trace;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+
+import com.beust.jcommander.IStringConverter;
+import com.beust.jcommander.Parameter;
+
+
+public class ContinuousWalk {
+
+ static public class Opts extends ContinuousQuery.Opts {
+ class RandomAuthsConverter implements IStringConverter<RandomAuths> {
+ @Override
+ public RandomAuths convert(String value) {
+ try {
+ return new RandomAuths(value);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+ @Parameter(names="--authsFile", description="read the authorities to use from a file")
+ RandomAuths randomAuths = new RandomAuths();
+ }
+
+ static class BadChecksumException extends RuntimeException {
+ private static final long serialVersionUID = 1L;
+
+ public BadChecksumException(String msg) {
+ super(msg);
+ }
+
+ }
+
+ static class RandomAuths {
+ private List<Authorizations> auths;
+
+ RandomAuths() {
+ auths = Collections.singletonList(Constants.NO_AUTHS);
+ }
+
+ RandomAuths(String file) throws IOException {
+ if (file == null) {
+ auths = Collections.singletonList(Constants.NO_AUTHS);
+ return;
+ }
+
+ auths = new ArrayList<Authorizations>();
+
+ FileSystem fs = FileSystem.get(new Configuration());
+ BufferedReader in = new BufferedReader(new InputStreamReader(fs.open(new Path(file)), Constants.UTF8));
+ try {
+ String line;
+ while ((line = in.readLine()) != null) {
+ auths.add(new Authorizations(line.split(",")));
+ }
+ } finally {
+ in.close();
+ }
+ }
+
+ Authorizations getAuths(Random r) {
+ return auths.get(r.nextInt(auths.size()));
+ }
+ }
+
+ public static void main(String[] args) throws Exception {
+ Opts opts = new Opts();
+ opts.parseArgs(ContinuousWalk.class.getName(), args);
+
+ Connector conn = opts.getConnector();
+
+ Random r = new Random();
+
+ ArrayList<Value> values = new ArrayList<Value>();
+
+ while (true) {
- Scanner scanner = conn.createScanner(opts.getTableName(), opts.randomAuths.getAuths(r));
++ Scanner scanner = ContinuousUtil.createScanner(conn, opts.getTableName(), opts.randomAuths.getAuths(r));
+ String row = findAStartRow(opts.min, opts.max, scanner, r);
+
+ while (row != null) {
+
+ values.clear();
+
+ long t1 = System.currentTimeMillis();
+ Span span = Trace.on("walk");
+ try {
+ scanner.setRange(new Range(new Text(row)));
+ for (Entry<Key,Value> entry : scanner) {
+ validate(entry.getKey(), entry.getValue());
+ values.add(entry.getValue());
+ }
+ } finally {
+ span.stop();
+ }
+ long t2 = System.currentTimeMillis();
+
+ System.out.printf("SRQ %d %s %d %d%n", t1, row, (t2 - t1), values.size());
+
+ if (values.size() > 0) {
+ row = getPrevRow(values.get(r.nextInt(values.size())));
+ } else {
+ System.out.printf("MIS %d %s%n", t1, row);
+ System.err.printf("MIS %d %s%n", t1, row);
+ row = null;
+ }
+
+ if (opts.sleepTime > 0)
+ Thread.sleep(opts.sleepTime);
+ }
+
+ if (opts.sleepTime > 0)
+ Thread.sleep(opts.sleepTime);
+ }
+ }
+
+ private static String findAStartRow(long min, long max, Scanner scanner, Random r) {
+
+ byte[] scanStart = ContinuousIngest.genRow(min, max, r);
+ scanner.setRange(new Range(new Text(scanStart), null));
+ scanner.setBatchSize(100);
+
+ int count = 0;
+ String pr = null;
+
+ long t1 = System.currentTimeMillis();
+
+ for (Entry<Key,Value> entry : scanner) {
+ validate(entry.getKey(), entry.getValue());
+ pr = getPrevRow(entry.getValue());
+ count++;
+ if (pr != null)
+ break;
+ }
+
+ long t2 = System.currentTimeMillis();
+
+ System.out.printf("FSR %d %s %d %d%n", t1, new String(scanStart, Constants.UTF8), (t2 - t1), count);
+
+ return pr;
+ }
+
+ static int getPrevRowOffset(byte val[]) {
+ if (val.length == 0)
+ throw new IllegalArgumentException();
+ if (val[53] != ':')
+ throw new IllegalArgumentException(new String(val, Constants.UTF8));
+
+ // prev row starts at 54
+ if (val[54] != ':') {
+ if (val[54 + 16] != ':')
+ throw new IllegalArgumentException(new String(val, Constants.UTF8));
+ return 54;
+ }
+
+ return -1;
+ }
+
+ static String getPrevRow(Value value) {
+
+ byte[] val = value.get();
+ int offset = getPrevRowOffset(val);
+ if (offset > 0) {
+ return new String(val, offset, 16, Constants.UTF8);
+ }
+
+ return null;
+ }
+
+ static int getChecksumOffset(byte val[]) {
+ if (val[val.length - 1] != ':') {
+ if (val[val.length - 9] != ':')
+ throw new IllegalArgumentException(new String(val, Constants.UTF8));
+ return val.length - 8;
+ }
+
+ return -1;
+ }
+
+ static void validate(Key key, Value value) throws BadChecksumException {
+ int ckOff = getChecksumOffset(value.get());
+ if (ckOff < 0)
+ return;
+
+ long storedCksum = Long.parseLong(new String(value.get(), ckOff, 8, Constants.UTF8), 16);
+
+ CRC32 cksum = new CRC32();
+
+ cksum.update(key.getRowData().toArray());
+ cksum.update(key.getColumnFamilyData().toArray());
+ cksum.update(key.getColumnQualifierData().toArray());
+ cksum.update(key.getColumnVisibilityData().toArray());
+ cksum.update(value.get(), 0, ckOff);
+
+ if (cksum.getValue() != storedCksum) {
+ throw new BadChecksumException("Checksum invalid " + key + " " + value);
+ }
+ }
+}
[10/16] git commit: ACCUMULO-2399 Alert user CI table should exist
Posted by md...@apache.org.
ACCUMULO-2399 Alert user CI table should exist
Extract common table checking functionality for continuous clients and
fail fast when the table does not exist. No longer create the table in
ingest to make the requirement explicit.
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/759582b7
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/759582b7
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/759582b7
Branch: refs/heads/1.4.5-SNAPSHOT
Commit: 759582b78d5d72870a4a8c359ef6134c4dd97993
Parents: a8bbb91
Author: Mike Drob <md...@cloudera.com>
Authored: Tue Feb 25 15:40:12 2014 -0500
Committer: Mike Drob <md...@cloudera.com>
Committed: Tue Mar 4 08:36:13 2014 -0500
----------------------------------------------------------------------
.../test/continuous/ContinuousBatchWalker.java | 2 +-
.../test/continuous/ContinuousIngest.java | 9 ++--
.../server/test/continuous/ContinuousQuery.java | 2 +-
.../test/continuous/ContinuousScanner.java | 2 +-
.../server/test/continuous/ContinuousUtil.java | 49 ++++++++++++++++++++
.../server/test/continuous/ContinuousWalk.java | 2 +-
6 files changed, 57 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/759582b7/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousBatchWalker.java
----------------------------------------------------------------------
diff --git a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousBatchWalker.java b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousBatchWalker.java
index 4659aff..fe738b6 100644
--- a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousBatchWalker.java
+++ b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousBatchWalker.java
@@ -98,7 +98,7 @@ public class ContinuousBatchWalker {
Authorizations auths = randomAuths.getAuths(r);
Connector conn = new ZooKeeperInstance(instanceName, zooKeepers).getConnector(user, password.getBytes());
- Scanner scanner = conn.createScanner(table, auths);
+ Scanner scanner = ContinuousUtil.createScanner(conn, table, auths);
BatchScanner bs = conn.createBatchScanner(table, auths, numQueryThreads);
while (true) {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/759582b7/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousIngest.java
----------------------------------------------------------------------
diff --git a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousIngest.java b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousIngest.java
index b63efbc..1699ab4 100644
--- a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousIngest.java
+++ b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousIngest.java
@@ -36,7 +36,7 @@ import org.apache.accumulo.core.client.BatchWriter;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.MutationsRejectedException;
-import org.apache.accumulo.core.client.TableExistsException;
+import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.client.ZooKeeperInstance;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Value;
@@ -150,10 +150,9 @@ public class ContinuousIngest {
String path = ZooUtil.getRoot(instance) + Constants.ZTRACERS;
Tracer.getInstance().addReceiver(new ZooSpanClient(zooKeepers, path, localhost, "cingest", 1000));
- if (!conn.tableOperations().exists(table))
- try {
- conn.tableOperations().create(table);
- } catch (TableExistsException tee) {}
+ if (!conn.tableOperations().exists(table)) {
+ throw new TableNotFoundException(null, table, "Consult the README and create the table before starting ingest.");
+ }
BatchWriter bw = conn.createBatchWriter(table, maxMemory, maxLatency, maxWriteThreads);
bw = Trace.wrapAll(bw, new CountSampler(1024));
http://git-wip-us.apache.org/repos/asf/accumulo/blob/759582b7/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousQuery.java
----------------------------------------------------------------------
diff --git a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousQuery.java b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousQuery.java
index c8ae6ec..accd867 100644
--- a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousQuery.java
+++ b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousQuery.java
@@ -50,7 +50,7 @@ public class ContinuousQuery {
long sleepTime = Long.parseLong(args[7]);
Connector conn = new ZooKeeperInstance(instanceName, zooKeepers).getConnector(user, password.getBytes());
- Scanner scanner = conn.createScanner(table, new Authorizations());
+ Scanner scanner = ContinuousUtil.createScanner(conn, table, new Authorizations());
Random r = new Random();
http://git-wip-us.apache.org/repos/asf/accumulo/blob/759582b7/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousScanner.java
----------------------------------------------------------------------
diff --git a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousScanner.java b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousScanner.java
index 0ac3df6..13a91db 100644
--- a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousScanner.java
+++ b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousScanner.java
@@ -97,7 +97,7 @@ public class ContinuousScanner {
Instance instance = new ZooKeeperInstance(instanceName, zooKeepers);
Connector conn = instance.getConnector(user, password.getBytes());
Authorizations auths = randomAuths.getAuths(r);
- Scanner scanner = conn.createScanner(table, auths);
+ Scanner scanner = ContinuousUtil.createScanner(conn, table, auths);
double delta = Math.min(.05, .05 / (numToScan / 1000.0));
// System.out.println("Delta "+delta);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/759582b7/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousUtil.java
----------------------------------------------------------------------
diff --git a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousUtil.java b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousUtil.java
new file mode 100644
index 0000000..a788b59
--- /dev/null
+++ b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousUtil.java
@@ -0,0 +1,49 @@
+/*
+ * 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.server.test.continuous;
+
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.security.Authorizations;
+
+/**
+ * Useful utility methods common to the Continuous test suite.
+ */
+final class ContinuousUtil {
+ private ContinuousUtil() {}
+
+ /**
+ * Attempt to create a table scanner, or fail if the table does not exist.
+ *
+ * @param connector
+ * A populated connector object
+ * @param table
+ * The table name to scan over
+ * @param auths
+ * The authorizations to use for the scanner
+ * @return a scanner for the requested table
+ * @throws TableNotFoundException
+ * If the table does not exist
+ */
+ static Scanner createScanner(Connector connector, String table, Authorizations auths) throws TableNotFoundException {
+ if (!connector.tableOperations().exists(table)) {
+ throw new TableNotFoundException(null, table, "Consult the README and create the table before starting test processes.");
+ }
+ return connector.createScanner(table, auths);
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/759582b7/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousWalk.java
----------------------------------------------------------------------
diff --git a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousWalk.java b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousWalk.java
index 7d1e7f9..15b6090 100644
--- a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousWalk.java
+++ b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousWalk.java
@@ -154,7 +154,7 @@ public class ContinuousWalk {
ArrayList<Value> values = new ArrayList<Value>();
while (true) {
- Scanner scanner = conn.createScanner(table, randomAuths.getAuths(r));
+ Scanner scanner = ContinuousUtil.createScanner(conn, table, randomAuths.getAuths(r));
String row = findAStartRow(min, max, scanner, r);
while (row != null) {
[06/16] git commit: Merge branch 1.5.2-SNAPSHOT into 1.6.0-SNAPSHOT
Posted by md...@apache.org.
Merge branch 1.5.2-SNAPSHOT into 1.6.0-SNAPSHOT
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/e0571816
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/e0571816
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/e0571816
Branch: refs/heads/master
Commit: e0571816af3bae8d3423d6020046d617cb6e6782
Parents: 18a5500 bee78fa
Author: Mike Drob <md...@cloudera.com>
Authored: Tue Mar 4 08:33:04 2014 -0500
Committer: Mike Drob <md...@cloudera.com>
Committed: Tue Mar 4 08:33:04 2014 -0500
----------------------------------------------------------------------
----------------------------------------------------------------------
[13/16] git commit: Merge branch '1.4.5-SNAPSHOT' into 1.5.2-SNAPSHOT
Posted by md...@apache.org.
Merge branch '1.4.5-SNAPSHOT' into 1.5.2-SNAPSHOT
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/bd283aec
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/bd283aec
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/bd283aec
Branch: refs/heads/master
Commit: bd283aec0aa10d4c9deba8b5df097d48a424780b
Parents: bee78fa 759582b
Author: Mike Drob <md...@cloudera.com>
Authored: Tue Mar 4 09:15:16 2014 -0500
Committer: Mike Drob <md...@cloudera.com>
Committed: Tue Mar 4 09:15:16 2014 -0500
----------------------------------------------------------------------
.../test/continuous/ContinuousBatchWalker.java | 2 +-
.../test/continuous/ContinuousIngest.java | 9 ++--
.../test/continuous/ContinuousQuery.java | 2 +-
.../test/continuous/ContinuousScanner.java | 2 +-
.../test/continuous/ContinuousUtil.java | 49 ++++++++++++++++++++
.../test/continuous/ContinuousWalk.java | 6 +--
6 files changed, 59 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/bd283aec/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousBatchWalker.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/continuous/ContinuousBatchWalker.java
index d021164,0000000..3304d24
mode 100644,000000..100644
--- a/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousBatchWalker.java
+++ b/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousBatchWalker.java
@@@ -1,182 -1,0 +1,182 @@@
+/*
+ * 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.continuous;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.accumulo.core.cli.BatchScannerOpts;
+import org.apache.accumulo.core.cli.ScannerOpts;
+import org.apache.accumulo.core.client.BatchScanner;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Scanner;
+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.security.Authorizations;
+import org.apache.accumulo.core.util.UtilWaitThread;
+import org.apache.hadoop.io.Text;
+
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.validators.PositiveInteger;
+
+public class ContinuousBatchWalker {
+
+ static class Opts extends ContinuousWalk.Opts {
+ @Parameter(names="--numToScan", description="Number rows to scan between sleeps", required=true, validateWith=PositiveInteger.class)
+ long numToScan = 0;
+ }
+
+ public static void main(String[] args) throws Exception {
+
+ Opts opts = new Opts();
+ ScannerOpts scanOpts = new ScannerOpts();
+ BatchScannerOpts bsOpts = new BatchScannerOpts();
+ opts.parseArgs(ContinuousBatchWalker.class.getName(), args, scanOpts, bsOpts);
+
+ Random r = new Random();
+ Authorizations auths = opts.randomAuths.getAuths(r);
+
+ Connector conn = opts.getConnector();
- Scanner scanner = conn.createScanner(opts.getTableName(), auths);
++ Scanner scanner = ContinuousUtil.createScanner(conn, opts.getTableName(), auths);
+ scanner.setBatchSize(scanOpts.scanBatchSize);
+
+ BatchScanner bs = conn.createBatchScanner(opts.getTableName(), auths, bsOpts.scanThreads);
+ bs.setTimeout(bsOpts.scanTimeout, TimeUnit.MILLISECONDS);
+
+ while (true) {
+ Set<Text> batch = getBatch(scanner, opts.min, opts.max, scanOpts.scanBatchSize, r);
+ List<Range> ranges = new ArrayList<Range>(batch.size());
+
+ for (Text row : batch) {
+ ranges.add(new Range(row));
+ }
+
+ runBatchScan(scanOpts.scanBatchSize, bs, batch, ranges);
+
+ UtilWaitThread.sleep(opts.sleepTime);
+ }
+
+ }
+
+ /*
+ * private static void runSequentialScan(Scanner scanner, List<Range> ranges) { Set<Text> srowsSeen = new HashSet<Text>(); long st1 =
+ * System.currentTimeMillis(); int scount = 0; for (Range range : ranges) { scanner.setRange(range);
+ *
+ * for (Entry<Key,Value> entry : scanner) { srowsSeen.add(entry.getKey().getRow()); scount++; } }
+ *
+ *
+ * long st2 = System.currentTimeMillis(); System.out.println("SRQ "+(st2 - st1)+" "+srowsSeen.size() +" "+scount); }
+ */
+
+ private static void runBatchScan(int batchSize, BatchScanner bs, Set<Text> batch, List<Range> ranges) {
+ bs.setRanges(ranges);
+
+ Set<Text> rowsSeen = new HashSet<Text>();
+
+ int count = 0;
+
+ long t1 = System.currentTimeMillis();
+
+ for (Entry<Key,Value> entry : bs) {
+ ContinuousWalk.validate(entry.getKey(), entry.getValue());
+
+ rowsSeen.add(entry.getKey().getRow());
+
+ addRow(batchSize, entry.getValue());
+
+ count++;
+ }
+
+ long t2 = System.currentTimeMillis();
+
+ if (!rowsSeen.equals(batch)) {
+ HashSet<Text> copy1 = new HashSet<Text>(rowsSeen);
+ HashSet<Text> copy2 = new HashSet<Text>(batch);
+
+ copy1.removeAll(batch);
+ copy2.removeAll(rowsSeen);
+
+ System.out.printf("DIF %d %d %d%n", t1, copy1.size(), copy2.size());
+ System.err.printf("DIF %d %d %d%n", t1, copy1.size(), copy2.size());
+ System.err.println("Extra seen : " + copy1);
+ System.err.println("Not seen : " + copy2);
+ } else {
+ System.out.printf("BRQ %d %d %d %d %d%n", t1, (t2 - t1), rowsSeen.size(), count, (int) (rowsSeen.size() / ((t2 - t1) / 1000.0)));
+ }
+
+ }
+
+ private static void addRow(int batchSize, Value v) {
+ byte[] val = v.get();
+
+ int offset = ContinuousWalk.getPrevRowOffset(val);
+ if (offset > 1) {
+ Text prevRow = new Text();
+ prevRow.set(val, offset, 16);
+ if (rowsToQuery.size() < 3 * batchSize) {
+ rowsToQuery.add(prevRow);
+ }
+ }
+ }
+
+ private static HashSet<Text> rowsToQuery = new HashSet<Text>();
+
+ private static Set<Text> getBatch(Scanner scanner, long min, long max, int batchSize, Random r) {
+
+ while (rowsToQuery.size() < batchSize) {
+ byte[] scanStart = ContinuousIngest.genRow(min, max, r);
+ scanner.setRange(new Range(new Text(scanStart), null));
+
+ int count = 0;
+
+ long t1 = System.currentTimeMillis();
+
+ Iterator<Entry<Key,Value>> iter = scanner.iterator();
+ while (iter.hasNext() && rowsToQuery.size() < 3 * batchSize) {
+ Entry<Key,Value> entry = iter.next();
+ ContinuousWalk.validate(entry.getKey(), entry.getValue());
+ addRow(batchSize, entry.getValue());
+ count++;
+ }
+
+ long t2 = System.currentTimeMillis();
+
+ System.out.println("FSB " + t1 + " " + (t2 - t1) + " " + count);
+
+ UtilWaitThread.sleep(100);
+ }
+
+ HashSet<Text> ret = new HashSet<Text>();
+
+ Iterator<Text> iter = rowsToQuery.iterator();
+
+ for (int i = 0; i < batchSize; i++) {
+ ret.add(iter.next());
+ iter.remove();
+ }
+
+ return ret;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/bd283aec/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousIngest.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/continuous/ContinuousIngest.java
index 23cf15d,0000000..e3f0485
mode 100644,000000..100644
--- a/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousIngest.java
+++ b/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousIngest.java
@@@ -1,312 -1,0 +1,311 @@@
+/*
+ * 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.continuous;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+import java.util.UUID;
+import java.util.zip.CRC32;
+import java.util.zip.Checksum;
+
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.cli.BatchWriterOpts;
+import org.apache.accumulo.core.cli.ClientOnDefaultTable;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.MutationsRejectedException;
- import org.apache.accumulo.core.client.TableExistsException;
++import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.security.ColumnVisibility;
+import org.apache.accumulo.core.util.FastFormat;
+import org.apache.accumulo.trace.instrument.CountSampler;
+import org.apache.accumulo.trace.instrument.Trace;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.log4j.FileAppender;
+import org.apache.log4j.Level;
+import org.apache.log4j.Logger;
+import org.apache.log4j.PatternLayout;
+
+import com.beust.jcommander.IStringConverter;
+import com.beust.jcommander.Parameter;
+
+
+public class ContinuousIngest {
+
+ static public class BaseOpts extends ClientOnDefaultTable {
+ public class DebugConverter implements IStringConverter<String> {
+ @Override
+ public String convert(String debugLog) {
+ Logger logger = Logger.getLogger(Constants.CORE_PACKAGE_NAME);
+ logger.setLevel(Level.TRACE);
+ logger.setAdditivity(false);
+ try {
+ logger.addAppender(new FileAppender(new PatternLayout("%d{dd HH:mm:ss,SSS} [%-8c{2}] %-5p: %m%n"), debugLog, true));
+ } catch (IOException ex) {
+ throw new RuntimeException(ex);
+ }
+ return debugLog;
+ }
+ }
+
+ @Parameter(names="--min", description="lowest random row number to use")
+ long min = 0;
+
+ @Parameter(names="--max", description="maximum random row number to use")
+ long max = Long.MAX_VALUE;
+
+ @Parameter(names="--debugLog", description="file to write debugging output", converter=DebugConverter.class)
+ String debugLog = null;
+
+ BaseOpts() { super("ci"); }
+ }
+
+ public static class ShortConverter implements IStringConverter<Short> {
+ @Override
+ public Short convert(String value) {
+ return Short.valueOf(value);
+ }
+ }
+
+ static public class Opts extends BaseOpts {
+ @Parameter(names="--num", description="the number of entries to ingest")
+ long num = Long.MAX_VALUE;
+
+ @Parameter(names="--maxColF", description="maximum column family value to use", converter=ShortConverter.class)
+ short maxColF = Short.MAX_VALUE;
+
+ @Parameter(names="--maxColQ", description="maximum column qualifier value to use", converter=ShortConverter.class)
+ short maxColQ = Short.MAX_VALUE;
+
+ @Parameter(names="--addCheckSum", description="turn on checksums")
+ boolean checksum = false;
+
+ @Parameter(names="--visibilities", description="read the visibilities to ingest with from a file")
+ String visFile = null;
+ }
+
+ private static final byte[] EMPTY_BYTES = new byte[0];
+
+ private static List<ColumnVisibility> visibilities;
+
+ private static void initVisibilities(Opts opts) throws Exception {
+ if (opts.visFile == null) {
+ visibilities = Collections.singletonList(new ColumnVisibility());
+ return;
+ }
+
+ visibilities = new ArrayList<ColumnVisibility>();
+
+ FileSystem fs = FileSystem.get(new Configuration());
+ BufferedReader in = new BufferedReader(new InputStreamReader(fs.open(new Path(opts.visFile)), Constants.UTF8));
+
+ String line;
+
+ while ((line = in.readLine()) != null) {
+ visibilities.add(new ColumnVisibility(line));
+ }
+
+ in.close();
+ }
+
+ private static ColumnVisibility getVisibility(Random rand) {
+ return visibilities.get(rand.nextInt(visibilities.size()));
+ }
+
+ public static void main(String[] args) throws Exception {
+
+ Opts opts = new Opts();
+ BatchWriterOpts bwOpts = new BatchWriterOpts();
+ opts.parseArgs(ContinuousIngest.class.getName(), args, bwOpts);
+
+ initVisibilities(opts);
+
+ if (opts.min < 0 || opts.max < 0 || opts.max <= opts.min) {
+ throw new IllegalArgumentException("bad min and max");
+ }
+ Connector conn = opts.getConnector();
+
- if (!conn.tableOperations().exists(opts.getTableName()))
- try {
- conn.tableOperations().create(opts.getTableName());
- } catch (TableExistsException tee) {}
++ if (!conn.tableOperations().exists(opts.getTableName())) {
++ throw new TableNotFoundException(null, opts.getTableName(), "Consult the README and create the table before starting ingest.");
++ }
+
+ BatchWriter bw = conn.createBatchWriter(opts.getTableName(), bwOpts.getBatchWriterConfig());
+ bw = Trace.wrapAll(bw, new CountSampler(1024));
+
+ Random r = new Random();
+
+ byte[] ingestInstanceId = UUID.randomUUID().toString().getBytes(Constants.UTF8);
+
+ System.out.printf("UUID %d %s%n", System.currentTimeMillis(), new String(ingestInstanceId, Constants.UTF8));
+
+ long count = 0;
+ final int flushInterval = 1000000;
+ final int maxDepth = 25;
+
+ // always want to point back to flushed data. This way the previous item should
+ // always exist in accumulo when verifying data. To do this make insert N point
+ // back to the row from insert (N - flushInterval). The array below is used to keep
+ // track of this.
+ long prevRows[] = new long[flushInterval];
+ long firstRows[] = new long[flushInterval];
+ int firstColFams[] = new int[flushInterval];
+ int firstColQuals[] = new int[flushInterval];
+
+ long lastFlushTime = System.currentTimeMillis();
+
+ out: while (true) {
+ // generate first set of nodes
+ ColumnVisibility cv = getVisibility(r);
+
+ for (int index = 0; index < flushInterval; index++) {
+ long rowLong = genLong(opts.min, opts.max, r);
+ prevRows[index] = rowLong;
+ firstRows[index] = rowLong;
+
+ int cf = r.nextInt(opts.maxColF);
+ int cq = r.nextInt(opts.maxColQ);
+
+ firstColFams[index] = cf;
+ firstColQuals[index] = cq;
+
+ Mutation m = genMutation(rowLong, cf, cq, cv, ingestInstanceId, count, null, r, opts.checksum);
+ count++;
+ bw.addMutation(m);
+ }
+
+ lastFlushTime = flush(bw, count, flushInterval, lastFlushTime);
+ if (count >= opts.num)
+ break out;
+
+ // generate subsequent sets of nodes that link to previous set of nodes
+ for (int depth = 1; depth < maxDepth; depth++) {
+ for (int index = 0; index < flushInterval; index++) {
+ long rowLong = genLong(opts.min, opts.max, r);
+ byte[] prevRow = genRow(prevRows[index]);
+ prevRows[index] = rowLong;
+ Mutation m = genMutation(rowLong, r.nextInt(opts.maxColF), r.nextInt(opts.maxColQ), cv, ingestInstanceId, count, prevRow, r, opts.checksum);
+ count++;
+ bw.addMutation(m);
+ }
+
+ lastFlushTime = flush(bw, count, flushInterval, lastFlushTime);
+ if (count >= opts.num)
+ break out;
+ }
+
+ // create one big linked list, this makes all of the first inserts
+ // point to something
+ for (int index = 0; index < flushInterval - 1; index++) {
+ Mutation m = genMutation(firstRows[index], firstColFams[index], firstColQuals[index], cv, ingestInstanceId, count, genRow(prevRows[index + 1]), r,
+ opts.checksum);
+ count++;
+ bw.addMutation(m);
+ }
+ lastFlushTime = flush(bw, count, flushInterval, lastFlushTime);
+ if (count >= opts.num)
+ break out;
+ }
+
+ bw.close();
+ opts.stopTracing();
+ }
+
+ private static long flush(BatchWriter bw, long count, final int flushInterval, long lastFlushTime) throws MutationsRejectedException {
+ long t1 = System.currentTimeMillis();
+ bw.flush();
+ long t2 = System.currentTimeMillis();
+ System.out.printf("FLUSH %d %d %d %d %d%n", t2, (t2 - lastFlushTime), (t2 - t1), count, flushInterval);
+ lastFlushTime = t2;
+ return lastFlushTime;
+ }
+
+ public static Mutation genMutation(long rowLong, int cfInt, int cqInt, ColumnVisibility cv, byte[] ingestInstanceId, long count, byte[] prevRow, Random r,
+ boolean checksum) {
+ // Adler32 is supposed to be faster, but according to wikipedia is not good for small data.... so used CRC32 instead
+ CRC32 cksum = null;
+
+ byte[] rowString = genRow(rowLong);
+
+ byte[] cfString = FastFormat.toZeroPaddedString(cfInt, 4, 16, EMPTY_BYTES);
+ byte[] cqString = FastFormat.toZeroPaddedString(cqInt, 4, 16, EMPTY_BYTES);
+
+ if (checksum) {
+ cksum = new CRC32();
+ cksum.update(rowString);
+ cksum.update(cfString);
+ cksum.update(cqString);
+ cksum.update(cv.getExpression());
+ }
+
+ Mutation m = new Mutation(new Text(rowString));
+
+ m.put(new Text(cfString), new Text(cqString), cv, createValue(ingestInstanceId, count, prevRow, cksum));
+ return m;
+ }
+
+ public static final long genLong(long min, long max, Random r) {
+ return ((r.nextLong() & 0x7fffffffffffffffl) % (max - min)) + min;
+ }
+
+ static final byte[] genRow(long min, long max, Random r) {
+ return genRow(genLong(min, max, r));
+ }
+
+ static final byte[] genRow(long rowLong) {
+ return FastFormat.toZeroPaddedString(rowLong, 16, 16, EMPTY_BYTES);
+ }
+
+ private static Value createValue(byte[] ingestInstanceId, long count, byte[] prevRow, Checksum cksum) {
+ int dataLen = ingestInstanceId.length + 16 + (prevRow == null ? 0 : prevRow.length) + 3;
+ if (cksum != null)
+ dataLen += 8;
+ byte val[] = new byte[dataLen];
+ System.arraycopy(ingestInstanceId, 0, val, 0, ingestInstanceId.length);
+ int index = ingestInstanceId.length;
+ val[index++] = ':';
+ int added = FastFormat.toZeroPaddedString(val, index, count, 16, 16, EMPTY_BYTES);
+ if (added != 16)
+ throw new RuntimeException(" " + added);
+ index += 16;
+ val[index++] = ':';
+ if (prevRow != null) {
+ System.arraycopy(prevRow, 0, val, index, prevRow.length);
+ index += prevRow.length;
+ }
+
+ val[index++] = ':';
+
+ if (cksum != null) {
+ cksum.update(val, 0, index);
+ cksum.getValue();
+ FastFormat.toZeroPaddedString(val, index, cksum.getValue(), 8, 16, EMPTY_BYTES);
+ }
+
+ // System.out.println("val "+new String(val));
+
+ return new Value(val);
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/bd283aec/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousQuery.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/continuous/ContinuousQuery.java
index 117c136,0000000..4bbc85f
mode 100644,000000..100644
--- a/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousQuery.java
+++ b/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousQuery.java
@@@ -1,71 -1,0 +1,71 @@@
+/*
+ * 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.continuous;
+
+import java.util.Map.Entry;
+import java.util.Random;
+
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.cli.ScannerOpts;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Scanner;
+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.test.continuous.ContinuousIngest.BaseOpts;
+import org.apache.hadoop.io.Text;
+
+import com.beust.jcommander.Parameter;
+
+public class ContinuousQuery {
+
+ public static class Opts extends BaseOpts {
+ @Parameter(names="--sleep", description="the time to wait between queries", converter=TimeConverter.class)
+ long sleepTime = 100;
+ }
+
+ public static void main(String[] args) throws Exception {
+ Opts opts = new Opts();
+ ScannerOpts scanOpts = new ScannerOpts();
+ opts.parseArgs(ContinuousQuery.class.getName(), args, scanOpts);
+
+ Connector conn = opts.getConnector();
- Scanner scanner = conn.createScanner(opts.getTableName(), opts.auths);
++ Scanner scanner = ContinuousUtil.createScanner(conn, opts.getTableName(), opts.auths);
+ scanner.setBatchSize(scanOpts.scanBatchSize);
+
+ Random r = new Random();
+
+ while (true) {
+ byte[] row = ContinuousIngest.genRow(opts.min, opts.max, r);
+
+ int count = 0;
+
+ long t1 = System.currentTimeMillis();
+ scanner.setRange(new Range(new Text(row)));
+ for (Entry<Key,Value> entry : scanner) {
+ ContinuousWalk.validate(entry.getKey(), entry.getValue());
+ count++;
+ }
+ long t2 = System.currentTimeMillis();
+
+ System.out.printf("SRQ %d %s %d %d%n", t1, new String(row, Constants.UTF8), (t2 - t1), count);
+
+ if (opts.sleepTime > 0)
+ Thread.sleep(opts.sleepTime);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/bd283aec/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousScanner.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/continuous/ContinuousScanner.java
index c331bab,0000000..fcc8fec
mode 100644,000000..100644
--- a/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousScanner.java
+++ b/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousScanner.java
@@@ -1,104 -1,0 +1,104 @@@
+/*
+ * 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.continuous;
+
+import java.util.Iterator;
+import java.util.Map.Entry;
+import java.util.Random;
+
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.cli.ScannerOpts;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Scanner;
+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.security.Authorizations;
+import org.apache.accumulo.core.util.UtilWaitThread;
+import org.apache.hadoop.io.Text;
+
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.validators.PositiveInteger;
+
+public class ContinuousScanner {
+
+ static class Opts extends ContinuousWalk.Opts {
+ @Parameter(names="--numToScan", description="Number rows to scan between sleeps", required=true, validateWith=PositiveInteger.class)
+ long numToScan = 0;
+ }
+
+ public static void main(String[] args) throws Exception {
+ Opts opts = new Opts();
+ ScannerOpts scanOpts = new ScannerOpts();
+ opts.parseArgs(ContinuousScanner.class.getName(), args, scanOpts);
+
+ Random r = new Random();
+
+ long distance = 1000000000000l;
+
+ Connector conn = opts.getConnector();
+ Authorizations auths = opts.randomAuths.getAuths(r);
- Scanner scanner = conn.createScanner(opts.getTableName(), auths);
++ Scanner scanner = ContinuousUtil.createScanner(conn, opts.getTableName(), auths);
+ scanner.setBatchSize(scanOpts.scanBatchSize);
+
+ double delta = Math.min(.05, .05 / (opts.numToScan / 1000.0));
+
+ while (true) {
+ long startRow = ContinuousIngest.genLong(opts.min, opts.max - distance, r);
+ byte[] scanStart = ContinuousIngest.genRow(startRow);
+ byte[] scanStop = ContinuousIngest.genRow(startRow + distance);
+
+ scanner.setRange(new Range(new Text(scanStart), new Text(scanStop)));
+
+ int count = 0;
+ Iterator<Entry<Key,Value>> iter = scanner.iterator();
+
+ long t1 = System.currentTimeMillis();
+
+ while (iter.hasNext()) {
+ Entry<Key,Value> entry = iter.next();
+ ContinuousWalk.validate(entry.getKey(), entry.getValue());
+ count++;
+ }
+
+ long t2 = System.currentTimeMillis();
+
+ // System.out.println("P1 " +count +" "+((1-delta) * numToScan)+" "+((1+delta) * numToScan)+" "+numToScan);
+
+ if (count < (1 - delta) * opts.numToScan || count > (1 + delta) * opts.numToScan) {
+ if (count == 0) {
+ distance = distance * 10;
+ if (distance < 0)
+ distance = 1000000000000l;
+ } else {
+ double ratio = (double) opts.numToScan / count;
+ // move ratio closer to 1 to make change slower
+ ratio = ratio - (ratio - 1.0) * (2.0 / 3.0);
+ distance = (long) (ratio * distance);
+ }
+
+ // System.out.println("P2 "+delta +" "+numToScan+" "+distance+" "+((double)numToScan/count ));
+ }
+
+ System.out.printf("SCN %d %s %d %d%n", t1, new String(scanStart, Constants.UTF8), (t2 - t1), count);
+
+ if (opts.sleepTime > 0)
+ UtilWaitThread.sleep(opts.sleepTime);
+ }
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/bd283aec/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousUtil.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/continuous/ContinuousUtil.java
index 0000000,0000000..a8b2930
new file mode 100644
--- /dev/null
+++ b/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousUtil.java
@@@ -1,0 -1,0 +1,49 @@@
++/*
++ * 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.continuous;
++
++import org.apache.accumulo.core.client.Connector;
++import org.apache.accumulo.core.client.Scanner;
++import org.apache.accumulo.core.client.TableNotFoundException;
++import org.apache.accumulo.core.security.Authorizations;
++
++/**
++ * Useful utility methods common to the Continuous test suite.
++ */
++final class ContinuousUtil {
++ private ContinuousUtil() {}
++
++ /**
++ * Attempt to create a table scanner, or fail if the table does not exist.
++ *
++ * @param connector
++ * A populated connector object
++ * @param table
++ * The table name to scan over
++ * @param auths
++ * The authorizations to use for the scanner
++ * @return a scanner for the requested table
++ * @throws TableNotFoundException
++ * If the table does not exist
++ */
++ static Scanner createScanner(Connector connector, String table, Authorizations auths) throws TableNotFoundException {
++ if (!connector.tableOperations().exists(table)) {
++ throw new TableNotFoundException(null, table, "Consult the README and create the table before starting test processes.");
++ }
++ return connector.createScanner(table, auths);
++ }
++}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/bd283aec/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousWalk.java
----------------------------------------------------------------------
diff --cc test/src/main/java/org/apache/accumulo/test/continuous/ContinuousWalk.java
index 4032dfa,0000000..34a5e9b
mode 100644,000000..100644
--- a/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousWalk.java
+++ b/test/src/main/java/org/apache/accumulo/test/continuous/ContinuousWalk.java
@@@ -1,237 -1,0 +1,237 @@@
+/*
+ * 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.continuous;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Random;
+import java.util.zip.CRC32;
+
- import org.apache.accumulo.trace.instrument.Span;
- import org.apache.accumulo.trace.instrument.Trace;
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Scanner;
+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.security.Authorizations;
++import org.apache.accumulo.trace.instrument.Span;
++import org.apache.accumulo.trace.instrument.Trace;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+
+import com.beust.jcommander.IStringConverter;
+import com.beust.jcommander.Parameter;
+
+
+public class ContinuousWalk {
+
+ static public class Opts extends ContinuousQuery.Opts {
+ class RandomAuthsConverter implements IStringConverter<RandomAuths> {
+ @Override
+ public RandomAuths convert(String value) {
+ try {
+ return new RandomAuths(value);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+ @Parameter(names="--authsFile", description="read the authorities to use from a file")
+ RandomAuths randomAuths = new RandomAuths();
+ }
+
+ static class BadChecksumException extends RuntimeException {
+ private static final long serialVersionUID = 1L;
+
+ public BadChecksumException(String msg) {
+ super(msg);
+ }
+
+ }
+
+ static class RandomAuths {
+ private List<Authorizations> auths;
+
+ RandomAuths() {
+ auths = Collections.singletonList(Constants.NO_AUTHS);
+ }
+
+ RandomAuths(String file) throws IOException {
+ if (file == null) {
+ auths = Collections.singletonList(Constants.NO_AUTHS);
+ return;
+ }
+
+ auths = new ArrayList<Authorizations>();
+
+ FileSystem fs = FileSystem.get(new Configuration());
+ BufferedReader in = new BufferedReader(new InputStreamReader(fs.open(new Path(file)), Constants.UTF8));
+ try {
+ String line;
+ while ((line = in.readLine()) != null) {
+ auths.add(new Authorizations(line.split(",")));
+ }
+ } finally {
+ in.close();
+ }
+ }
+
+ Authorizations getAuths(Random r) {
+ return auths.get(r.nextInt(auths.size()));
+ }
+ }
+
+ public static void main(String[] args) throws Exception {
+ Opts opts = new Opts();
+ opts.parseArgs(ContinuousWalk.class.getName(), args);
+
+ Connector conn = opts.getConnector();
+
+ Random r = new Random();
+
+ ArrayList<Value> values = new ArrayList<Value>();
+
+ while (true) {
- Scanner scanner = conn.createScanner(opts.getTableName(), opts.randomAuths.getAuths(r));
++ Scanner scanner = ContinuousUtil.createScanner(conn, opts.getTableName(), opts.randomAuths.getAuths(r));
+ String row = findAStartRow(opts.min, opts.max, scanner, r);
+
+ while (row != null) {
+
+ values.clear();
+
+ long t1 = System.currentTimeMillis();
+ Span span = Trace.on("walk");
+ try {
+ scanner.setRange(new Range(new Text(row)));
+ for (Entry<Key,Value> entry : scanner) {
+ validate(entry.getKey(), entry.getValue());
+ values.add(entry.getValue());
+ }
+ } finally {
+ span.stop();
+ }
+ long t2 = System.currentTimeMillis();
+
+ System.out.printf("SRQ %d %s %d %d%n", t1, row, (t2 - t1), values.size());
+
+ if (values.size() > 0) {
+ row = getPrevRow(values.get(r.nextInt(values.size())));
+ } else {
+ System.out.printf("MIS %d %s%n", t1, row);
+ System.err.printf("MIS %d %s%n", t1, row);
+ row = null;
+ }
+
+ if (opts.sleepTime > 0)
+ Thread.sleep(opts.sleepTime);
+ }
+
+ if (opts.sleepTime > 0)
+ Thread.sleep(opts.sleepTime);
+ }
+ }
+
+ private static String findAStartRow(long min, long max, Scanner scanner, Random r) {
+
+ byte[] scanStart = ContinuousIngest.genRow(min, max, r);
+ scanner.setRange(new Range(new Text(scanStart), null));
+ scanner.setBatchSize(100);
+
+ int count = 0;
+ String pr = null;
+
+ long t1 = System.currentTimeMillis();
+
+ for (Entry<Key,Value> entry : scanner) {
+ validate(entry.getKey(), entry.getValue());
+ pr = getPrevRow(entry.getValue());
+ count++;
+ if (pr != null)
+ break;
+ }
+
+ long t2 = System.currentTimeMillis();
+
+ System.out.printf("FSR %d %s %d %d%n", t1, new String(scanStart, Constants.UTF8), (t2 - t1), count);
+
+ return pr;
+ }
+
+ static int getPrevRowOffset(byte val[]) {
+ if (val.length == 0)
+ throw new IllegalArgumentException();
+ if (val[53] != ':')
+ throw new IllegalArgumentException(new String(val, Constants.UTF8));
+
+ // prev row starts at 54
+ if (val[54] != ':') {
+ if (val[54 + 16] != ':')
+ throw new IllegalArgumentException(new String(val, Constants.UTF8));
+ return 54;
+ }
+
+ return -1;
+ }
+
+ static String getPrevRow(Value value) {
+
+ byte[] val = value.get();
+ int offset = getPrevRowOffset(val);
+ if (offset > 0) {
+ return new String(val, offset, 16, Constants.UTF8);
+ }
+
+ return null;
+ }
+
+ static int getChecksumOffset(byte val[]) {
+ if (val[val.length - 1] != ':') {
+ if (val[val.length - 9] != ':')
+ throw new IllegalArgumentException(new String(val, Constants.UTF8));
+ return val.length - 8;
+ }
+
+ return -1;
+ }
+
+ static void validate(Key key, Value value) throws BadChecksumException {
+ int ckOff = getChecksumOffset(value.get());
+ if (ckOff < 0)
+ return;
+
+ long storedCksum = Long.parseLong(new String(value.get(), ckOff, 8, Constants.UTF8), 16);
+
+ CRC32 cksum = new CRC32();
+
+ cksum.update(key.getRowData().toArray());
+ cksum.update(key.getColumnFamilyData().toArray());
+ cksum.update(key.getColumnQualifierData().toArray());
+ cksum.update(key.getColumnVisibilityData().toArray());
+ cksum.update(value.get(), 0, ckOff);
+
+ if (cksum.getValue() != storedCksum) {
+ throw new BadChecksumException("Checksum invalid " + key + " " + value);
+ }
+ }
+}
[09/16] git commit: ACCUMULO-2399 Alert user CI table should exist
Posted by md...@apache.org.
ACCUMULO-2399 Alert user CI table should exist
Extract common table checking functionality for continuous clients and
fail fast when the table does not exist. No longer create the table in
ingest to make the requirement explicit.
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/759582b7
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/759582b7
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/759582b7
Branch: refs/heads/master
Commit: 759582b78d5d72870a4a8c359ef6134c4dd97993
Parents: a8bbb91
Author: Mike Drob <md...@cloudera.com>
Authored: Tue Feb 25 15:40:12 2014 -0500
Committer: Mike Drob <md...@cloudera.com>
Committed: Tue Mar 4 08:36:13 2014 -0500
----------------------------------------------------------------------
.../test/continuous/ContinuousBatchWalker.java | 2 +-
.../test/continuous/ContinuousIngest.java | 9 ++--
.../server/test/continuous/ContinuousQuery.java | 2 +-
.../test/continuous/ContinuousScanner.java | 2 +-
.../server/test/continuous/ContinuousUtil.java | 49 ++++++++++++++++++++
.../server/test/continuous/ContinuousWalk.java | 2 +-
6 files changed, 57 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/759582b7/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousBatchWalker.java
----------------------------------------------------------------------
diff --git a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousBatchWalker.java b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousBatchWalker.java
index 4659aff..fe738b6 100644
--- a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousBatchWalker.java
+++ b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousBatchWalker.java
@@ -98,7 +98,7 @@ public class ContinuousBatchWalker {
Authorizations auths = randomAuths.getAuths(r);
Connector conn = new ZooKeeperInstance(instanceName, zooKeepers).getConnector(user, password.getBytes());
- Scanner scanner = conn.createScanner(table, auths);
+ Scanner scanner = ContinuousUtil.createScanner(conn, table, auths);
BatchScanner bs = conn.createBatchScanner(table, auths, numQueryThreads);
while (true) {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/759582b7/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousIngest.java
----------------------------------------------------------------------
diff --git a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousIngest.java b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousIngest.java
index b63efbc..1699ab4 100644
--- a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousIngest.java
+++ b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousIngest.java
@@ -36,7 +36,7 @@ import org.apache.accumulo.core.client.BatchWriter;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.MutationsRejectedException;
-import org.apache.accumulo.core.client.TableExistsException;
+import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.client.ZooKeeperInstance;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Value;
@@ -150,10 +150,9 @@ public class ContinuousIngest {
String path = ZooUtil.getRoot(instance) + Constants.ZTRACERS;
Tracer.getInstance().addReceiver(new ZooSpanClient(zooKeepers, path, localhost, "cingest", 1000));
- if (!conn.tableOperations().exists(table))
- try {
- conn.tableOperations().create(table);
- } catch (TableExistsException tee) {}
+ if (!conn.tableOperations().exists(table)) {
+ throw new TableNotFoundException(null, table, "Consult the README and create the table before starting ingest.");
+ }
BatchWriter bw = conn.createBatchWriter(table, maxMemory, maxLatency, maxWriteThreads);
bw = Trace.wrapAll(bw, new CountSampler(1024));
http://git-wip-us.apache.org/repos/asf/accumulo/blob/759582b7/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousQuery.java
----------------------------------------------------------------------
diff --git a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousQuery.java b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousQuery.java
index c8ae6ec..accd867 100644
--- a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousQuery.java
+++ b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousQuery.java
@@ -50,7 +50,7 @@ public class ContinuousQuery {
long sleepTime = Long.parseLong(args[7]);
Connector conn = new ZooKeeperInstance(instanceName, zooKeepers).getConnector(user, password.getBytes());
- Scanner scanner = conn.createScanner(table, new Authorizations());
+ Scanner scanner = ContinuousUtil.createScanner(conn, table, new Authorizations());
Random r = new Random();
http://git-wip-us.apache.org/repos/asf/accumulo/blob/759582b7/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousScanner.java
----------------------------------------------------------------------
diff --git a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousScanner.java b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousScanner.java
index 0ac3df6..13a91db 100644
--- a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousScanner.java
+++ b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousScanner.java
@@ -97,7 +97,7 @@ public class ContinuousScanner {
Instance instance = new ZooKeeperInstance(instanceName, zooKeepers);
Connector conn = instance.getConnector(user, password.getBytes());
Authorizations auths = randomAuths.getAuths(r);
- Scanner scanner = conn.createScanner(table, auths);
+ Scanner scanner = ContinuousUtil.createScanner(conn, table, auths);
double delta = Math.min(.05, .05 / (numToScan / 1000.0));
// System.out.println("Delta "+delta);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/759582b7/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousUtil.java
----------------------------------------------------------------------
diff --git a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousUtil.java b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousUtil.java
new file mode 100644
index 0000000..a788b59
--- /dev/null
+++ b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousUtil.java
@@ -0,0 +1,49 @@
+/*
+ * 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.server.test.continuous;
+
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.security.Authorizations;
+
+/**
+ * Useful utility methods common to the Continuous test suite.
+ */
+final class ContinuousUtil {
+ private ContinuousUtil() {}
+
+ /**
+ * Attempt to create a table scanner, or fail if the table does not exist.
+ *
+ * @param connector
+ * A populated connector object
+ * @param table
+ * The table name to scan over
+ * @param auths
+ * The authorizations to use for the scanner
+ * @return a scanner for the requested table
+ * @throws TableNotFoundException
+ * If the table does not exist
+ */
+ static Scanner createScanner(Connector connector, String table, Authorizations auths) throws TableNotFoundException {
+ if (!connector.tableOperations().exists(table)) {
+ throw new TableNotFoundException(null, table, "Consult the README and create the table before starting test processes.");
+ }
+ return connector.createScanner(table, auths);
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/759582b7/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousWalk.java
----------------------------------------------------------------------
diff --git a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousWalk.java b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousWalk.java
index 7d1e7f9..15b6090 100644
--- a/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousWalk.java
+++ b/src/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousWalk.java
@@ -154,7 +154,7 @@ public class ContinuousWalk {
ArrayList<Value> values = new ArrayList<Value>();
while (true) {
- Scanner scanner = conn.createScanner(table, randomAuths.getAuths(r));
+ Scanner scanner = ContinuousUtil.createScanner(conn, table, randomAuths.getAuths(r));
String row = findAStartRow(min, max, scanner, r);
while (row != null) {