accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From e..@apache.org
Subject [1/2] accumulo git commit: ACCUMULO-3663 initial implementation
Date Tue, 17 Mar 2015 12:41:39 GMT
Repository: accumulo
Updated Branches:
  refs/heads/master 5c670fa4d -> ca64067f2


ACCUMULO-3663 initial implementation


Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/52b7785d
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/52b7785d
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/52b7785d

Branch: refs/heads/master
Commit: 52b7785db7a679610cb759c53735c852217fe363
Parents: 5c670fa
Author: Eric C. Newton <eric.newton@gmail.com>
Authored: Mon Mar 16 16:42:07 2015 -0400
Committer: Eric C. Newton <eric.newton@gmail.com>
Committed: Mon Mar 16 16:42:07 2015 -0400

----------------------------------------------------------------------
 .../accumulo/test/randomwalk/Environment.java   |   2 +-
 .../apache/accumulo/test/randomwalk/State.java  |   2 +-
 .../test/randomwalk/concurrent/Replication.java | 160 +++++++++++++++++++
 .../replication/ReplicationRandomWalkIT.java    |  55 +++++++
 .../randomwalk/conf/modules/Concurrent.xml      |   5 +
 5 files changed, 222 insertions(+), 2 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo/blob/52b7785d/test/src/main/java/org/apache/accumulo/test/randomwalk/Environment.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/accumulo/test/randomwalk/Environment.java b/test/src/main/java/org/apache/accumulo/test/randomwalk/Environment.java
