accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From els...@apache.org
Subject [09/50] [abbrv] git commit: ACCUMULO-2581 Add tests for work assignment using the distributed work queue
Date Fri, 09 May 2014 15:28:47 GMT
ACCUMULO-2581 Add tests for work assignment using the distributed work queue


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

Branch: refs/heads/ACCUMULO-378
Commit: cc9a72ac736c50fa798a821fcd603fc033994eff
Parents: 31f4e83
Author: Josh Elser <elserj@apache.org>
Authored: Thu May 1 21:16:08 2014 -0400
Committer: Josh Elser <elserj@apache.org>
Committed: Thu May 1 21:16:08 2014 -0400

----------------------------------------------------------------------
 .../accumulo/core/replication/StatusUtil.java   |  2 +-
 .../server/zookeeper/DistributedWorkQueue.java  | 12 +++
 .../replication/ReplicationWorkAssigner.java    | 17 ++--
 .../ReplicationWorkAssignerTest.java            | 97 ++++++++++++++++++--
 4 files changed, 112 insertions(+), 16 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo/blob/cc9a72ac/core/src/main/java/org/apache/accumulo/core/replication/StatusUtil.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/replication/StatusUtil.java b/core/src/main/java/org/apache/accumulo/core/replication/StatusUtil.java
index 74327fd..842d945 100644
--- a/core/src/main/java/org/apache/accumulo/core/replication/StatusUtil.java
+++ b/core/src/main/java/org/apache/accumulo/core/replication/StatusUtil.java
@@ -200,7 +200,7 @@ public class StatusUtil {
    */
   public static boolean isWorkRequired(Status status) {
     if (status.getInfiniteEnd()) {
-      return Long.MAX_VALUE == status.getBegin();
+      return Long.MAX_VALUE != status.getBegin();
     } else {
       return status.getBegin() < status.getEnd();
     }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/cc9a72ac/server/base/src/main/java/org/apache/accumulo/server/zookeeper/DistributedWorkQueue.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/zookeeper/DistributedWorkQueue.java
b/server/base/src/main/java/org/apache/accumulo/server/zookeeper/DistributedWorkQueue.java
index 4738c2b..b1907fb 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/zookeeper/DistributedWorkQueue.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/zookeeper/DistributedWorkQueue.java
@@ -16,6 +16,7 @@
  */
 package org.apache.accumulo.server.zookeeper;
 
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -205,6 +206,17 @@ public class DistributedWorkQueue {
       }
     }, r.nextInt(60 * 1000), 60 * 1000);
   }
