You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by ct...@apache.org on 2014/11/01 05:56:56 UTC
[02/50] [abbrv] Merge branch '1.5' into 1.6
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/test/java/org/apache/accumulo/test/VolumeIT.java
----------------------------------------------------------------------
diff --cc test/src/test/java/org/apache/accumulo/test/VolumeIT.java
index a338d5f,0000000..d5c940d
mode 100644,000000..100644
--- a/test/src/test/java/org/apache/accumulo/test/VolumeIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/VolumeIT.java
@@@ -1,553 -1,0 +1,553 @@@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.accumulo.test;
+
++import static com.google.common.base.Charsets.UTF_8;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.io.BufferedOutputStream;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
- import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.AccumuloException;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.MutationsRejectedException;
+import org.apache.accumulo.core.client.Scanner;
+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.client.admin.DiskUsage;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.KeyExtent;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.metadata.MetadataTable;
+import org.apache.accumulo.core.metadata.RootTable;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.DataFileColumnFamily;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.security.TablePermission;
+import org.apache.accumulo.core.util.CachedConfiguration;
+import org.apache.accumulo.core.zookeeper.ZooUtil;
+import org.apache.accumulo.minicluster.impl.MiniAccumuloConfigImpl;
+import org.apache.accumulo.server.ServerConstants;
+import org.apache.accumulo.server.init.Initialize;
+import org.apache.accumulo.server.util.Admin;
+import org.apache.accumulo.test.functional.ConfigurableMacIT;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RawLocalFileSystem;
+import org.apache.hadoop.io.Text;
+import org.apache.zookeeper.ZooKeeper;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class VolumeIT extends ConfigurableMacIT {
+
+ private static final Text EMPTY = new Text();
+ private static final Value EMPTY_VALUE = new Value(new byte[] {});
+ private File volDirBase;
+ private Path v1, v2;
+
+ @Override
+ protected int defaultTimeoutSeconds() {
+ return 5 * 60;
+ }
+
+ @SuppressWarnings("deprecation")
+ @Override
+ public void configure(MiniAccumuloConfigImpl cfg, Configuration hadoopCoreSite) {
+ File baseDir = cfg.getDir();
+ volDirBase = new File(baseDir, "volumes");
+ File v1f = new File(volDirBase, "v1");
+ File v2f = new File(volDirBase, "v2");
+ v1f.mkdir();
+ v2f.mkdir();
+ v1 = new Path("file://" + v1f.getAbsolutePath());
+ v2 = new Path("file://" + v2f.getAbsolutePath());
+
+ // Run MAC on two locations in the local file system
+ URI v1Uri = v1.toUri();
+ cfg.setProperty(Property.INSTANCE_DFS_DIR, v1Uri.getPath());
+ cfg.setProperty(Property.INSTANCE_DFS_URI, v1Uri.getScheme() + v1Uri.getHost());
+ cfg.setProperty(Property.INSTANCE_VOLUMES, v1.toString() + "," + v2.toString());
+
+ // use raw local file system so walogs sync and flush will work
+ hadoopCoreSite.set("fs.file.impl", RawLocalFileSystem.class.getName());
+
+ super.configure(cfg, hadoopCoreSite);
+ }
+
+ @Test
+ public void test() throws Exception {
+ // create a table
+ Connector connector = getConnector();
+ String tableName = getUniqueNames(1)[0];
+ connector.tableOperations().create(tableName);
+ SortedSet<Text> partitions = new TreeSet<Text>();
+ // with some splits
+ for (String s : "d,m,t".split(","))
+ partitions.add(new Text(s));
+ connector.tableOperations().addSplits(tableName, partitions);
+ // scribble over the splits
+ BatchWriter bw = connector.createBatchWriter(tableName, new BatchWriterConfig());
+ String[] rows = "a,b,c,d,e,f,g,h,i,j,k,l,m,n,o,p,q,r,s,t,u,v,w,x,y,z".split(",");
+ for (String s : rows) {
+ Mutation m = new Mutation(new Text(s));
+ m.put(EMPTY, EMPTY, EMPTY_VALUE);
+ bw.addMutation(m);
+ }
+ bw.close();
+ // write the data to disk, read it back
+ connector.tableOperations().flush(tableName, null, null, true);
+ Scanner scanner = connector.createScanner(tableName, Authorizations.EMPTY);
+ int i = 0;
+ for (Entry<Key,Value> entry : scanner) {
+ assertEquals(rows[i++], entry.getKey().getRow().toString());
+ }
+ // verify the new files are written to the different volumes
+ scanner = connector.createScanner(MetadataTable.NAME, Authorizations.EMPTY);
+ scanner.setRange(new Range("1", "1<"));
+ scanner.fetchColumnFamily(DataFileColumnFamily.NAME);
+ int fileCount = 0;
+
+ for (Entry<Key,Value> entry : scanner) {
+ boolean inV1 = entry.getKey().getColumnQualifier().toString().contains(v1.toString());
+ boolean inV2 = entry.getKey().getColumnQualifier().toString().contains(v2.toString());
+ assertTrue(inV1 || inV2);
+ fileCount++;
+ }
+ assertEquals(4, fileCount);
+ List<DiskUsage> diskUsage = connector.tableOperations().getDiskUsage(Collections.singleton(tableName));
+ assertEquals(1, diskUsage.size());
+ long usage = diskUsage.get(0).getUsage().longValue();
+ System.out.println("usage " + usage);
+ assertTrue(usage > 700 && usage < 800);
+ }
+
+ private void verifyData(List<String> expected, Scanner createScanner) {
+
+ List<String> actual = new ArrayList<String>();
+
+ for (Entry<Key,Value> entry : createScanner) {
+ Key k = entry.getKey();
+ actual.add(k.getRow() + ":" + k.getColumnFamily() + ":" + k.getColumnQualifier() + ":" + entry.getValue());
+ }
+
+ Collections.sort(expected);
+ Collections.sort(actual);
+
+ Assert.assertEquals(expected, actual);
+ }
+
+ @Test
+ public void testRelativePaths() throws Exception {
+
+ List<String> expected = new ArrayList<String>();
+
+ Connector connector = getConnector();
+ String tableName = getUniqueNames(1)[0];
+ connector.tableOperations().create(tableName, false);
+
+ String tableId = connector.tableOperations().tableIdMap().get(tableName);
+
+ SortedSet<Text> partitions = new TreeSet<Text>();
+ // with some splits
+ for (String s : "c,g,k,p,s,v".split(","))
+ partitions.add(new Text(s));
+
+ connector.tableOperations().addSplits(tableName, partitions);
+
+ BatchWriter bw = connector.createBatchWriter(tableName, new BatchWriterConfig());
+
+ // create two files in each tablet
+
+ String[] rows = "a,b,c,d,e,f,g,h,i,j,k,l,m,n,o,p,q,r,s,t,u,v,w,x,y,z".split(",");
+ for (String s : rows) {
+ Mutation m = new Mutation(s);
+ m.put("cf1", "cq1", "1");
+ bw.addMutation(m);
+ expected.add(s + ":cf1:cq1:1");
+ }
+
+ bw.flush();
+ connector.tableOperations().flush(tableName, null, null, true);
+
+ for (String s : rows) {
+ Mutation m = new Mutation(s);
+ m.put("cf1", "cq1", "2");
+ bw.addMutation(m);
+ expected.add(s + ":cf1:cq1:2");
+ }
+
+ bw.close();
+ connector.tableOperations().flush(tableName, null, null, true);
+
+ verifyData(expected, connector.createScanner(tableName, Authorizations.EMPTY));
+
+ connector.tableOperations().offline(tableName, true);
+
+ connector.securityOperations().grantTablePermission("root", MetadataTable.NAME, TablePermission.WRITE);
+
+ Scanner metaScanner = connector.createScanner(MetadataTable.NAME, Authorizations.EMPTY);
+ metaScanner.fetchColumnFamily(MetadataSchema.TabletsSection.DataFileColumnFamily.NAME);
+ metaScanner.setRange(new KeyExtent(new Text(tableId), null, null).toMetadataRange());
+
+ BatchWriter mbw = connector.createBatchWriter(MetadataTable.NAME, new BatchWriterConfig());
+
+ for (Entry<Key,Value> entry : metaScanner) {
+ String cq = entry.getKey().getColumnQualifier().toString();
+ if (cq.startsWith(v1.toString())) {
+ Path path = new Path(cq);
+ String relPath = "/" + path.getParent().getName() + "/" + path.getName();
+ Mutation fileMut = new Mutation(entry.getKey().getRow());
+ fileMut.putDelete(entry.getKey().getColumnFamily(), entry.getKey().getColumnQualifier());
+ fileMut.put(entry.getKey().getColumnFamily().toString(), relPath, entry.getValue().toString());
+ mbw.addMutation(fileMut);
+ }
+ }
+
+ mbw.close();
+
+ connector.tableOperations().online(tableName, true);
+
+ verifyData(expected, connector.createScanner(tableName, Authorizations.EMPTY));
+
+ connector.tableOperations().compact(tableName, null, null, true, true);
+
+ verifyData(expected, connector.createScanner(tableName, Authorizations.EMPTY));
+
+ for (Entry<Key,Value> entry : metaScanner) {
+ String cq = entry.getKey().getColumnQualifier().toString();
+ Path path = new Path(cq);
+ Assert.assertTrue("relative path not deleted " + path.toString(), path.depth() > 2);
+ }
+
+ }
+
+ @Test
+ public void testAddVolumes() throws Exception {
+
+ String[] tableNames = getUniqueNames(2);
+
+ // grab this before shutting down cluster
+ String uuid = new ZooKeeperInstance(cluster.getInstanceName(), cluster.getZooKeepers()).getInstanceID();
+
+ verifyVolumesUsed(tableNames[0], false, v1, v2);
+
+ Assert.assertEquals(0, cluster.exec(Admin.class, "stopAll").waitFor());
+ cluster.stop();
+
+ Configuration conf = new Configuration(false);
+ conf.addResource(new Path(cluster.getConfig().getConfDir().toURI().toString(), "accumulo-site.xml"));
+
+ File v3f = new File(volDirBase, "v3");
+ v3f.mkdir();
+ Path v3 = new Path("file://" + v3f.getAbsolutePath());
+
+ conf.set(Property.INSTANCE_VOLUMES.getKey(), v1.toString() + "," + v2.toString() + "," + v3.toString());
+ BufferedOutputStream fos = new BufferedOutputStream(new FileOutputStream(new File(cluster.getConfig().getConfDir(), "accumulo-site.xml")));
+ conf.writeXml(fos);
+ fos.close();
+
+ // initialize volume
+ Assert.assertEquals(0, cluster.exec(Initialize.class, "--add-volumes").waitFor());
+
+ // check that all volumes are initialized
+ for (Path volumePath : Arrays.asList(v1, v2, v3)) {
+ FileSystem fs = volumePath.getFileSystem(CachedConfiguration.getInstance());
+ Path vp = new Path(volumePath, ServerConstants.INSTANCE_ID_DIR);
+ FileStatus[] iids = fs.listStatus(vp);
+ Assert.assertEquals(1, iids.length);
+ Assert.assertEquals(uuid, iids[0].getPath().getName());
+ }
+
+ // start cluster and verify that new volume is used
+ cluster.start();
+
+ verifyVolumesUsed(tableNames[1], false, v1, v2, v3);
+ }
+
+ @Test
+ public void testNonConfiguredVolumes() throws Exception {
+
+ String[] tableNames = getUniqueNames(2);
+
+ // grab this before shutting down cluster
+ String uuid = new ZooKeeperInstance(cluster.getInstanceName(), cluster.getZooKeepers()).getInstanceID();
+
+ verifyVolumesUsed(tableNames[0], false, v1, v2);
+
+ Assert.assertEquals(0, cluster.exec(Admin.class, "stopAll").waitFor());
+ cluster.stop();
+
+ Configuration conf = new Configuration(false);
+ conf.addResource(new Path(cluster.getConfig().getConfDir().toURI().toString(), "accumulo-site.xml"));
+
+ File v3f = new File(volDirBase, "v3");
+ v3f.mkdir();
+ Path v3 = new Path("file://" + v3f.getAbsolutePath());
+
+ conf.set(Property.INSTANCE_VOLUMES.getKey(), v2.toString() + "," + v3.toString());
+ BufferedOutputStream fos = new BufferedOutputStream(new FileOutputStream(new File(cluster.getConfig().getConfDir(), "accumulo-site.xml")));
+ conf.writeXml(fos);
+ fos.close();
+
+ // initialize volume
+ Assert.assertEquals(0, cluster.exec(Initialize.class, "--add-volumes").waitFor());
+
+ // check that all volumes are initialized
+ for (Path volumePath : Arrays.asList(v1, v2, v3)) {
+ FileSystem fs = volumePath.getFileSystem(CachedConfiguration.getInstance());
+ Path vp = new Path(volumePath, ServerConstants.INSTANCE_ID_DIR);
+ FileStatus[] iids = fs.listStatus(vp);
+ Assert.assertEquals(1, iids.length);
+ Assert.assertEquals(uuid, iids[0].getPath().getName());
+ }
+
+ // start cluster and verify that new volume is used
+ cluster.start();
+
+ // Make sure we can still read the tables (tableNames[0] is very likely to have a file still on v1)
+ List<String> expected = new ArrayList<String>();
+ for (int i = 0; i < 100; i++) {
+ String row = String.format("%06d", i * 100 + 3);
+ expected.add(row + ":cf1:cq1:1");
+ }
+
+ verifyData(expected, getConnector().createScanner(tableNames[0], Authorizations.EMPTY));
+
+ // v1 should not have any data for tableNames[1]
+ verifyVolumesUsed(tableNames[1], false, v2, v3);
+ }
+
+ private void writeData(String tableName, Connector conn) throws AccumuloException, AccumuloSecurityException, TableExistsException, TableNotFoundException,
+ MutationsRejectedException {
+ TreeSet<Text> splits = new TreeSet<Text>();
+ for (int i = 1; i < 100; i++) {
+ splits.add(new Text(String.format("%06d", i * 100)));
+ }
+
+ conn.tableOperations().create(tableName);
+ conn.tableOperations().addSplits(tableName, splits);
+
+ BatchWriter bw = conn.createBatchWriter(tableName, new BatchWriterConfig());
+ for (int i = 0; i < 100; i++) {
+ String row = String.format("%06d", i * 100 + 3);
+ Mutation m = new Mutation(row);
+ m.put("cf1", "cq1", "1");
+ bw.addMutation(m);
+ }
+
+ bw.close();
+ }
+
+ private void verifyVolumesUsed(String tableName, boolean shouldExist, Path... paths) throws AccumuloException, AccumuloSecurityException,
+ TableExistsException, TableNotFoundException, MutationsRejectedException {
+
+ Connector conn = getConnector();
+
+ List<String> expected = new ArrayList<String>();
+ for (int i = 0; i < 100; i++) {
+ String row = String.format("%06d", i * 100 + 3);
+ expected.add(row + ":cf1:cq1:1");
+ }
+
+ if (!conn.tableOperations().exists(tableName)) {
+ Assert.assertFalse(shouldExist);
+
+ writeData(tableName, conn);
+
+ verifyData(expected, conn.createScanner(tableName, Authorizations.EMPTY));
+
+ conn.tableOperations().flush(tableName, null, null, true);
+ }
+
+ verifyData(expected, conn.createScanner(tableName, Authorizations.EMPTY));
+
+ String tableId = conn.tableOperations().tableIdMap().get(tableName);
+ Scanner metaScanner = conn.createScanner(MetadataTable.NAME, Authorizations.EMPTY);
+ MetadataSchema.TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN.fetch(metaScanner);
+ metaScanner.fetchColumnFamily(MetadataSchema.TabletsSection.DataFileColumnFamily.NAME);
+ metaScanner.setRange(new KeyExtent(new Text(tableId), null, null).toMetadataRange());
+
+ int counts[] = new int[paths.length];
+
+ outer: for (Entry<Key,Value> entry : metaScanner) {
+ String cf = entry.getKey().getColumnFamily().toString();
+ String cq = entry.getKey().getColumnQualifier().toString();
+
+ String path;
+ if (cf.equals(MetadataSchema.TabletsSection.DataFileColumnFamily.NAME.toString()))
+ path = cq;
+ else
+ path = entry.getValue().toString();
+
+ for (int i = 0; i < paths.length; i++) {
+ if (path.startsWith(paths[i].toString())) {
+ counts[i]++;
+ continue outer;
+ }
+ }
+
+ Assert.fail("Unexpected volume " + path);
+ }
+
+ // if a volume is chosen randomly for each tablet, then the probability that a volume will not be chosen for any tablet is ((num_volumes -
+ // 1)/num_volumes)^num_tablets. For 100 tablets and 3 volumes the probability that only 2 volumes would be chosen is 2.46e-18
+
+ int sum = 0;
+ for (int count : counts) {
+ Assert.assertTrue(count > 0);
+ sum += count;
+ }
+
+ Assert.assertEquals(200, sum);
+ }
+
+ @Test
+ public void testRemoveVolumes() throws Exception {
+ String[] tableNames = getUniqueNames(2);
+
+ verifyVolumesUsed(tableNames[0], false, v1, v2);
+
+ Assert.assertEquals(0, cluster.exec(Admin.class, "stopAll").waitFor());
+ cluster.stop();
+
+ Configuration conf = new Configuration(false);
+ conf.addResource(new Path(cluster.getConfig().getConfDir().toURI().toString(), "accumulo-site.xml"));
+
+ conf.set(Property.INSTANCE_VOLUMES.getKey(), v2.toString());
+ BufferedOutputStream fos = new BufferedOutputStream(new FileOutputStream(new File(cluster.getConfig().getConfDir(), "accumulo-site.xml")));
+ conf.writeXml(fos);
+ fos.close();
+
+ // start cluster and verify that volume was decommisioned
+ cluster.start();
+
+ Connector conn = cluster.getConnector("root", ROOT_PASSWORD);
+ conn.tableOperations().compact(tableNames[0], null, null, true, true);
+
+ verifyVolumesUsed(tableNames[0], true, v2);
+
+ // check that root tablet is not on volume 1
+ String zpath = ZooUtil.getRoot(new ZooKeeperInstance(cluster.getInstanceName(), cluster.getZooKeepers())) + RootTable.ZROOT_TABLET_PATH;
+ ZooKeeper zookeeper = new ZooKeeper(cluster.getZooKeepers(), 30000, null);
- String rootTabletDir = new String(zookeeper.getData(zpath, false, null), Constants.UTF8);
++ String rootTabletDir = new String(zookeeper.getData(zpath, false, null), UTF_8);
+ Assert.assertTrue(rootTabletDir.startsWith(v2.toString()));
+ zookeeper.close();
+
+ conn.tableOperations().clone(tableNames[0], tableNames[1], true, new HashMap<String,String>(), new HashSet<String>());
+
+ conn.tableOperations().flush(MetadataTable.NAME, null, null, true);
+ conn.tableOperations().flush(RootTable.NAME, null, null, true);
+
+ verifyVolumesUsed(tableNames[0], true, v2);
+ verifyVolumesUsed(tableNames[1], true, v2);
+
+ }
+
+ private void testReplaceVolume(boolean cleanShutdown) throws Exception {
+ String[] tableNames = getUniqueNames(3);
+
+ verifyVolumesUsed(tableNames[0], false, v1, v2);
+
+ // write to 2nd table, but do not flush data to disk before shutdown
+ writeData(tableNames[1], cluster.getConnector("root", ROOT_PASSWORD));
+
+ if (cleanShutdown)
+ Assert.assertEquals(0, cluster.exec(Admin.class, "stopAll").waitFor());
+
+ cluster.stop();
+
+ File v1f = new File(v1.toUri());
+ File v8f = new File(new File(v1.getParent().toUri()), "v8");
+ Assert.assertTrue("Failed to rename " + v1f + " to " + v8f, v1f.renameTo(v8f));
+ Path v8 = new Path(v8f.toURI());
+
+ File v2f = new File(v2.toUri());
+ File v9f = new File(new File(v2.getParent().toUri()), "v9");
+ Assert.assertTrue("Failed to rename " + v2f + " to " + v9f, v2f.renameTo(v9f));
+ Path v9 = new Path(v9f.toURI());
+
+ Configuration conf = new Configuration(false);
+ conf.addResource(new Path(cluster.getConfig().getConfDir().toURI().toString(), "accumulo-site.xml"));
+
+ conf.set(Property.INSTANCE_VOLUMES.getKey(), v8 + "," + v9);
+ conf.set(Property.INSTANCE_VOLUMES_REPLACEMENTS.getKey(), v1 + " " + v8 + "," + v2 + " " + v9);
+ BufferedOutputStream fos = new BufferedOutputStream(new FileOutputStream(new File(cluster.getConfig().getConfDir(), "accumulo-site.xml")));
+ conf.writeXml(fos);
+ fos.close();
+
+ // start cluster and verify that volumes were replaced
+ cluster.start();
+
+ verifyVolumesUsed(tableNames[0], true, v8, v9);
+ verifyVolumesUsed(tableNames[1], true, v8, v9);
+
+ // verify writes to new dir
+ getConnector().tableOperations().compact(tableNames[0], null, null, true, true);
+ getConnector().tableOperations().compact(tableNames[1], null, null, true, true);
+
+ verifyVolumesUsed(tableNames[0], true, v8, v9);
+ verifyVolumesUsed(tableNames[1], true, v8, v9);
+
+ // check that root tablet is not on volume 1 or 2
+ String zpath = ZooUtil.getRoot(new ZooKeeperInstance(cluster.getInstanceName(), cluster.getZooKeepers())) + RootTable.ZROOT_TABLET_PATH;
+ ZooKeeper zookeeper = new ZooKeeper(cluster.getZooKeepers(), 30000, null);
- String rootTabletDir = new String(zookeeper.getData(zpath, false, null), Constants.UTF8);
++ String rootTabletDir = new String(zookeeper.getData(zpath, false, null), UTF_8);
+ Assert.assertTrue(rootTabletDir.startsWith(v8.toString()) || rootTabletDir.startsWith(v9.toString()));
+ zookeeper.close();
+
+ getConnector().tableOperations().clone(tableNames[1], tableNames[2], true, new HashMap<String,String>(), new HashSet<String>());
+
+ getConnector().tableOperations().flush(MetadataTable.NAME, null, null, true);
+ getConnector().tableOperations().flush(RootTable.NAME, null, null, true);
+
+ verifyVolumesUsed(tableNames[0], true, v8, v9);
+ verifyVolumesUsed(tableNames[1], true, v8, v9);
+ verifyVolumesUsed(tableNames[2], true, v8, v9);
+ }
+
+ @Test
+ public void testCleanReplaceVolumes() throws Exception {
+ testReplaceVolume(true);
+ }
+
+ @Test
+ public void testDirtyReplaceVolumes() throws Exception {
+ testReplaceVolume(false);
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/test/java/org/apache/accumulo/test/functional/AddSplitIT.java
----------------------------------------------------------------------
diff --cc test/src/test/java/org/apache/accumulo/test/functional/AddSplitIT.java
index f657034,0000000..05de342
mode 100644,000000..100644
--- a/test/src/test/java/org/apache/accumulo/test/functional/AddSplitIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/functional/AddSplitIT.java
@@@ -1,140 -1,0 +1,141 @@@
+/*
+ * 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.functional;
+
++import static com.google.common.base.Charsets.UTF_8;
++
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.Map.Entry;
+import java.util.TreeSet;
+
- import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.AccumuloException;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.MutationsRejectedException;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Mutation;
+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 org.junit.Test;
+
+public class AddSplitIT extends SimpleMacIT {
+
+ @Override
+ protected int defaultTimeoutSeconds() {
+ return 60;
+ }
+
+ @Test
+ public void addSplitTest() throws Exception {
+
+ String tableName = getUniqueNames(1)[0];
+ Connector c = getConnector();
+ c.tableOperations().create(tableName);
+
+ insertData(tableName, 1l);
+
+ TreeSet<Text> splits = new TreeSet<Text>();
+ splits.add(new Text(String.format("%09d", 333)));
+ splits.add(new Text(String.format("%09d", 666)));
+
+ c.tableOperations().addSplits(tableName, splits);
+
+ UtilWaitThread.sleep(100);
+
+ Collection<Text> actualSplits = c.tableOperations().listSplits(tableName);
+
+ if (!splits.equals(new TreeSet<Text>(actualSplits))) {
+ throw new Exception(splits + " != " + actualSplits);
+ }
+
+ verifyData(tableName, 1l);
+ insertData(tableName, 2l);
+
+ // did not clear splits on purpose, it should ignore existing split points
+ // and still create the three additional split points
+
+ splits.add(new Text(String.format("%09d", 200)));
+ splits.add(new Text(String.format("%09d", 500)));
+ splits.add(new Text(String.format("%09d", 800)));
+
+ c.tableOperations().addSplits(tableName, splits);
+
+ UtilWaitThread.sleep(100);
+
+ actualSplits = c.tableOperations().listSplits(tableName);
+
+ if (!splits.equals(new TreeSet<Text>(actualSplits))) {
+ throw new Exception(splits + " != " + actualSplits);
+ }
+
+ verifyData(tableName, 2l);
+ }
+
+ private void verifyData(String tableName, long ts) throws Exception {
+ Scanner scanner = getConnector().createScanner(tableName, Authorizations.EMPTY);
+
+ Iterator<Entry<Key,Value>> iter = scanner.iterator();
+
+ for (int i = 0; i < 10000; i++) {
+ if (!iter.hasNext()) {
+ throw new Exception("row " + i + " not found");
+ }
+
+ Entry<Key,Value> entry = iter.next();
+
+ String row = String.format("%09d", i);
+
+ if (!entry.getKey().getRow().equals(new Text(row))) {
+ throw new Exception("unexpected row " + entry.getKey() + " " + i);
+ }
+
+ if (entry.getKey().getTimestamp() != ts) {
+ throw new Exception("unexpected ts " + entry.getKey() + " " + ts);
+ }
+
+ if (Integer.parseInt(entry.getValue().toString()) != i) {
+ throw new Exception("unexpected value " + entry + " " + i);
+ }
+ }
+
+ if (iter.hasNext()) {
+ throw new Exception("found more than expected " + iter.next());
+ }
+
+ }
+
+ private void insertData(String tableName, long ts) throws AccumuloException, AccumuloSecurityException, TableNotFoundException, MutationsRejectedException {
+ BatchWriter bw = getConnector().createBatchWriter(tableName, null);
+
+ for (int i = 0; i < 10000; i++) {
+ String row = String.format("%09d", i);
+
+ Mutation m = new Mutation(new Text(row));
- m.put(new Text("cf1"), new Text("cq1"), ts, new Value(Integer.toString(i).getBytes(Constants.UTF8)));
++ m.put(new Text("cf1"), new Text("cq1"), ts, new Value(Integer.toString(i).getBytes(UTF_8)));
+ bw.addMutation(m);
+ }
+
+ bw.close();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/test/java/org/apache/accumulo/test/functional/BadIteratorMincIT.java
----------------------------------------------------------------------
diff --cc test/src/test/java/org/apache/accumulo/test/functional/BadIteratorMincIT.java
index 0139a10,0000000..9c4492e
mode 100644,000000..100644
--- a/test/src/test/java/org/apache/accumulo/test/functional/BadIteratorMincIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/functional/BadIteratorMincIT.java
@@@ -1,110 -1,0 +1,110 @@@
+/*
+ * 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.functional;
+
++import static com.google.common.base.Charsets.UTF_8;
+import static org.junit.Assert.assertEquals;
+
+import java.util.EnumSet;
+import java.util.Map.Entry;
+
- import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.IteratorSetting;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.util.UtilWaitThread;
+import org.apache.hadoop.io.Text;
+import org.junit.Test;
+
+public class BadIteratorMincIT extends SimpleMacIT {
+
+ @Override
+ protected int defaultTimeoutSeconds() {
+ return 60;
+ }
+
+ @Test
+ public void test() throws Exception {
+ Connector c = getConnector();
+
+ String tableName = getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ IteratorSetting is = new IteratorSetting(30, BadIterator.class);
+ c.tableOperations().attachIterator(tableName, is, EnumSet.of(IteratorScope.minc));
+ BatchWriter bw = c.createBatchWriter(tableName, new BatchWriterConfig());
+
+ Mutation m = new Mutation(new Text("r1"));
- m.put(new Text("acf"), new Text(tableName), new Value("1".getBytes(Constants.UTF8)));
++ m.put(new Text("acf"), new Text(tableName), new Value("1".getBytes(UTF_8)));
+
+ bw.addMutation(m);
+ bw.close();
+
+ c.tableOperations().flush(tableName, null, null, false);
+ UtilWaitThread.sleep(1000);
+
+ // minc should fail, so there should be no files
+ FunctionalTestUtils.checkRFiles(c, tableName, 1, 1, 0, 0);
+
+ // try to scan table
+ Scanner scanner = c.createScanner(tableName, Authorizations.EMPTY);
+ int count = FunctionalTestUtils.count(scanner);
+ assertEquals("Did not see expected # entries " + count, 1, count);
+
+ // remove the bad iterator
+ c.tableOperations().removeIterator(tableName, BadIterator.class.getSimpleName(), EnumSet.of(IteratorScope.minc));
+
+ UtilWaitThread.sleep(5000);
+
+ // minc should complete
+ FunctionalTestUtils.checkRFiles(c, tableName, 1, 1, 1, 1);
+
+ count = 0;
+ for (@SuppressWarnings("unused")
+ Entry<Key,Value> entry : scanner) {
+ count++;
+ }
+
+ if (count != 1)
+ throw new Exception("Did not see expected # entries " + count);
+
+ // now try putting bad iterator back and deleting the table
+ c.tableOperations().attachIterator(tableName, is, EnumSet.of(IteratorScope.minc));
+ bw = c.createBatchWriter(tableName, new BatchWriterConfig());
+ m = new Mutation(new Text("r2"));
- m.put(new Text("acf"), new Text(tableName), new Value("1".getBytes(Constants.UTF8)));
++ m.put(new Text("acf"), new Text(tableName), new Value("1".getBytes(UTF_8)));
+ bw.addMutation(m);
+ bw.close();
+
+ // make sure property is given time to propagate
+ UtilWaitThread.sleep(500);
+
+ c.tableOperations().flush(tableName, null, null, false);
+
+ // make sure the flush has time to start
+ UtilWaitThread.sleep(1000);
+
+ // this should not hang
+ c.tableOperations().delete(tableName);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/test/java/org/apache/accumulo/test/functional/BatchScanSplitIT.java
----------------------------------------------------------------------
diff --cc test/src/test/java/org/apache/accumulo/test/functional/BatchScanSplitIT.java
index c67407d,0000000..688a326
mode 100644,000000..100644
--- a/test/src/test/java/org/apache/accumulo/test/functional/BatchScanSplitIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/functional/BatchScanSplitIT.java
@@@ -1,125 -1,0 +1,126 @@@
+/*
+ * 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.functional;
+
++import static com.google.common.base.Charsets.UTF_8;
++
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map.Entry;
+import java.util.Random;
+
- import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.BatchScanner;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.util.UtilWaitThread;
+import org.apache.accumulo.minicluster.impl.MiniAccumuloConfigImpl;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.Text;
+import org.junit.Test;
+
+public class BatchScanSplitIT extends ConfigurableMacIT {
+
+ @Override
+ public void configure(MiniAccumuloConfigImpl cfg, Configuration hadoopCoreSite) {
+ cfg.setSiteConfig(Collections.singletonMap(Property.TSERV_MAJC_DELAY.getKey(), "0"));
+ }
+
+ @Override
+ protected int defaultTimeoutSeconds() {
+ return 2 * 60;
+ }
+
+ @Test
+ public void test() throws Exception {
+ Connector c = getConnector();
+ String tableName = getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+
+ int numRows = 1 << 18;
+
+ BatchWriter bw = getConnector().createBatchWriter(tableName, new BatchWriterConfig());
+
+ for (int i = 0; i < numRows; i++) {
+ Mutation m = new Mutation(new Text(String.format("%09x", i)));
- m.put(new Text("cf1"), new Text("cq1"), new Value(String.format("%016x", numRows - i).getBytes(Constants.UTF8)));
++ m.put(new Text("cf1"), new Text("cq1"), new Value(String.format("%016x", numRows - i).getBytes(UTF_8)));
+ bw.addMutation(m);
+ }
+
+ bw.close();
+
+ getConnector().tableOperations().flush(tableName, null, null, true);
+
+ getConnector().tableOperations().setProperty(tableName, Property.TABLE_SPLIT_THRESHOLD.getKey(), "4K");
+
+ Collection<Text> splits = getConnector().tableOperations().listSplits(tableName);
+ while (splits.size() < 2) {
+ UtilWaitThread.sleep(1);
+ splits = getConnector().tableOperations().listSplits(tableName);
+ }
+
+ System.out.println("splits : " + splits);
+
+ Random random = new Random(19011230);
+ HashMap<Text,Value> expected = new HashMap<Text,Value>();
+ ArrayList<Range> ranges = new ArrayList<Range>();
+ for (int i = 0; i < 100; i++) {
+ int r = random.nextInt(numRows);
+ Text row = new Text(String.format("%09x", r));
- expected.put(row, new Value(String.format("%016x", numRows - r).getBytes(Constants.UTF8)));
++ expected.put(row, new Value(String.format("%016x", numRows - r).getBytes(UTF_8)));
+ ranges.add(new Range(row));
+ }
+
+ // logger.setLevel(Level.TRACE);
+
+ HashMap<Text,Value> found = new HashMap<Text,Value>();
+
+ for (int i = 0; i < 20; i++) {
+ BatchScanner bs = getConnector().createBatchScanner(tableName, Authorizations.EMPTY, 4);
+
+ found.clear();
+
+ long t1 = System.currentTimeMillis();
+
+ bs.setRanges(ranges);
+
+ for (Entry<Key,Value> entry : bs) {
+ found.put(entry.getKey().getRow(), entry.getValue());
+ }
+ bs.close();
+
+ long t2 = System.currentTimeMillis();
+
+ log.info(String.format("rate : %06.2f%n", ranges.size() / ((t2 - t1) / 1000.0)));
+
+ if (!found.equals(expected))
+ throw new Exception("Found and expected differ " + found + " " + expected);
+ }
+
+ splits = getConnector().tableOperations().listSplits(tableName);
+ log.info("splits : " + splits);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/test/java/org/apache/accumulo/test/functional/BatchWriterFlushIT.java
----------------------------------------------------------------------
diff --cc test/src/test/java/org/apache/accumulo/test/functional/BatchWriterFlushIT.java
index e2dd903,0000000..465936e
mode 100644,000000..100644
--- a/test/src/test/java/org/apache/accumulo/test/functional/BatchWriterFlushIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/functional/BatchWriterFlushIT.java
@@@ -1,181 -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.functional;
+
++import static com.google.common.base.Charsets.UTF_8;
++
+import java.util.Iterator;
+import java.util.Map.Entry;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+
- import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.AccumuloException;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.MutationsRejectedException;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.util.UtilWaitThread;
+import org.apache.hadoop.io.Text;
+import org.junit.Test;
+
+public class BatchWriterFlushIT extends SimpleMacIT {
+
+ private static final int NUM_TO_FLUSH = 100000;
+
+ @Override
+ protected int defaultTimeoutSeconds() {
+ return 90;
+ }
+
+ @Test
+ public void run() throws Exception {
+ Connector c = getConnector();
+ String[] tableNames = getUniqueNames(2);
+ String bwft = tableNames[0];
+ c.tableOperations().create(bwft);
+ String bwlt = tableNames[1];
+ c.tableOperations().create(bwlt);
+ runFlushTest(bwft);
+ runLatencyTest(bwlt);
+
+ }
+
+ private void runLatencyTest(String tableName) throws Exception {
+ // should automatically flush after 2 seconds
+ BatchWriter bw = getConnector().createBatchWriter(tableName, new BatchWriterConfig().setMaxLatency(1000, TimeUnit.MILLISECONDS));
+ Scanner scanner = getConnector().createScanner(tableName, Authorizations.EMPTY);
+
+ Mutation m = new Mutation(new Text(String.format("r_%10d", 1)));
- m.put(new Text("cf"), new Text("cq"), new Value("1".getBytes(Constants.UTF8)));
++ m.put(new Text("cf"), new Text("cq"), new Value("1".getBytes(UTF_8)));
+ bw.addMutation(m);
+
+ UtilWaitThread.sleep(500);
+
+ int count = 0;
+ for (@SuppressWarnings("unused")
+ Entry<Key,Value> entry : scanner) {
+ count++;
+ }
+
+ if (count != 0) {
+ throw new Exception("Flushed too soon");
+ }
+
+ UtilWaitThread.sleep(1500);
+
+ for (@SuppressWarnings("unused")
+ Entry<Key,Value> entry : scanner) {
+ count++;
+ }
+
+ if (count != 1) {
+ throw new Exception("Did not flush");
+ }
+
+ bw.close();
+ }
+
+ private void runFlushTest(String tableName) throws AccumuloException, AccumuloSecurityException, TableNotFoundException, MutationsRejectedException,
+ Exception {
+ BatchWriter bw = getConnector().createBatchWriter(tableName, new BatchWriterConfig());
+ Scanner scanner = getConnector().createScanner(tableName, Authorizations.EMPTY);
+ Random r = new Random();
+
+ for (int i = 0; i < 4; i++) {
+ for (int j = 0; j < NUM_TO_FLUSH; j++) {
+ int row = i * NUM_TO_FLUSH + j;
+
+ Mutation m = new Mutation(new Text(String.format("r_%10d", row)));
+ m.put(new Text("cf"), new Text("cq"), new Value(("" + row).getBytes()));
+ bw.addMutation(m);
+ }
+
+ bw.flush();
+
+ // do a few random lookups into the data just flushed
+
+ for (int k = 0; k < 10; k++) {
+ int rowToLookup = r.nextInt(NUM_TO_FLUSH) + i * NUM_TO_FLUSH;
+
+ scanner.setRange(new Range(new Text(String.format("r_%10d", rowToLookup))));
+
+ Iterator<Entry<Key,Value>> iter = scanner.iterator();
+
+ if (!iter.hasNext())
+ throw new Exception(" row " + rowToLookup + " not found after flush");
+
+ Entry<Key,Value> entry = iter.next();
+
+ if (iter.hasNext())
+ throw new Exception("Scanner returned too much");
+
+ verifyEntry(rowToLookup, entry);
+ }
+
+ // scan all data just flushed
+ scanner.setRange(new Range(new Text(String.format("r_%10d", i * NUM_TO_FLUSH)), true, new Text(String.format("r_%10d", (i + 1) * NUM_TO_FLUSH)), false));
+ Iterator<Entry<Key,Value>> iter = scanner.iterator();
+
+ for (int j = 0; j < NUM_TO_FLUSH; j++) {
+ int row = i * NUM_TO_FLUSH + j;
+
+ if (!iter.hasNext())
+ throw new Exception("Scan stopped permaturely at " + row);
+
+ Entry<Key,Value> entry = iter.next();
+
+ verifyEntry(row, entry);
+ }
+
+ if (iter.hasNext())
+ throw new Exception("Scanner returned too much");
+
+ }
+
+ bw.close();
+
+ // test adding a mutation to a closed batch writer
+ boolean caught = false;
+ try {
+ bw.addMutation(new Mutation(new Text("foobar")));
+ } catch (IllegalStateException ise) {
+ caught = true;
+ }
+
+ if (!caught) {
+ throw new Exception("Adding to closed batch writer did not fail");
+ }
+ }
+
+ private void verifyEntry(int row, Entry<Key,Value> entry) throws Exception {
+ if (!entry.getKey().getRow().toString().equals(String.format("r_%10d", row))) {
+ throw new Exception("Unexpected key returned, expected " + row + " got " + entry.getKey());
+ }
+
+ if (!entry.getValue().toString().equals("" + row)) {
+ throw new Exception("Unexpected value, expected " + row + " got " + entry.getValue());
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/test/java/org/apache/accumulo/test/functional/BulkFileIT.java
----------------------------------------------------------------------
diff --cc test/src/test/java/org/apache/accumulo/test/functional/BulkFileIT.java
index 7974c6c,0000000..ffe55bd
mode 100644,000000..100644
--- a/test/src/test/java/org/apache/accumulo/test/functional/BulkFileIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/functional/BulkFileIT.java
@@@ -1,120 -1,0 +1,121 @@@
+/*
+ * 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.functional;
+
++import static com.google.common.base.Charsets.UTF_8;
++
+import java.util.Iterator;
+import java.util.Map.Entry;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
- 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.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.file.FileOperations;
+import org.apache.accumulo.core.file.FileSKVWriter;
+import org.apache.accumulo.core.file.rfile.RFile;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.volume.VolumeConfiguration;
+import org.apache.accumulo.server.conf.ServerConfiguration;
+import org.apache.accumulo.server.trace.TraceFileSystem;
+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.junit.Test;
+
+public class BulkFileIT extends SimpleMacIT {
+
+ @Override
+ protected int defaultTimeoutSeconds() {
+ return 2 * 60;
+ }
+
+ @Test
+ public void testBulkFile() throws Exception {
+ Connector c = getConnector();
+ String tableName = getUniqueNames(1)[0];
+ c.tableOperations().create(tableName);
+ SortedSet<Text> splits = new TreeSet<Text>();
+ for (String split : "0333 0666 0999 1333 1666".split(" "))
+ splits.add(new Text(split));
+ c.tableOperations().addSplits(tableName, splits);
+ Configuration conf = new Configuration();
+ AccumuloConfiguration aconf = ServerConfiguration.getDefaultConfiguration();
+ FileSystem fs = TraceFileSystem.wrap(VolumeConfiguration.getDefaultVolume(conf, aconf).getFileSystem());
+
+ String dir = rootPath() + "/bulk_test_diff_files_89723987592_" + getUniqueNames(1)[0];
+
+ fs.delete(new Path(dir), true);
+
+ FileSKVWriter writer1 = FileOperations.getInstance().openWriter(dir + "/f1." + RFile.EXTENSION, fs, conf, aconf);
+ writer1.startDefaultLocalityGroup();
+ writeData(writer1, 0, 333);
+ writer1.close();
+
+ FileSKVWriter writer2 = FileOperations.getInstance().openWriter(dir + "/f2." + RFile.EXTENSION, fs, conf, aconf);
+ writer2.startDefaultLocalityGroup();
+ writeData(writer2, 334, 999);
+ writer2.close();
+
+ FileSKVWriter writer3 = FileOperations.getInstance().openWriter(dir + "/f3." + RFile.EXTENSION, fs, conf, aconf);
+ writer3.startDefaultLocalityGroup();
+ writeData(writer3, 1000, 1999);
+ writer3.close();
+
+ FunctionalTestUtils.bulkImport(c, fs, tableName, dir);
+
+ FunctionalTestUtils.checkRFiles(c, tableName, 6, 6, 1, 1);
+
+ verifyData(tableName, 0, 1999);
+
+ }
+
+ private void verifyData(String table, int s, int e) throws Exception {
+ Scanner scanner = getConnector().createScanner(table, Authorizations.EMPTY);
+
+ Iterator<Entry<Key,Value>> iter = scanner.iterator();
+
+ for (int i = s; i <= e; i++) {
+ if (!iter.hasNext())
+ throw new Exception("row " + i + " not found");
+
+ Entry<Key,Value> entry = iter.next();
+
+ String row = String.format("%04d", i);
+
+ if (!entry.getKey().getRow().equals(new Text(row)))
+ throw new Exception("unexpected row " + entry.getKey() + " " + i);
+
+ if (Integer.parseInt(entry.getValue().toString()) != i)
+ throw new Exception("unexpected value " + entry + " " + i);
+ }
+
+ if (iter.hasNext())
+ throw new Exception("found more than expected " + iter.next());
+ }
+
+ private void writeData(FileSKVWriter w, int s, int e) throws Exception {
+ for (int i = s; i <= e; i++) {
- w.append(new Key(new Text(String.format("%04d", i))), new Value(Integer.toString(i).getBytes(Constants.UTF8)));
++ w.append(new Key(new Text(String.format("%04d", i))), new Value(Integer.toString(i).getBytes(UTF_8)));
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/test/java/org/apache/accumulo/test/functional/ConcurrencyIT.java
----------------------------------------------------------------------
diff --cc test/src/test/java/org/apache/accumulo/test/functional/ConcurrencyIT.java
index 785a997,0000000..92bd714
mode 100644,000000..100644
--- a/test/src/test/java/org/apache/accumulo/test/functional/ConcurrencyIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/functional/ConcurrencyIT.java
@@@ -1,147 -1,0 +1,148 @@@
+/*
+ * 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.functional;
+
++import static com.google.common.base.Charsets.UTF_8;
++
+import java.util.Collections;
+import java.util.EnumSet;
+
- import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.AccumuloException;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.IteratorSetting;
+import org.apache.accumulo.core.client.MutationsRejectedException;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.TableExistsException;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.util.UtilWaitThread;
+import org.apache.accumulo.minicluster.impl.MiniAccumuloConfigImpl;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.Text;
+import org.junit.Test;
+
+public class ConcurrencyIT extends ConfigurableMacIT {
+
+ static class ScanTask extends Thread {
+
+ int count = 0;
+ Scanner scanner;
+
+ ScanTask(Connector conn, long time) throws Exception {
+ scanner = conn.createScanner("cct", Authorizations.EMPTY);
+ IteratorSetting slow = new IteratorSetting(30, "slow", SlowIterator.class);
+ SlowIterator.setSleepTime(slow, time);
+ scanner.addScanIterator(slow);
+ }
+
+ @Override
+ public void run() {
+ count = FunctionalTestUtils.count(scanner);
+ }
+
+ }
+
+ @Override
+ public void configure(MiniAccumuloConfigImpl cfg, Configuration hadoopCoreSite) {
+ cfg.setSiteConfig(Collections.singletonMap(Property.TSERV_MAJC_DELAY.getKey(), "1"));
+ }
+
+ @Override
+ protected int defaultTimeoutSeconds() {
+ return 2 * 60;
+ }
+
+ /*
+ * Below is a diagram of the operations in this test over time.
+ *
+ * Scan 0 |------------------------------| Scan 1 |----------| Minc 1 |-----| Scan 2 |----------| Scan 3 |---------------| Minc 2 |-----| Majc 1 |-----|
+ */
+
+ @Test
+ public void run() throws Exception {
+ Connector c = getConnector();
+ runTest(c);
+ }
+
+ static void runTest(Connector c) throws AccumuloException, AccumuloSecurityException, TableExistsException, TableNotFoundException,
+ MutationsRejectedException, Exception, InterruptedException {
+ c.tableOperations().create("cct");
+ IteratorSetting is = new IteratorSetting(10, SlowIterator.class);
+ SlowIterator.setSleepTime(is, 50);
+ c.tableOperations().attachIterator("cct", is, EnumSet.of(IteratorScope.minc, IteratorScope.majc));
+ c.tableOperations().setProperty("cct", Property.TABLE_MAJC_RATIO.getKey(), "1.0");
+
+ BatchWriter bw = c.createBatchWriter("cct", new BatchWriterConfig());
+ for (int i = 0; i < 50; i++) {
+ Mutation m = new Mutation(new Text(String.format("%06d", i)));
- m.put(new Text("cf1"), new Text("cq1"), new Value("foo".getBytes(Constants.UTF8)));
++ m.put(new Text("cf1"), new Text("cq1"), new Value("foo".getBytes(UTF_8)));
+ bw.addMutation(m);
+ }
+ bw.flush();
+
+ ScanTask st0 = new ScanTask(c, 300);
+ st0.start();
+
+ ScanTask st1 = new ScanTask(c, 100);
+ st1.start();
+
+ UtilWaitThread.sleep(50);
+ c.tableOperations().flush("cct", null, null, true);
+
+ for (int i = 0; i < 50; i++) {
+ Mutation m = new Mutation(new Text(String.format("%06d", i)));
- m.put(new Text("cf1"), new Text("cq1"), new Value("foo".getBytes(Constants.UTF8)));
++ m.put(new Text("cf1"), new Text("cq1"), new Value("foo".getBytes(UTF_8)));
+ bw.addMutation(m);
+ }
+
+ bw.flush();
+
+ ScanTask st2 = new ScanTask(c, 100);
+ st2.start();
+
+ st1.join();
+ st2.join();
+ if (st1.count != 50)
+ throw new Exception("Thread 1 did not see 50, saw " + st1.count);
+
+ if (st2.count != 50)
+ throw new Exception("Thread 2 did not see 50, saw " + st2.count);
+
+ ScanTask st3 = new ScanTask(c, 150);
+ st3.start();
+
+ UtilWaitThread.sleep(50);
+ c.tableOperations().flush("cct", null, null, false);
+
+ st3.join();
+ if (st3.count != 50)
+ throw new Exception("Thread 3 did not see 50, saw " + st3.count);
+
+ st0.join();
+ if (st0.count != 50)
+ throw new Exception("Thread 0 did not see 50, saw " + st0.count);
+
+ bw.close();
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/test/java/org/apache/accumulo/test/functional/ConstraintIT.java
----------------------------------------------------------------------
diff --cc test/src/test/java/org/apache/accumulo/test/functional/ConstraintIT.java
index d5e9188,0000000..7e5944e
mode 100644,000000..100644
--- a/test/src/test/java/org/apache/accumulo/test/functional/ConstraintIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/functional/ConstraintIT.java
@@@ -1,317 -1,0 +1,318 @@@
+/*
+ * 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.functional;
+
++import static com.google.common.base.Charsets.UTF_8;
++
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map.Entry;
+
- import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.MutationsRejectedException;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.data.ConstraintViolationSummary;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.util.UtilWaitThread;
+import org.apache.accumulo.examples.simple.constraints.AlphaNumKeyConstraint;
+import org.apache.accumulo.examples.simple.constraints.NumericValueConstraint;
+import org.apache.hadoop.io.Text;
+import org.junit.Test;
+
+public class ConstraintIT extends SimpleMacIT {
+
+ @Override
+ protected int defaultTimeoutSeconds() {
+ return 30;
+ }
+
+ @Test
+ public void run() throws Exception {
+ String[] tableNames = getUniqueNames(3);
+ Connector c = getConnector();
+ for (String table : tableNames) {
+ c.tableOperations().create(table);
+ c.tableOperations().addConstraint(table, NumericValueConstraint.class.getName());
+ c.tableOperations().addConstraint(table, AlphaNumKeyConstraint.class.getName());
+ }
+
+ // Logger logger = Logger.getLogger(Constants.CORE_PACKAGE_NAME);
+ // logger.setLevel(Level.TRACE);
+
+ test1(tableNames[0]);
+
+ // logger.setLevel(Level.TRACE);
+
+ test2(tableNames[1], false);
+ test2(tableNames[2], true);
+ }
+
+ private void test1(String tableName) throws Exception {
+ BatchWriter bw = getConnector().createBatchWriter(tableName, new BatchWriterConfig());
+
+ Mutation mut1 = new Mutation(new Text("r1"));
- mut1.put(new Text("cf1"), new Text("cq1"), new Value("123".getBytes(Constants.UTF8)));
++ mut1.put(new Text("cf1"), new Text("cq1"), new Value("123".getBytes(UTF_8)));
+
+ bw.addMutation(mut1);
+
+ // should not throw any exceptions
+ bw.close();
+
+ bw = getConnector().createBatchWriter(tableName, new BatchWriterConfig());
+
+ // create a mutation with a non numeric value
+ Mutation mut2 = new Mutation(new Text("r1"));
- mut2.put(new Text("cf1"), new Text("cq1"), new Value("123a".getBytes(Constants.UTF8)));
++ mut2.put(new Text("cf1"), new Text("cq1"), new Value("123a".getBytes(UTF_8)));
+
+ bw.addMutation(mut2);
+
+ boolean sawMRE = false;
+
+ try {
+ bw.close();
+ // should not get here
+ throw new Exception("Test failed, constraint did not catch bad mutation");
+ } catch (MutationsRejectedException mre) {
+ sawMRE = true;
+
+ // verify constraint violation summary
+ List<ConstraintViolationSummary> cvsl = mre.getConstraintViolationSummaries();
+
+ if (cvsl.size() != 1) {
+ throw new Exception("Unexpected constraints");
+ }
+
+ for (ConstraintViolationSummary cvs : cvsl) {
+ if (!cvs.constrainClass.equals(NumericValueConstraint.class.getName())) {
+ throw new Exception("Unexpected constraint class " + cvs.constrainClass);
+ }
+
+ if (cvs.numberOfViolatingMutations != 1) {
+ throw new Exception("Unexpected # violating mutations " + cvs.numberOfViolatingMutations);
+ }
+ }
+ }
+
+ if (!sawMRE) {
+ throw new Exception("Did not see MutationsRejectedException");
+ }
+
+ // verify mutation did not go through
+ Scanner scanner = getConnector().createScanner(tableName, Authorizations.EMPTY);
+ scanner.setRange(new Range(new Text("r1")));
+
+ Iterator<Entry<Key,Value>> iter = scanner.iterator();
+ Entry<Key,Value> entry = iter.next();
+
+ if (!entry.getKey().getRow().equals(new Text("r1")) || !entry.getKey().getColumnFamily().equals(new Text("cf1"))
- || !entry.getKey().getColumnQualifier().equals(new Text("cq1")) || !entry.getValue().equals(new Value("123".getBytes(Constants.UTF8)))) {
++ || !entry.getKey().getColumnQualifier().equals(new Text("cq1")) || !entry.getValue().equals(new Value("123".getBytes(UTF_8)))) {
+ throw new Exception("Unexpected key or value " + entry.getKey() + " " + entry.getValue());
+ }
+
+ if (iter.hasNext()) {
+ entry = iter.next();
+ throw new Exception("Unexpected extra key or value " + entry.getKey() + " " + entry.getValue());
+ }
+
+ // remove the numeric value constraint
+ getConnector().tableOperations().removeConstraint(tableName, 2);
+ UtilWaitThread.sleep(1000);
+
+ // now should be able to add a non numeric value
+ bw = getConnector().createBatchWriter(tableName, new BatchWriterConfig());
+ bw.addMutation(mut2);
+ bw.close();
+
+ // verify mutation went through
+ iter = scanner.iterator();
+ entry = iter.next();
+
+ if (!entry.getKey().getRow().equals(new Text("r1")) || !entry.getKey().getColumnFamily().equals(new Text("cf1"))
- || !entry.getKey().getColumnQualifier().equals(new Text("cq1")) || !entry.getValue().equals(new Value("123a".getBytes(Constants.UTF8)))) {
++ || !entry.getKey().getColumnQualifier().equals(new Text("cq1")) || !entry.getValue().equals(new Value("123a".getBytes(UTF_8)))) {
+ throw new Exception("Unexpected key or value " + entry.getKey() + " " + entry.getValue());
+ }
+
+ if (iter.hasNext()) {
+ entry = iter.next();
+ throw new Exception("Unexpected extra key or value " + entry.getKey() + " " + entry.getValue());
+ }
+
+ // add a constraint that references a non-existant class
+ getConnector().tableOperations().setProperty(tableName, Property.TABLE_CONSTRAINT_PREFIX + "1", "com.foobar.nonExistantClass");
+ UtilWaitThread.sleep(1000);
+
+ // add a mutation
+ bw = getConnector().createBatchWriter(tableName, new BatchWriterConfig());
+
+ Mutation mut3 = new Mutation(new Text("r1"));
- mut3.put(new Text("cf1"), new Text("cq1"), new Value("foo".getBytes(Constants.UTF8)));
++ mut3.put(new Text("cf1"), new Text("cq1"), new Value("foo".getBytes(UTF_8)));
+
+ bw.addMutation(mut3);
+
+ sawMRE = false;
+
+ try {
+ bw.close();
+ // should not get here
+ throw new Exception("Test failed, mutation went through when table had bad constraints");
+ } catch (MutationsRejectedException mre) {
+ sawMRE = true;
+ }
+
+ if (!sawMRE) {
+ throw new Exception("Did not see MutationsRejectedException");
+ }
+
+ // verify the mutation did not go through
+ iter = scanner.iterator();
+ entry = iter.next();
+
+ if (!entry.getKey().getRow().equals(new Text("r1")) || !entry.getKey().getColumnFamily().equals(new Text("cf1"))
- || !entry.getKey().getColumnQualifier().equals(new Text("cq1")) || !entry.getValue().equals(new Value("123a".getBytes(Constants.UTF8)))) {
++ || !entry.getKey().getColumnQualifier().equals(new Text("cq1")) || !entry.getValue().equals(new Value("123a".getBytes(UTF_8)))) {
+ throw new Exception("Unexpected key or value " + entry.getKey() + " " + entry.getValue());
+ }
+
+ if (iter.hasNext()) {
+ entry = iter.next();
+ throw new Exception("Unexpected extra key or value " + entry.getKey() + " " + entry.getValue());
+ }
+
+ // remove the bad constraint
+ getConnector().tableOperations().removeConstraint(tableName, 1);
+ UtilWaitThread.sleep(1000);
+
+ // try the mutation again
+ bw = getConnector().createBatchWriter(tableName, new BatchWriterConfig());
+ bw.addMutation(mut3);
+ bw.close();
+
+ // verify it went through
+ iter = scanner.iterator();
+ entry = iter.next();
+
+ if (!entry.getKey().getRow().equals(new Text("r1")) || !entry.getKey().getColumnFamily().equals(new Text("cf1"))
- || !entry.getKey().getColumnQualifier().equals(new Text("cq1")) || !entry.getValue().equals(new Value("foo".getBytes(Constants.UTF8)))) {
++ || !entry.getKey().getColumnQualifier().equals(new Text("cq1")) || !entry.getValue().equals(new Value("foo".getBytes(UTF_8)))) {
+ throw new Exception("Unexpected key or value " + entry.getKey() + " " + entry.getValue());
+ }
+
+ if (iter.hasNext()) {
+ entry = iter.next();
+ throw new Exception("Unexpected extra key or value " + entry.getKey() + " " + entry.getValue());
+ }
+ }
+
+ private Mutation newMut(String row, String cf, String cq, String val) {
+ Mutation mut1 = new Mutation(new Text(row));
- mut1.put(new Text(cf), new Text(cq), new Value(val.getBytes(Constants.UTF8)));
++ mut1.put(new Text(cf), new Text(cq), new Value(val.getBytes(UTF_8)));
+ return mut1;
+ }
+
+ private void test2(String table, boolean doFlush) throws Exception {
+ // test sending multiple mutations with multiple constrain violations... all of the non violating mutations
+ // should go through
+ int numericErrors = 2;
+
+ BatchWriter bw = getConnector().createBatchWriter(table, new BatchWriterConfig());
+ bw.addMutation(newMut("r1", "cf1", "cq1", "123"));
+ bw.addMutation(newMut("r1", "cf1", "cq2", "I'm a bad value"));
+ if (doFlush) {
+ try {
+ bw.flush();
+ throw new Exception("Didn't find a bad mutation");
+ } catch (MutationsRejectedException mre) {
+ // ignored
+ try {
+ bw.close();
+ } catch (MutationsRejectedException ex) {
+ // ignored
+ }
+ bw = getConnector().createBatchWriter(table, new BatchWriterConfig());
+ numericErrors = 1;
+ }
+ }
+ bw.addMutation(newMut("r1", "cf1", "cq3", "I'm a naughty value"));
+ bw.addMutation(newMut("@bad row@", "cf1", "cq2", "456"));
+ bw.addMutation(newMut("r1", "cf1", "cq4", "789"));
+
+ boolean sawMRE = false;
+
+ try {
+ bw.close();
+ // should not get here
+ throw new Exception("Test failed, constraint did not catch bad mutation");
+ } catch (MutationsRejectedException mre) {
+ System.out.println(mre);
+
+ sawMRE = true;
+
+ // verify constraint violation summary
+ List<ConstraintViolationSummary> cvsl = mre.getConstraintViolationSummaries();
+
+ if (cvsl.size() != 2) {
+ throw new Exception("Unexpected constraints");
+ }
+
+ HashMap<String,Integer> expected = new HashMap<String,Integer>();
+
+ expected.put("org.apache.accumulo.examples.simple.constraints.NumericValueConstraint", numericErrors);
+ expected.put("org.apache.accumulo.examples.simple.constraints.AlphaNumKeyConstraint", 1);
+
+ for (ConstraintViolationSummary cvs : cvsl) {
+ if (expected.get(cvs.constrainClass) != cvs.numberOfViolatingMutations) {
+ throw new Exception("Unexpected " + cvs.constrainClass + " " + cvs.numberOfViolatingMutations);
+ }
+ }
+ }
+
+ if (!sawMRE) {
+ throw new Exception("Did not see MutationsRejectedException");
+ }
+
+ Scanner scanner = getConnector().createScanner(table, Authorizations.EMPTY);
+
+ Iterator<Entry<Key,Value>> iter = scanner.iterator();
+
+ Entry<Key,Value> entry = iter.next();
+
+ if (!entry.getKey().getRow().equals(new Text("r1")) || !entry.getKey().getColumnFamily().equals(new Text("cf1"))
- || !entry.getKey().getColumnQualifier().equals(new Text("cq1")) || !entry.getValue().equals(new Value("123".getBytes(Constants.UTF8)))) {
++ || !entry.getKey().getColumnQualifier().equals(new Text("cq1")) || !entry.getValue().equals(new Value("123".getBytes(UTF_8)))) {
+ throw new Exception("Unexpected key or value " + entry.getKey() + " " + entry.getValue());
+ }
+
+ entry = iter.next();
+
+ if (!entry.getKey().getRow().equals(new Text("r1")) || !entry.getKey().getColumnFamily().equals(new Text("cf1"))
- || !entry.getKey().getColumnQualifier().equals(new Text("cq4")) || !entry.getValue().equals(new Value("789".getBytes(Constants.UTF8)))) {
++ || !entry.getKey().getColumnQualifier().equals(new Text("cq4")) || !entry.getValue().equals(new Value("789".getBytes(UTF_8)))) {
+ throw new Exception("Unexpected key or value " + entry.getKey() + " " + entry.getValue());
+ }
+
+ if (iter.hasNext()) {
+ entry = iter.next();
+ throw new Exception("Unexpected extra key or value " + entry.getKey() + " " + entry.getValue());
+ }
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/test/java/org/apache/accumulo/test/functional/CreateAndUseIT.java
----------------------------------------------------------------------
diff --cc test/src/test/java/org/apache/accumulo/test/functional/CreateAndUseIT.java
index b9b54c7,0000000..5b5249b
mode 100644,000000..100644
--- a/test/src/test/java/org/apache/accumulo/test/functional/CreateAndUseIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/functional/CreateAndUseIT.java
@@@ -1,128 -1,0 +1,129 @@@
+/*
+ * 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.functional;
+
++import static com.google.common.base.Charsets.UTF_8;
++
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.Map.Entry;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
- import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.BatchScanner;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.hadoop.io.Text;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import com.google.common.collect.Iterators;
+
+public class CreateAndUseIT extends SimpleMacIT {
+
+ @Override
+ protected int defaultTimeoutSeconds() {
+ return 4 * 60;
+ }
+
+ private static SortedSet<Text> splits;
+
+ @BeforeClass
+ public static void createData() throws Exception {
+ splits = new TreeSet<Text>();
+
+ for (int i = 1; i < 256; i++) {
+ splits.add(new Text(String.format("%08x", i << 8)));
+ }
+ }
+
+ @Test
+ public void verifyDataIsPresent() throws Exception {
+ Text cf = new Text("cf1");
+ Text cq = new Text("cq1");
+
+ String tableName = getUniqueNames(1)[0];
+ getConnector().tableOperations().create(tableName);
+ getConnector().tableOperations().addSplits(tableName, splits);
+ BatchWriter bw = getConnector().createBatchWriter(tableName, new BatchWriterConfig());
+
+ for (int i = 1; i < 257; i++) {
+ Mutation m = new Mutation(new Text(String.format("%08x", (i << 8) - 16)));
- m.put(cf, cq, new Value(Integer.toString(i).getBytes(Constants.UTF8)));
++ m.put(cf, cq, new Value(Integer.toString(i).getBytes(UTF_8)));
+
+ bw.addMutation(m);
+ }
+
+ bw.close();
+ Scanner scanner1 = getConnector().createScanner(tableName, Authorizations.EMPTY);
+
+ int ei = 1;
+
+ for (Entry<Key,Value> entry : scanner1) {
+ Assert.assertEquals(String.format("%08x", (ei << 8) - 16), entry.getKey().getRow().toString());
+ Assert.assertEquals(Integer.toString(ei), entry.getValue().toString());
+
+ ei++;
+ }
+
+ Assert.assertEquals("Did not see expected number of rows", 257, ei);
+ }
+
+ @Test
+ public void createTableAndScan() throws Exception {
+ String table2 = getUniqueNames(1)[0];
+ getConnector().tableOperations().create(table2);
+ getConnector().tableOperations().addSplits(table2, splits);
+ Scanner scanner2 = getConnector().createScanner(table2, Authorizations.EMPTY);
+ int count = 0;
+ for (Entry<Key,Value> entry : scanner2) {
+ if (entry != null)
+ count++;
+ }
+
+ if (count != 0) {
+ throw new Exception("Did not see expected number of entries, count = " + count);
+ }
+ }
+
+ @Test
+ public void createTableAndBatchScan() throws Exception {
+ ArrayList<Range> ranges = new ArrayList<Range>();
+ for (int i = 1; i < 257; i++) {
+ ranges.add(new Range(new Text(String.format("%08x", (i << 8) - 16))));
+ }
+
+ String table3 = getUniqueNames(1)[0];
+ getConnector().tableOperations().create(table3);
+ getConnector().tableOperations().addSplits(table3, splits);
+ BatchScanner bs = getConnector().createBatchScanner(table3, Authorizations.EMPTY, 3);
+ bs.setRanges(ranges);
+ Iterator<Entry<Key,Value>> iter = bs.iterator();
+ int count = Iterators.size(iter);
+ bs.close();
+
+ Assert.assertEquals("Did not expect to find any entries", 0, count);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/test/java/org/apache/accumulo/test/functional/DeleteEverythingIT.java
----------------------------------------------------------------------
diff --cc test/src/test/java/org/apache/accumulo/test/functional/DeleteEverythingIT.java
index 83b5b24,0000000..0578ef4
mode 100644,000000..100644
--- a/test/src/test/java/org/apache/accumulo/test/functional/DeleteEverythingIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/functional/DeleteEverythingIT.java
@@@ -1,94 -1,0 +1,94 @@@
+/*
+ * 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.functional;
+
++import static com.google.common.base.Charsets.UTF_8;
+import static org.junit.Assert.*;
+
+import java.util.Collections;
+import java.util.Map.Entry;
+
- import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.util.UtilWaitThread;
+import org.apache.accumulo.minicluster.impl.MiniAccumuloConfigImpl;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.Text;
+import org.junit.Test;
+
+public class DeleteEverythingIT extends ConfigurableMacIT {
+
+ @Override
+ public void configure(MiniAccumuloConfigImpl cfg, Configuration hadoopCoreSite) {
+ cfg.setSiteConfig(Collections.singletonMap(Property.TSERV_MAJC_DELAY.getKey(), "1s"));
+ }
+
+ @Override
+ protected int defaultTimeoutSeconds() {
+ return 60;
+ }
+
+ @Test
+ public void run() throws Exception {
+ Connector c = getConnector();
+ c.tableOperations().create("de");
+ BatchWriter bw = getConnector().createBatchWriter("de", new BatchWriterConfig());
+ Mutation m = new Mutation(new Text("foo"));
- m.put(new Text("bar"), new Text("1910"), new Value("5".getBytes(Constants.UTF8)));
++ m.put(new Text("bar"), new Text("1910"), new Value("5".getBytes(UTF_8)));
+ bw.addMutation(m);
+ bw.flush();
+
+ getConnector().tableOperations().flush("de", null, null, true);
+
+ FunctionalTestUtils.checkRFiles(c, "de", 1, 1, 1, 1);
+
+ m = new Mutation(new Text("foo"));
+ m.putDelete(new Text("bar"), new Text("1910"));
+ bw.addMutation(m);
+ bw.flush();
+
+ Scanner scanner = getConnector().createScanner("de", Authorizations.EMPTY);
+ scanner.setRange(new Range());
+ int count = FunctionalTestUtils.count(scanner);
+ assertEquals("count == " + count, 0, count);
+ getConnector().tableOperations().flush("de", null, null, true);
+
+ getConnector().tableOperations().setProperty("de", Property.TABLE_MAJC_RATIO.getKey(), "1.0");
+ UtilWaitThread.sleep(4000);
+
+ FunctionalTestUtils.checkRFiles(c, "de", 1, 1, 0, 0);
+
+ bw.close();
+
+ count = 0;
+ for (@SuppressWarnings("unused")
+ Entry<Key,Value> entry : scanner) {
+ count++;
+ }
+
+ if (count != 0)
+ throw new Exception("count == " + count);
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9b20a9d4/test/src/test/java/org/apache/accumulo/test/functional/DeleteRowsSplitIT.java
----------------------------------------------------------------------
diff --cc test/src/test/java/org/apache/accumulo/test/functional/DeleteRowsSplitIT.java
index a8237a6,0000000..d35ba9f
mode 100644,000000..100644
--- a/test/src/test/java/org/apache/accumulo/test/functional/DeleteRowsSplitIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/functional/DeleteRowsSplitIT.java
@@@ -1,143 -1,0 +1,143 @@@
+/*
+ * 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.functional;
+
++import static com.google.common.base.Charsets.UTF_8;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
- import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Mutation;
+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 org.apache.log4j.Logger;
+import org.junit.Test;
+
+// attempt to reproduce ACCUMULO-315
+public class DeleteRowsSplitIT extends SimpleMacIT {
+
+ @Override
+ protected int defaultTimeoutSeconds() {
+ return 4 * 60;
+ }
+
+ private static final Logger log = Logger.getLogger(DeleteRowsSplitIT.class);
+
+ private static final String LETTERS = "abcdefghijklmnopqrstuvwxyz";
+ static final SortedSet<Text> SPLITS = new TreeSet<Text>();
+ static final List<String> ROWS = new ArrayList<String>();
+ static {
- for (byte b : LETTERS.getBytes(Constants.UTF8)) {
++ for (byte b : LETTERS.getBytes(UTF_8)) {
+ SPLITS.add(new Text(new byte[] {b}));
- ROWS.add(new String(new byte[] {b}, Constants.UTF8));
++ ROWS.add(new String(new byte[] {b}, UTF_8));
+ }
+ }
+
+ @Test
+ public void run() throws Exception {
+ // Delete ranges of rows, and verify the are removed
+ // Do this while adding many splits
+ final String tableName = getUniqueNames(1)[0];
+
+ // Eliminate whole tablets
+ for (int test = 0; test < 10; test++) {
+ // create a table
+ log.info("Test " + test);
+ getConnector().tableOperations().create(tableName);
+
+ // put some data in it
+ fillTable(tableName);
+
+ // generate a random delete range
+ final Text start = new Text();
+ final Text end = new Text();
+ generateRandomRange(start, end);
+
+ // initiate the delete range
+ final boolean fail[] = {false};
+ Thread t = new Thread() {
+ @Override
+ public void run() {
+ try {
+ // split the table
+ final SortedSet<Text> afterEnd = SPLITS.tailSet(new Text(end.toString() + "\0"));
+ getConnector().tableOperations().addSplits(tableName, afterEnd);
+ } catch (Exception ex) {
+ log.error(ex, ex);
+ synchronized (fail) {
+ fail[0] = true;
+ }
+ }
+ }
+ };
+ t.start();
+
+ UtilWaitThread.sleep(test * 2);
+
+ getConnector().tableOperations().deleteRows(tableName, start, end);
+
+ t.join();
+ synchronized (fail) {
+ assertTrue(!fail[0]);
+ }
+
+ // scan the table
+ Scanner scanner = getConnector().createScanner(tableName, Authorizations.EMPTY);
+ for (Entry<Key,Value> entry : scanner) {
+ Text row = entry.getKey().getRow();
+ assertTrue(row.compareTo(start) <= 0 || row.compareTo(end) > 0);
+ }
+
+ // delete the table
+ getConnector().tableOperations().delete(tableName);
+ }
+ }
+
+ private void generateRandomRange(Text start, Text end) {
+ List<String> bunch = new ArrayList<String>(ROWS);
+ Collections.shuffle(bunch);
+ if (bunch.get(0).compareTo((bunch.get(1))) < 0) {
+ start.set(bunch.get(0));
+ end.set(bunch.get(1));
+ } else {
+ start.set(bunch.get(1));
+ end.set(bunch.get(0));
+ }
+
+ }
+
+ private void fillTable(String table) throws Exception {
+ BatchWriter bw = getConnector().createBatchWriter(table, new BatchWriterConfig());
+ for (String row : ROWS) {
+ Mutation m = new Mutation(row);
+ m.put("cf", "cq", "value");
+ bw.addMutation(m);
+ }
+ bw.close();
+ }
+}