index 92f6427..c39f390 100644
--- a/test/src/main/java/org/apache/accumulo/test/randomwalk/Environment.java
+++ b/test/src/main/java/org/apache/accumulo/test/randomwalk/Environment.java
@@ -83,7 +83,7 @@ public class Environment {
    * @throws NullPointerException
    *           if p is null
    */
-  Environment(Properties p) {
+  public Environment(Properties p) {
     checkNotNull(p);
     this.p = p;
   }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/52b7785d/test/src/main/java/org/apache/accumulo/test/randomwalk/State.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/accumulo/test/randomwalk/State.java b/test/src/main/java/org/apache/accumulo/test/randomwalk/State.java
index 18e21e2..9b74ad4 100644
--- a/test/src/main/java/org/apache/accumulo/test/randomwalk/State.java
+++ b/test/src/main/java/org/apache/accumulo/test/randomwalk/State.java
@@ -28,7 +28,7 @@ public class State {
   /**
    * Creates new empty state.
    */
-  State() {}
+  public State() {}
 
   /**
    * Sets a state object.

http://git-wip-us.apache.org/repos/asf/accumulo/blob/52b7785d/test/src/main/java/org/apache/accumulo/test/randomwalk/concurrent/Replication.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/accumulo/test/randomwalk/concurrent/Replication.java
b/test/src/main/java/org/apache/accumulo/test/randomwalk/concurrent/Replication.java
new file mode 100644
index 0000000..bc2d64c
--- /dev/null
+++ b/test/src/main/java/org/apache/accumulo/test/randomwalk/concurrent/Replication.java
@@ -0,0 +1,160 @@
+/*
+ * 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.randomwalk.concurrent;
+
+import static org.apache.accumulo.core.conf.Property.MASTER_REPLICATION_SCAN_INTERVAL;
+import static org.apache.accumulo.core.conf.Property.REPLICATION_NAME;
+import static org.apache.accumulo.core.conf.Property.REPLICATION_PEERS;
+import static org.apache.accumulo.core.conf.Property.REPLICATION_PEER_PASSWORD;
+import static org.apache.accumulo.core.conf.Property.REPLICATION_PEER_USER;
+import static org.apache.accumulo.core.conf.Property.REPLICATION_WORK_ASSIGNMENT_SLEEP;
+import static org.apache.accumulo.core.conf.Property.REPLICATION_WORK_PROCESSOR_DELAY;
+import static org.apache.accumulo.core.conf.Property.REPLICATION_WORK_PROCESSOR_PERIOD;
+import static org.apache.accumulo.core.conf.Property.TABLE_REPLICATION;
+import static org.apache.accumulo.core.conf.Property.TABLE_REPLICATION_TARGET;
+import static org.apache.accumulo.server.replication.ReplicaSystemFactory.getPeerConfigurationValue;
+
+import java.util.Map.Entry;
+import java.util.Properties;
+import java.util.Random;
+import java.util.SortedSet;
+import java.util.TreeSet;
+import java.util.UUID;
+
+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.Scanner;
+import org.apache.accumulo.core.client.admin.InstanceOperations;
+import org.apache.accumulo.core.client.admin.TableOperations;
+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.replication.ReplicationTable;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.util.UtilWaitThread;
+import org.apache.accumulo.test.randomwalk.Environment;
+import org.apache.accumulo.test.randomwalk.State;
+import org.apache.accumulo.test.randomwalk.Test;
+import org.apache.accumulo.tserver.replication.AccumuloReplicaSystem;
+import org.apache.hadoop.io.Text;
+
+public class Replication extends Test {
+
+  @Override
+  public void visit(State state, Environment env, Properties props) throws Exception {
+    final Connector c = env.getConnector();
+    final Instance inst = c.getInstance();
+    final String instName = inst.getInstanceName();
+    final InstanceOperations iOps = c.instanceOperations();
+    final TableOperations tOps = c.tableOperations();
+    iOps.setProperty(REPLICATION_NAME.getKey(), instName);
+    iOps.setProperty(MASTER_REPLICATION_SCAN_INTERVAL.getKey(), "1s");
+    iOps.setProperty(REPLICATION_WORK_ASSIGNMENT_SLEEP.getKey(), "1s");
+    iOps.setProperty(REPLICATION_WORK_PROCESSOR_DELAY.getKey(), "1s");
+    iOps.setProperty(REPLICATION_WORK_PROCESSOR_PERIOD.getKey(), "1s");
+    iOps.setProperty(REPLICATION_PEERS.getKey() + instName,
+        getPeerConfigurationValue(AccumuloReplicaSystem.class, instName + "," + inst.getZooKeepers()));
+    iOps.setProperty(REPLICATION_PEER_USER.getKey() + instName , env.getUserName());
+    iOps.setProperty(REPLICATION_PEER_PASSWORD.getKey() + instName, env.getPassword());
+
+    ReplicationTable.setOnline(c);
+    boolean online = ReplicationTable.isOnline(c);
+    for (int i = 0; i < 10; i++) {
+      if (online) break;
+      UtilWaitThread.sleep(2000);
+      online = ReplicationTable.isOnline(c);
+    }
+    assertTrue("Replication table was not online", online);
+
+    final String sourceTable = ("repl-source-" + UUID.randomUUID()).replace('-', '_');
+    final String destTable = ("repl-dest-" + UUID.randomUUID()).replace('-', '_');
+    final String tables[] = new String[] { sourceTable, destTable };
+
+    for (String tableName : tables) {
+      log.debug("creating " + tableName);
+      tOps.create(tableName);
+    }
+    // zookeeper propagation wait
+    UtilWaitThread.sleep(5 * 1000);
+
+    final String destID = tOps.tableIdMap().get(destTable);
+    tOps.setProperty(sourceTable, TABLE_REPLICATION.getKey(), "true");
+    tOps.setProperty(sourceTable, TABLE_REPLICATION_TARGET.getKey() + instName, destID);
+
+    Random rand = new Random(System.currentTimeMillis());
+    for (String tableName : tables) {
+      if (rand.nextBoolean()) {
+        splitTable(tOps, tableName);
+      }
+    }
+
+    final BatchWriter bw = c.createBatchWriter(sourceTable, null);
+    final int ROWS = 1000;
+    final int COLS = 50;
+    for (int row = 0; row < ROWS; row++) {
+      Mutation m = new Mutation(Integer.toString(row));
+      for (int col = 0; col < COLS; col++) {
+        m.put("", Integer.toString(col), "");
+      }
+      bw.addMutation(m);
+    }
+    bw.close();
+    UtilWaitThread.sleep(1000);
+
+    c.replicationOperations().drain(sourceTable);
+
+    Scanner scanner = c.createScanner(destTable, Authorizations.EMPTY);
+    int row = 0;
+    int col = 0;
+    for (Entry<Key,Value> entry : scanner) {
+      assertEquals(row, Integer.parseInt(entry.getKey().getRow().toString()));
+      assertEquals(col, Integer.parseInt(entry.getKey().getColumnQualifier().toString()));
+      col++;
+      if (col == COLS) {
+        row++;
+        col = 0;
+      }
+    }
+    assertEquals(ROWS, row);
+    assertEquals(COLS, col);
+
+    for (String tableName : tables) {
+      log.debug("Deleting " + tableName);
+      tOps.delete(tableName);
+    }
+  }
+
+  private void assertEquals(int expected, int actual) {
+    if (expected != actual)
+      throw new RuntimeException(String.format("%d fails to match expected value %d", actual,
expected));
+  }
+
+  private void assertTrue(String string, boolean test) {
+      if (!test)
+        throw new RuntimeException(string);
+  }
+
+  private void splitTable(TableOperations tOps, String tableName) throws Exception {
+    SortedSet<Text> splits = new TreeSet<>();
+    for (int i = 1; i <= 9; i++) {
+      splits.add(new Text(Integer.toString(i)));
+    }
+    log.debug("Adding splits to " + tableName);
+    tOps.addSplits(tableName, splits);
+  }
+}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/52b7785d/test/src/test/java/org/apache/accumulo/test/replication/ReplicationRandomWalkIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/replication/ReplicationRandomWalkIT.java
b/test/src/test/java/org/apache/accumulo/test/replication/ReplicationRandomWalkIT.java
new file mode 100644
index 0000000..e91d5ac
--- /dev/null
+++ b/test/src/test/java/org/apache/accumulo/test/replication/ReplicationRandomWalkIT.java
@@ -0,0 +1,55 @@
+/*
+ * 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.replication;
+
+import java.util.Properties;
+
+import org.apache.accumulo.core.client.AccumuloException;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.test.functional.ConfigurableMacIT;
+import org.apache.accumulo.test.randomwalk.Environment;
+import org.apache.accumulo.test.randomwalk.concurrent.Replication;
+import org.junit.Test;
+
+public class ReplicationRandomWalkIT extends ConfigurableMacIT {
+
+  @Test
+  public void runReplicationRandomWalkStep() throws Exception {
+    Replication r = new Replication();
+
+    Environment env = new Environment(new Properties()) {
+      @Override
+      public String getUserName() {
+        return "root";
+      }
+
+      @Override
+      public String getPassword() {
+        return ROOT_PASSWORD;
+      }
+
+      @Override
+      public Connector getConnector() throws AccumuloException, AccumuloSecurityException
{
+        return ReplicationRandomWalkIT.this.getConnector();
+      }
+
+    };
+    r.visit(null, env, null);
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/52b7785d/test/system/randomwalk/conf/modules/Concurrent.xml
----------------------------------------------------------------------
diff --git a/test/system/randomwalk/conf/modules/Concurrent.xml b/test/system/randomwalk/conf/modules/Concurrent.xml
index 96fbbce..74e8fef 100644
--- a/test/system/randomwalk/conf/modules/Concurrent.xml
+++ b/test/system/randomwalk/conf/modules/Concurrent.xml
@@ -31,6 +31,7 @@
   <edge id="ct.DeleteTable" weight="1000"/>
   <edge id="ct.RenameTable" weight="1000"/>
   <edge id="ct.ScanTable" weight="500"/>
+  <edge id="ct.Replication" weight="500"/>
   <edge id="ct.IsolatedScan" weight="500"/>
   <edge id="ct.AddSplits" weight="1000"/>
   <edge id="ct.ListSplits" weight="1000"/>
@@ -90,6 +91,10 @@
   <edge id="dummy.ToAll" weight="1"/>
 </node>
 
+<node id="ct.Replication">
+  <edge id="dummy.ToAll" weight="1"/>
+</node>
+
 <node id="ct.IsolatedScan">
   <edge id="dummy.ToAll" weight="1"/>
 </node>


Mime
View raw message