+
+  /**
+   * Adds work to the queue, automatically converting the String to bytes using {@link StandardCharsets#UTF_8}
+   * @param workId
+   * @param data
+   * @throws KeeperException
+   * @throws InterruptedException
+   */
+  public void addWork(String workId, String data) throws KeeperException, InterruptedException
{
+    addWork(workId, data.getBytes(StandardCharsets.UTF_8));
+  }
   
   public void addWork(String workId, byte[] data) throws KeeperException, InterruptedException
{
     if (workId.equalsIgnoreCase(LOCKS_NODE))

http://git-wip-us.apache.org/repos/asf/accumulo/blob/cc9a72ac/server/master/src/main/java/org/apache/accumulo/master/replication/ReplicationWorkAssigner.java
----------------------------------------------------------------------
diff --git a/server/master/src/main/java/org/apache/accumulo/master/replication/ReplicationWorkAssigner.java
b/server/master/src/main/java/org/apache/accumulo/master/replication/ReplicationWorkAssigner.java
index f3a8825..370fcfd 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/replication/ReplicationWorkAssigner.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/replication/ReplicationWorkAssigner.java
@@ -215,11 +215,12 @@ public class ReplicationWorkAssigner implements Runnable {
         if (StatusUtil.isWorkRequired(status)) {
           Path p = new Path(file);
           String filename = p.getName();
+          entry.getKey().getColumnQualifier(buffer);
+          String key = filename + "|" + buffer;
 
           // And, we haven't already queued this file up for work already
-          if (!queuedWork.contains(filename)) {
-            entry.getKey().getColumnQualifier(buffer);
-            queueWork(file, filename, buffer);
+          if (!queuedWork.contains(key)) {
+            queueWork(key, file);
           }
         }
       }
@@ -238,10 +239,9 @@ public class ReplicationWorkAssigner implements Runnable {
    * @param filename
    *          Filename
    */
-  protected void queueWork(String path, String filename, Text serializedTarget) {
+  protected void queueWork(String key, String path) {
     try {
-      String key = filename + "|" + serializedTarget;
-      workQueue.addWork(key, path.getBytes(StandardCharsets.UTF_8));
+      workQueue.addWork(key, path);
       queuedWork.add(key);
     } catch (KeeperException | InterruptedException e) {
       log.warn("Could not queue work for {}", path, e);
@@ -252,11 +252,12 @@ public class ReplicationWorkAssigner implements Runnable {
    * Iterate over the queued work to remove entries that have been completed.
    */
   protected void cleanupFinishedWork() {
-    Iterator<String> work = queuedWork.iterator();
+    final Iterator<String> work = queuedWork.iterator();
+    final String instanceId = master.getInstance().getInstanceID();
     while (work.hasNext()) {
       String filename = work.next();
       // Null equates to the work was finished
-      if (null == zooCache.get(ZooUtil.getRoot(master.getInstance()) + Constants.ZREPLICATION
+ "/" + filename)) {
+      if (null == zooCache.get(ZooUtil.getRoot(instanceId) + Constants.ZREPLICATION + "/"
+ filename)) {
         work.remove();
       }
     }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/cc9a72ac/server/master/src/test/java/org/apache/accumulo/master/replication/ReplicationWorkAssignerTest.java
----------------------------------------------------------------------
diff --git a/server/master/src/test/java/org/apache/accumulo/master/replication/ReplicationWorkAssignerTest.java
b/server/master/src/test/java/org/apache/accumulo/master/replication/ReplicationWorkAssignerTest.java
index b596126..89a1ddd 100644
--- a/server/master/src/test/java/org/apache/accumulo/master/replication/ReplicationWorkAssignerTest.java
+++ b/server/master/src/test/java/org/apache/accumulo/master/replication/ReplicationWorkAssignerTest.java
@@ -20,16 +20,20 @@ import static org.easymock.EasyMock.createMock;
 import static org.easymock.EasyMock.expect;
 import static org.easymock.EasyMock.expectLastCall;
 import static org.easymock.EasyMock.replay;
+import static org.easymock.EasyMock.verify;
 
 import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
 import java.util.HashSet;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Set;
 import java.util.UUID;
 
+import org.apache.accumulo.core.Constants;
 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.mock.MockInstance;
 import org.apache.accumulo.core.client.security.tokens.PasswordToken;
 import org.apache.accumulo.core.data.Mutation;
@@ -41,6 +45,7 @@ import org.apache.accumulo.core.security.TablePermission;
 import org.apache.accumulo.master.Master;
 import org.apache.accumulo.server.replication.ReplicationTable;
 import org.apache.accumulo.server.zookeeper.DistributedWorkQueue;
+import org.apache.accumulo.server.zookeeper.ZooCache;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.Text;
 import org.junit.Assert;
@@ -79,13 +84,14 @@ public class ReplicationWorkAssignerTest {
     assigner.setWorkQueue(workQueue);
 
     Path p = new Path("/accumulo/wal/tserver+port/" + UUID.randomUUID());
-    assigner.queueWork(p.toString(), p.getName(), serializedTarget);
     
-    workQueue.addWork(p.getName(), p.toString().getBytes(StandardCharsets.UTF_8));
+    workQueue.addWork(p.getName() + "|" + serializedTarget.toString(), p.toString());
     expectLastCall().once();
 
     replay(workQueue);
 
+    assigner.queueWork(p.getName() + "|" + serializedTarget, p.toString());
+
     Assert.assertEquals(1, queuedWork.size());
     Assert.assertEquals(p.getName() + "|" + serializedTarget, queuedWork.iterator().next());
   }
@@ -102,6 +108,8 @@ public class ReplicationWorkAssignerTest {
     assigner.setWorkQueue(workQueue);
     assigner.initializeQueuedWork();
 
+    verify(workQueue);
+
     Set<String> queuedWork = assigner.getQueuedWork();
     Assert.assertEquals("Expected existing work and queued work to be the same size", existingWork.size(),
queuedWork.size());
     Assert.assertTrue("Expected all existing work to be queued", queuedWork.containsAll(existingWork));
@@ -141,28 +149,34 @@ public class ReplicationWorkAssignerTest {
     @SuppressWarnings("unchecked")
     HashSet<String> queuedWork = createMock(HashSet.class);
     assigner.setQueuedWork(queuedWork);
+    assigner.setWorkQueue(workQueue);
+    assigner.setMaxQueueSize(Integer.MAX_VALUE);
 
     expect(queuedWork.size()).andReturn(0).anyTimes();
 
     // Make sure we expect the invocations in the correct order (accumulo is sorted)
     if (file1.compareTo(file2) <= 0) {
       String key = filename1 + "|" + serializedTarget1;
-      workQueue.addWork(key, file1.getBytes(StandardCharsets.UTF_8));
+      expect(queuedWork.contains(key)).andReturn(false);
+      workQueue.addWork(key, file1);
       expectLastCall().once();
       expect(queuedWork.add(key)).andReturn(true).once();
       
       key = filename2 + "|" + serializedTarget2;
-      workQueue.addWork(key, file2.getBytes(StandardCharsets.UTF_8));
+      expect(queuedWork.contains(key)).andReturn(false);
+      workQueue.addWork(key, file2);
       expectLastCall().once();
       expect(queuedWork.add(key)).andReturn(true).once();
     } else {
       String key = filename2 + "|" + serializedTarget2;
-      workQueue.addWork(key, file2.getBytes(StandardCharsets.UTF_8));
+      expect(queuedWork.contains(key)).andReturn(false);
+      workQueue.addWork(key, file2);
       expectLastCall().once();
       expect(queuedWork.add(key)).andReturn(true).once();
 
       key = filename1 + "|" + serializedTarget1;
-      workQueue.addWork(key, file1.getBytes(StandardCharsets.UTF_8));
+      expect(queuedWork.contains(key)).andReturn(false);
+      workQueue.addWork(key, file1);
       expectLastCall().once();
       expect(queuedWork.add(key)).andReturn(true).once();
     }
@@ -170,6 +184,8 @@ public class ReplicationWorkAssignerTest {
     replay(queuedWork, workQueue);
 
     assigner.createWork();
+
+    verify(queuedWork, workQueue);
   }
 
   @Test
@@ -192,6 +208,7 @@ public class ReplicationWorkAssignerTest {
     BatchWriter bw = ReplicationTable.getBatchWriter(conn);
     String filename1 = UUID.randomUUID().toString(), filename2 = UUID.randomUUID().toString();
     String file1 = "/accumulo/wal/tserver+port/" + filename1, file2 = "/accumulo/wal/tserver+port/"
+ filename2;
+
     Mutation m = new Mutation(file1);
     WorkSection.add(m, serializedTarget1, StatusUtil.newFileValue());
     bw.addMutation(m);
@@ -206,12 +223,78 @@ public class ReplicationWorkAssignerTest {
     @SuppressWarnings("unchecked")
     HashSet<String> queuedWork = createMock(HashSet.class);
     assigner.setQueuedWork(queuedWork);
+    assigner.setMaxQueueSize(Integer.MAX_VALUE);
 
-    expect(queuedWork.size()).andReturn(0).anyTimes();
+    expect(queuedWork.size()).andReturn(0).times(2);
 
     replay(queuedWork, workQueue);
 
     assigner.createWork();
+    
+    verify(queuedWork, workQueue);
+  }
+
+  @Test
+  public void workNotInZooKeeperIsCleanedUp() {
+    Set<String> queuedWork = new LinkedHashSet<>(Arrays.asList("wal1", "wal2"));
+    assigner.setQueuedWork(queuedWork);
+
+    Instance inst = createMock(Instance.class);
+    ZooCache cache = createMock(ZooCache.class);
+    assigner.setZooCache(cache);
+
+    expect(master.getInstance()).andReturn(inst);
+    expect(inst.getInstanceID()).andReturn("id");
+    expect(cache.get(Constants.ZROOT + "/id" + Constants.ZREPLICATION + "/wal1")).andReturn(null);
+    expect(cache.get(Constants.ZROOT + "/id" + Constants.ZREPLICATION + "/wal2")).andReturn(null);
+
+    replay(cache, inst, master);
+
+    assigner.cleanupFinishedWork();
+
+    verify(cache, inst, master);
+    Assert.assertTrue("Queued work was not emptied", queuedWork.isEmpty());
   }
 
+  @Test
+  public void workNotReAdded() throws Exception  {
+    Set<String> queuedWork = new HashSet<>();
+
+    assigner.setQueuedWork(queuedWork);
+
+    ReplicationTarget target = new ReplicationTarget("cluster1", "table1"); 
+    Text serializedTarget = ReplicationTarget.toText(target);
+
+    queuedWork.add("wal1|" + serializedTarget.toString());
+
+    MockInstance inst = new MockInstance(test.getMethodName());
+    Credentials creds = new Credentials("root", new PasswordToken(""));
+    Connector conn = inst.getConnector(creds.getPrincipal(), creds.getToken());
+
+    // Set the connector
+    assigner.setConnector(conn);
+
+    // Create and grant ourselves write to the replication table
+    ReplicationTable.create(conn);
+    conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.WRITE);
+
+    // Create two mutations, both of which need replication work done
+    BatchWriter bw = ReplicationTable.getBatchWriter(conn);
+    String file1 = "/accumulo/wal/tserver+port/wal1";
+    Mutation m = new Mutation(file1);
+    WorkSection.add(m, serializedTarget, StatusUtil.openWithUnknownLengthValue());
+    bw.addMutation(m);
+
+    bw.close();
+
+    DistributedWorkQueue workQueue = createMock(DistributedWorkQueue.class);
+    assigner.setWorkQueue(workQueue);
+    assigner.setMaxQueueSize(Integer.MAX_VALUE);
+
+    replay(workQueue);
+    
+    assigner.createWork();
+
+    verify(workQueue);
+  }
 }


Mime
View raw message