accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From els...@apache.org
Subject [25/51] [abbrv] git commit: ACCUMULO-378 Can't use '_' as the row separator for Order records.
Date Sat, 14 Jun 2014 04:55:25 GMT
ACCUMULO-378 Can't use '_' as the row separator for Order records.

The ULongLexicoder *might* create bytes that actually equal the '_', which
will mess up the splitting logic of the row key. Switch it to a \x00 instead
and find the last instance of it.


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

Branch: refs/heads/master
Commit: 49fc9855f996ae0f5b3cc20e03e77ea8f707d640
Parents: 0ff0e02
Author: Josh Elser <elserj@apache.org>
Authored: Wed May 28 20:07:52 2014 -0400
Committer: Josh Elser <elserj@apache.org>
Committed: Wed May 28 20:58:43 2014 -0400

----------------------------------------------------------------------
 .../core/replication/ReplicationSchema.java     | 30 ++++++++++++++++----
 .../core/replication/ReplicationSchemaTest.java | 12 ++++++++
 .../accumulo/master/replication/WorkDriver.java |  6 +++-
 3 files changed, 41 insertions(+), 7 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo/blob/49fc9855/core/src/main/java/org/apache/accumulo/core/replication/ReplicationSchema.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/replication/ReplicationSchema.java
b/core/src/main/java/org/apache/accumulo/core/replication/ReplicationSchema.java
index 8699bd2..ab350e6 100644
--- a/core/src/main/java/org/apache/accumulo/core/replication/ReplicationSchema.java
+++ b/core/src/main/java/org/apache/accumulo/core/replication/ReplicationSchema.java
@@ -153,11 +153,11 @@ public class ReplicationSchema {
    * Holds the order in which files needed for replication were closed. The intent is to
be able to guarantee that files which were closed earlier were
    * replicated first and we don't replay data in the wrong order on our peers
    * <p>
-   * <code>encodedTimeOfClosure_hdfs://localhost:8020/accumulo/wal/tserver+port/WAL
order:source_table_id [] -> Status Protobuf</code>
+   * <code>encodedTimeOfClosure\x00hdfs://localhost:8020/accumulo/wal/tserver+port/WAL
order:source_table_id [] -> Status Protobuf</code>
    */
   public static class OrderSection {
     public static final Text NAME = new Text("order");
-    public static final String ROW_SEPARATOR = "_";
+    public static final Text ROW_SEPARATOR = new Text(new byte[]{0});
     private static final ULongLexicoder longEncoder = new ULongLexicoder();
 
     /**
@@ -218,10 +218,10 @@ public class ReplicationSchema {
       Path p = new Path(file);
       String pathString = p.toUri().toString();
 
-      log.info("Normalized {} into {}", file, pathString);
+      log.trace("Normalized {} into {}", file, pathString);
 
       // Append the file as a suffix to the row
-      row.append((ROW_SEPARATOR + pathString).getBytes(), 0, pathString.length() + ROW_SEPARATOR.length());
+      row.append((ROW_SEPARATOR + pathString).getBytes(), 0, pathString.length() + ROW_SEPARATOR.getLength());
 
       // Make the mutation and add the column update
       return new Mutation(row);
@@ -249,7 +249,16 @@ public class ReplicationSchema {
 
     public static long getTimeClosed(Key k, Text buff) {
       k.getRow(buff);
-      int offset = buff.find(ROW_SEPARATOR);
+      int offset = 0;
+      // find the last offset
+      while (true) {
+        int nextOffset = buff.find(ROW_SEPARATOR.toString(), offset + 1);
+        if (-1 == nextOffset) {
+          break;
+        }
+        offset = nextOffset;
+      }
+
       if (-1 == offset) {
         throw new IllegalArgumentException("Row does not contain expected separator for OrderSection");
       }
@@ -266,7 +275,16 @@ public class ReplicationSchema {
 
     public static String getFile(Key k, Text buff) {
       k.getRow(buff);
-      int offset = buff.find(ROW_SEPARATOR);
+      int offset = 0;
+      // find the last offset
+      while (true) {
+        int nextOffset = buff.find(ROW_SEPARATOR.toString(), offset + 1);
+        if (-1 == nextOffset) {
+          break;
+        }
+        offset = nextOffset;
+      }
+
       if (-1 == offset) {
         throw new IllegalArgumentException("Row does not contain expected separator for OrderSection");
       }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/49fc9855/core/src/test/java/org/apache/accumulo/core/replication/ReplicationSchemaTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/accumulo/core/replication/ReplicationSchemaTest.java
b/core/src/test/java/org/apache/accumulo/core/replication/ReplicationSchemaTest.java
index d321153..3822641 100644
--- a/core/src/test/java/org/apache/accumulo/core/replication/ReplicationSchemaTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/replication/ReplicationSchemaTest.java
@@ -119,4 +119,16 @@ public class ReplicationSchemaTest {
     Assert.assertEquals("/accumulo/file", OrderSection.getFile(k, buff));
     Assert.assertEquals(now, OrderSection.getTimeClosed(k, buff));
   }
+
+  @Test
+  public void separatorDoesntInterferWithSplit() {
+    Text buff = new Text();
+    // Cycle through 2*128 values
+    for (long i = 1; i < 258; i++) {
+      Mutation m = OrderSection.createMutation("/accumulo/file", i);
+      Key k = new Key(new Text(m.getRow()));
+      Assert.assertEquals("/accumulo/file", OrderSection.getFile(k, buff));
+      Assert.assertEquals(i, OrderSection.getTimeClosed(k, buff));
+    }
+  }
 }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/49fc9855/server/master/src/main/java/org/apache/accumulo/master/replication/WorkDriver.java
----------------------------------------------------------------------
diff --git a/server/master/src/main/java/org/apache/accumulo/master/replication/WorkDriver.java
b/server/master/src/main/java/org/apache/accumulo/master/replication/WorkDriver.java
index 8c3e3e3..43a74a8 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/replication/WorkDriver.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/replication/WorkDriver.java
@@ -93,7 +93,11 @@ public class WorkDriver extends Daemon {
 
     while (master.stillMaster()) {
       // Assign the work using the configured implementation
-      assigner.assignWork();
+      try {
+        assigner.assignWork();
+      } catch (Exception e) {
+        log.error("Error while assigning work", e);
+      }
 
       long sleepTime = conf.getTimeInMillis(Property.REPLICATION_WORK_ASSIGNMENT_SLEEP);
       log.debug("Sleeping {} ms before next work assignment", sleepTime);


Mime
View raw message