accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From els...@apache.org
Subject [5/6] git commit: ACCUMULO-2925 Add test to ensure that replicationSource(s) are preserved before replay
Date Fri, 20 Jun 2014 01:25:24 GMT
ACCUMULO-2925 Add test to ensure that replicationSource(s) are preserved before replay


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

Branch: refs/heads/master
Commit: 06760572e9325ba1f622a998085853a0098de783
Parents: cc7f91b
Author: Josh Elser <elserj@apache.org>
Authored: Thu Jun 19 17:41:27 2014 -0700
Committer: Josh Elser <elserj@apache.org>
Committed: Thu Jun 19 17:41:27 2014 -0700

----------------------------------------------------------------------
 .../BatchWriterReplicationReplayerTest.java     | 76 ++++++++++++++++++++
 1 file changed, 76 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo/blob/06760572/server/tserver/src/test/java/org/apache/accumulo/tserver/replication/BatchWriterReplicationReplayerTest.java
----------------------------------------------------------------------
diff --git a/server/tserver/src/test/java/org/apache/accumulo/tserver/replication/BatchWriterReplicationReplayerTest.java
b/server/tserver/src/test/java/org/apache/accumulo/tserver/replication/BatchWriterReplicationReplayerTest.java
index 1e70490..fee5dcd 100644
--- a/server/tserver/src/test/java/org/apache/accumulo/tserver/replication/BatchWriterReplicationReplayerTest.java
+++ b/server/tserver/src/test/java/org/apache/accumulo/tserver/replication/BatchWriterReplicationReplayerTest.java
@@ -116,4 +116,80 @@ public class BatchWriterReplicationReplayerTest {
     verify(conn, conf, bw);
   }
 
+  @Test
+  public void replicationSourcesArePreserved() throws Exception {
+    final BatchWriterReplicationReplayer replayer = new BatchWriterReplicationReplayer();
+    final String tableName = "foo";
+    final long systemTimestamp = 1000;
+    final String peerName = "peer";
+    final BatchWriterConfig bwCfg = new BatchWriterConfig();
+    bwCfg.setMaxMemory(1l);
+
+    Connector conn = createMock(Connector.class);
+    AccumuloConfiguration conf = createMock(AccumuloConfiguration.class);
+    BatchWriter bw = createMock(BatchWriter.class);
+
+    LogFileKey key = new LogFileKey();
+    key.event = LogEvents.MANY_MUTATIONS;
+    key.seq = 1;
+    key.tid = 1;
+
+    WalEdits edits = new WalEdits();
+
+    // Make a mutation without timestamps
+    Mutation m = new Mutation("row");
+    m.put("cf", "cq1", "value");
+    m.put("cf", "cq2", "value");
+    m.put("cf", "cq3", "value");
+    m.put("cf", "cq4", "value");
+    m.put("cf", "cq5", "value");
+
+    // This Mutation "came" from a system called "peer"
+    m.addReplicationSource(peerName);
+
+    // Make it a TMutation
+    TMutation tMutation = m.toThrift();
+
+    // And then make a ServerMutation from the TMutation, adding in our systemTimestamp
+    ServerMutation sMutation = new ServerMutation(tMutation);
+    sMutation.setSystemTimestamp(systemTimestamp);
+
+    // Serialize the ServerMutation (what AccumuloReplicaSystem will be doing)
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    DataOutputStream out = new DataOutputStream(baos);
+
+    key.write(out);
+    out.writeInt(1);
+    sMutation.write(out);
+
+    out.close();
+
+    // Add it to our "input" to BatchWriterReplicationReplayer
+    edits.addToEdits(ByteBuffer.wrap(baos.toByteArray()));
+
+    Mutation expectedMutation = new Mutation("row");
+    expectedMutation.put("cf", "cq1", sMutation.getSystemTimestamp(), "value");
+    expectedMutation.put("cf", "cq2", sMutation.getSystemTimestamp(), "value");
+    expectedMutation.put("cf", "cq3", sMutation.getSystemTimestamp(), "value");
+    expectedMutation.put("cf", "cq4", sMutation.getSystemTimestamp(), "value");
+    expectedMutation.put("cf", "cq5", sMutation.getSystemTimestamp(), "value");
+
+    // We expect our peer name to be preserved in the mutation that gets written
+    expectedMutation.addReplicationSource(peerName);
+
+    expect(conf.getMemoryInBytes(Property.TSERV_REPLICATION_BW_REPLAYER_MEMORY)).andReturn(bwCfg.getMaxMemory());
+    expect(conn.createBatchWriter(tableName, bwCfg)).andReturn(bw);
+
+    bw.addMutations(Lists.newArrayList(expectedMutation));
+    expectLastCall().once();
+
+    bw.close();
+    expectLastCall().once();
+
+    replay(conn, conf, bw);
+
+    replayer.replicateLog(conn, conf, tableName, edits);
+
+    verify(conn, conf, bw);
+  }
 }


Mime
View raw message