accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From els...@apache.org
Subject [4/5] git commit: ACCUMULO-378 When re-syncing to where we left off on reads, we need to track all tids for our table
Date Thu, 22 May 2014 17:59:52 GMT
ACCUMULO-378 When re-syncing to where we left off on reads, we need to track all tids for our
table

Fixes an issue where when the DEFINE_TABLET wasn't contained in the batch
of log entries that we were reading, we ignored all of the mutations.When
we read past all of the old data, we still need to track the tids
for the table which we're replicating.


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

Branch: refs/heads/ACCUMULO-378
Commit: 59177233fd903d5c69592c67e603c58bc2a0ed2a
Parents: abea3c6
Author: Josh Elser <elserj@apache.org>
Authored: Thu May 22 13:37:14 2014 -0400
Committer: Josh Elser <elserj@apache.org>
Committed: Thu May 22 13:37:14 2014 -0400

----------------------------------------------------------------------
 .../replication/AccumuloReplicaSystem.java      | 43 +++++++---
 .../replication/AccumuloReplicaSystemTest.java  | 82 +++++++++++++++++++-
 2 files changed, 109 insertions(+), 16 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo/blob/59177233/server/tserver/src/main/java/org/apache/accumulo/tserver/replication/AccumuloReplicaSystem.java
----------------------------------------------------------------------
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/replication/AccumuloReplicaSystem.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/replication/AccumuloReplicaSystem.java
index 6cd3358..ca1382f 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/replication/AccumuloReplicaSystem.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/replication/AccumuloReplicaSystem.java
@@ -194,7 +194,7 @@ public class AccumuloReplicaSystem implements ReplicaSystem {
                     return kvs;
                   }
                 } else {
-                  WalReplication edits = getWalEdits(target, p, status, sizeLimit);
+                  WalReplication edits = getWalEdits(target, getWalStream(p), p, status,
sizeLimit);
 
                   // If we have some edits to send
                   if (0 < edits.walEdits.getEditsSize()) {
@@ -206,8 +206,9 @@ public class AccumuloReplicaSystem implements ReplicaSystem {
                     // We don't have to replicate every LogEvent in the file (only Mutation
LogEvents), but we
                     // want to track progress in the file relative to all LogEvents (to avoid
duplicative processing/replication)
                     return edits;
-                  } else if (edits.entriesConsumed == Long.MAX_VALUE) {
-                    // Even if we send no data, we must record the new begin value to account
for the inf+ length
+                  } else if (edits.entriesConsumed > 0) {
+                    // Even if we send no data, we want to record a non-zero new begin value
to avoid checking the same
+                    // log entries multiple times to determine if they should be sent
                     return edits;
                   }
                 }
@@ -249,13 +250,15 @@ public class AccumuloReplicaSystem implements ReplicaSystem {
     throw new UnsupportedOperationException();
   }
 
-  protected WalReplication getWalEdits(ReplicationTarget target, Path p, Status status, long
sizeLimit) throws IOException {
-    DFSLoggerInputStreams streams = DfsLogger.readHeaderAndReturnStream(fs, p, conf);
-    DataInputStream wal = streams.getDecryptingInputStream();
+  protected WalReplication getWalEdits(ReplicationTarget target, DataInputStream wal, Path
p, Status status, long sizeLimit) throws IOException {
     LogFileKey key = new LogFileKey();
     LogFileValue value = new LogFileValue();
 
+    Set<Integer> desiredTids = new HashSet<>();
+
     // Read through the stuff we've already processed in a previous replication attempt
+    // We also need to track the tids that occurred earlier in the file as mutations
+    // later on might use that tid
     for (long i = 0; i < status.getBegin(); i++) {
       try {
         key.readFields(wal);
@@ -264,9 +267,19 @@ public class AccumuloReplicaSystem implements ReplicaSystem {
         log.warn("Unexpectedly reached the end of file.");
         return new WalReplication(new WalEdits(), 0, 0, 0);
       }
+
+      switch (key.event) {
+        case DEFINE_TABLET:
+          if (target.getSourceTableId().equals(key.tablet.getTableId().toString())) {
+            desiredTids.add(key.tid);
+          }
+          break;
+        default:
+          break;
+      }
     }
 
-    WalReplication repl = getEdits(wal, sizeLimit, target, status, p);
+    WalReplication repl = getEdits(wal, sizeLimit, target, status, p, desiredTids);
 
     log.debug("Read {} WAL entries and retained {} bytes of WAL entries for replication to
peer '{}'", (Long.MAX_VALUE == repl.entriesConsumed) ? "all"
         : repl.entriesConsumed, repl.sizeInBytes, p);
@@ -274,7 +287,12 @@ public class AccumuloReplicaSystem implements ReplicaSystem {
     return repl;
   }
 
-  protected WalReplication getEdits(DataInputStream wal, long sizeLimit, ReplicationTarget
target, Status status, Path p) throws IOException {
+  protected DataInputStream getWalStream(Path p) throws IOException {
+    DFSLoggerInputStreams streams = DfsLogger.readHeaderAndReturnStream(fs, p, conf);
+    return streams.getDecryptingInputStream();
+  }
+
+  protected WalReplication getEdits(DataInputStream wal, long sizeLimit, ReplicationTarget
target, Status status, Path p, Set<Integer> desiredTids) throws IOException {
     WalEdits edits = new WalEdits();
     edits.edits = new ArrayList<ByteBuffer>();
     long size = 0l;
@@ -283,9 +301,6 @@ public class AccumuloReplicaSystem implements ReplicaSystem {
     LogFileKey key = new LogFileKey();
     LogFileValue value = new LogFileValue();
 
-    // Any tid for our table needs to be tracked
-    Set<Integer> desiredTids = new HashSet<>();
-
     while (size < sizeLimit) {
       try {
         key.readFields(wal);
@@ -303,6 +318,7 @@ public class AccumuloReplicaSystem implements ReplicaSystem {
 
       switch (key.event) {
         case DEFINE_TABLET:
+          // For new DEFINE_TABLETs, we also need to record the new tids we see
           if (target.getSourceTableId().equals(key.tablet.getTableId().toString())) {
             desiredTids.add(key.tid);
           }
@@ -349,7 +365,10 @@ public class AccumuloReplicaSystem implements ReplicaSystem {
       }
     }
 
-    log.debug("Removing {} mutations from WAL entry as they have already been replicated
to {}", value.mutations.size() - mutationsToSend, target.getPeerName());
+    int mutationsRemoved = value.mutations.size() - mutationsToSend;
+    if (mutationsRemoved > 0) {
+      log.debug("Removing {} mutations from WAL entry as they have already been replicated
to {}", mutationsRemoved, target.getPeerName());
+    }
 
     out.writeInt(mutationsToSend);
     for (Mutation m : value.mutations) {

http://git-wip-us.apache.org/repos/asf/accumulo/blob/59177233/server/tserver/src/test/java/org/apache/accumulo/tserver/replication/AccumuloReplicaSystemTest.java
----------------------------------------------------------------------
diff --git a/server/tserver/src/test/java/org/apache/accumulo/tserver/replication/AccumuloReplicaSystemTest.java
b/server/tserver/src/test/java/org/apache/accumulo/tserver/replication/AccumuloReplicaSystemTest.java
index 07d1201..85204e3 100644
--- a/server/tserver/src/test/java/org/apache/accumulo/tserver/replication/AccumuloReplicaSystemTest.java
+++ b/server/tserver/src/test/java/org/apache/accumulo/tserver/replication/AccumuloReplicaSystemTest.java
@@ -24,6 +24,7 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Map;
 import java.util.UUID;
 
@@ -146,7 +147,7 @@ public class AccumuloReplicaSystemTest {
 
     Status status = Status.newBuilder().setBegin(0).setEnd(0).setInfiniteEnd(true).setClosed(false).build();
     DataInputStream dis = new DataInputStream(new ByteArrayInputStream(baos.toByteArray()));
-    WalReplication repl = ars.getEdits(dis, Long.MAX_VALUE, new ReplicationTarget("peer",
"1", "1"), status, new Path("/accumulo/wals/tserver+port/wal"));
+    WalReplication repl = ars.getEdits(dis, Long.MAX_VALUE, new ReplicationTarget("peer",
"1", "1"), status, new Path("/accumulo/wals/tserver+port/wal"), new HashSet<Integer>());
 
     // We stopped because we got to the end of the file
     Assert.assertEquals(9, repl.entriesConsumed);
@@ -253,7 +254,7 @@ public class AccumuloReplicaSystemTest {
     // If it were still open, more data could be appended that we need to process
     Status status = Status.newBuilder().setBegin(0).setEnd(0).setInfiniteEnd(true).setClosed(true).build();
     DataInputStream dis = new DataInputStream(new ByteArrayInputStream(baos.toByteArray()));
-    WalReplication repl = ars.getEdits(dis, Long.MAX_VALUE, new ReplicationTarget("peer",
"1", "1"), status, new Path("/accumulo/wals/tserver+port/wal"));
+    WalReplication repl = ars.getEdits(dis, Long.MAX_VALUE, new ReplicationTarget("peer",
"1", "1"), status, new Path("/accumulo/wals/tserver+port/wal"), new HashSet<Integer>());
 
     // We stopped because we got to the end of the file
     Assert.assertEquals(Long.MAX_VALUE, repl.entriesConsumed);
@@ -318,7 +319,7 @@ public class AccumuloReplicaSystemTest {
     // If it were still open, more data could be appended that we need to process
     Status status = Status.newBuilder().setBegin(100).setEnd(0).setInfiniteEnd(true).setClosed(true).build();
     DataInputStream dis = new DataInputStream(new ByteArrayInputStream(new byte[0]));
-    WalReplication repl = ars.getEdits(dis, Long.MAX_VALUE, new ReplicationTarget("peer",
"1", "1"), status, new Path("/accumulo/wals/tserver+port/wal"));
+    WalReplication repl = ars.getEdits(dis, Long.MAX_VALUE, new ReplicationTarget("peer",
"1", "1"), status, new Path("/accumulo/wals/tserver+port/wal"), new HashSet<Integer>());
 
     // We stopped because we got to the end of the file
     Assert.assertEquals(Long.MAX_VALUE, repl.entriesConsumed);
@@ -340,7 +341,7 @@ public class AccumuloReplicaSystemTest {
     // If it were still open, more data could be appended that we need to process
     Status status = Status.newBuilder().setBegin(100).setEnd(0).setInfiniteEnd(true).setClosed(false).build();
     DataInputStream dis = new DataInputStream(new ByteArrayInputStream(new byte[0]));
-    WalReplication repl = ars.getEdits(dis, Long.MAX_VALUE, new ReplicationTarget("peer",
"1", "1"), status, new Path("/accumulo/wals/tserver+port/wal"));
+    WalReplication repl = ars.getEdits(dis, Long.MAX_VALUE, new ReplicationTarget("peer",
"1", "1"), status, new Path("/accumulo/wals/tserver+port/wal"), new HashSet<Integer>());
 
     // We stopped because we got to the end of the file
     Assert.assertEquals(0, repl.entriesConsumed);
@@ -348,4 +349,77 @@ public class AccumuloReplicaSystemTest {
     Assert.assertEquals(0, repl.sizeInRecords);
     Assert.assertEquals(0, repl.sizeInBytes);
   }
+
+  @Test
+  public void restartInFileKnowsAboutPreviousTableDefines() throws Exception {
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    DataOutputStream dos = new DataOutputStream(baos);
+
+    LogFileKey key = new LogFileKey();
+    LogFileValue value = new LogFileValue();
+
+    // What is seq used for?
+    key.seq = 1l;
+
+    /*
+     * Disclaimer: the following series of LogFileKey and LogFileValue pairs have *no* bearing
whatsoever in reality regarding what these entries would actually
+     * look like in a WAL. They are solely for testing that each LogEvents is handled, order
is not important.
+     */
+    key.event = LogEvents.DEFINE_TABLET;
+    key.tablet = new KeyExtent(new Text("1"), null, null);
+    key.tid = 1;
+
+    key.write(dos);
+    value.write(dos);
+
+    key.tablet = null;
+    key.event = LogEvents.MUTATION;
+    key.filename = "/accumulo/wals/tserver+port/" + UUID.randomUUID();
+    value.mutations = Arrays.<Mutation> asList(new ServerMutation(new Text("row")));
+
+    key.write(dos);
+    value.write(dos);
+
+    key.tablet = null;
+    key.event = LogEvents.MUTATION;
+    key.tid = 1;
+    key.filename = "/accumulo/wals/tserver+port/" + UUID.randomUUID();
+    value.mutations = Arrays.<Mutation> asList(new ServerMutation(new Text("row")));
+
+    key.write(dos);
+    value.write(dos);
+
+    dos.close();
+
+    Map<String,String> confMap = new HashMap<>();
+    confMap.put(Property.REPLICATION_NAME.getKey(), "source");
+    AccumuloConfiguration conf = new ConfigurationCopy(confMap);
+
+    AccumuloReplicaSystem ars = new AccumuloReplicaSystem();
+    ars.setConf(conf);
+
+    Status status = Status.newBuilder().setBegin(0).setEnd(0).setInfiniteEnd(true).setClosed(false).build();
+    DataInputStream dis = new DataInputStream(new ByteArrayInputStream(baos.toByteArray()));
+
+    // Only consume the first mutation, not the second
+    WalReplication repl = ars.getWalEdits(new ReplicationTarget("peer", "1", "1"), dis, new
Path("/accumulo/wals/tserver+port/wal"), status, 1);
+
+    // We stopped because we got to the end of the file
+    Assert.assertEquals(2, repl.entriesConsumed);
+    Assert.assertEquals(1, repl.walEdits.getEditsSize());
+    Assert.assertEquals(1, repl.sizeInRecords);
+    Assert.assertNotEquals(0, repl.sizeInBytes);
+
+    status = Status.newBuilder(status).setBegin(2).build();
+    dis = new DataInputStream(new ByteArrayInputStream(baos.toByteArray()));
+
+    // Consume the rest of the mutations
+    repl = ars.getWalEdits(new ReplicationTarget("peer", "1", "1"), dis, new Path("/accumulo/wals/tserver+port/wal"),
status, 1);
+
+    // We stopped because we got to the end of the file
+    Assert.assertEquals(1, repl.entriesConsumed);
+    Assert.assertEquals(1, repl.walEdits.getEditsSize());
+    Assert.assertEquals(1, repl.sizeInRecords);
+    Assert.assertNotEquals(0, repl.sizeInBytes);
+  }
 }


Mime
View raw message