accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From els...@apache.org
Subject [28/50] [abbrv] ACCUMULO-2583 First stab at setting up the actual wire transfer to the replication peer
Date Fri, 09 May 2014 15:29:06 GMT
http://git-wip-us.apache.org/repos/asf/accumulo/blob/baa7a4f7/server/master/src/main/java/org/apache/accumulo/master/replication/MasterReplicationCoordinator.java
----------------------------------------------------------------------
diff --git a/server/master/src/main/java/org/apache/accumulo/master/replication/MasterReplicationCoordinator.java
b/server/master/src/main/java/org/apache/accumulo/master/replication/MasterReplicationCoordinator.java
index 0497892..337a592 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/replication/MasterReplicationCoordinator.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/replication/MasterReplicationCoordinator.java
@@ -23,7 +23,7 @@ import java.util.Set;
 import org.apache.accumulo.core.conf.AccumuloConfiguration;
 import org.apache.accumulo.core.conf.Property;
 import org.apache.accumulo.core.replication.thrift.NoServersAvailableException;
-import org.apache.accumulo.core.replication.thrift.RemoteReplicationCoordinator;
+import org.apache.accumulo.core.replication.thrift.ReplicationCoordinator;
 import org.apache.accumulo.master.Master;
 import org.apache.accumulo.server.master.state.TServerInstance;
 import org.apache.thrift.TException;
@@ -33,7 +33,7 @@ import com.google.common.base.Preconditions;
 /**
  * Choose a tserver to service a replication task
  */
-public class MasterReplicationCoordinator implements RemoteReplicationCoordinator.Iface {
+public class MasterReplicationCoordinator implements ReplicationCoordinator.Iface {
 
   public static enum NoServersAvailable {
     NO_ONLINE_SERVERS

http://git-wip-us.apache.org/repos/asf/accumulo/blob/baa7a4f7/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 370fcfd..699381f 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
@@ -16,7 +16,6 @@
  */
 package org.apache.accumulo.master.replication;
 
-import java.nio.charset.StandardCharsets;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.Iterator;
@@ -38,6 +37,7 @@ import org.apache.accumulo.core.replication.proto.Replication.Status;
 import org.apache.accumulo.core.zookeeper.ZooUtil;
 import org.apache.accumulo.master.Master;
 import org.apache.accumulo.server.replication.ReplicationTable;
+import org.apache.accumulo.server.replication.ReplicationWorkAssignerHelper;
 import org.apache.accumulo.server.zookeeper.DistributedWorkQueue;
 import org.apache.accumulo.server.zookeeper.ZooCache;
 import org.apache.hadoop.fs.Path;
@@ -215,8 +215,8 @@ 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;
+          WorkSection.getTarget(entry.getKey(), buffer);
+          String key = ReplicationWorkAssignerHelper.getQueueKey(filename, buffer.toString());
 
           // And, we haven't already queued this file up for work already
           if (!queuedWork.contains(key)) {
@@ -234,10 +234,10 @@ public class ReplicationWorkAssigner implements Runnable {
   /**
    * Distribute the work for the given path with filename
    * 
+   * @param key
+   *          Unique key to identify this work in the queue
    * @param path
    *          Full path to a file
-   * @param filename
-   *          Filename
    */
   protected void queueWork(String key, String path) {
     try {

http://git-wip-us.apache.org/repos/asf/accumulo/blob/baa7a4f7/server/master/src/main/java/org/apache/accumulo/master/replication/WorkMaker.java
----------------------------------------------------------------------
diff --git a/server/master/src/main/java/org/apache/accumulo/master/replication/WorkMaker.java
b/server/master/src/main/java/org/apache/accumulo/master/replication/WorkMaker.java
index 40e9838..cfc756c 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/replication/WorkMaker.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/replication/WorkMaker.java
@@ -142,7 +142,7 @@ public class WorkMaker {
         buffer.reset();
 
         // Set up the writable
-        target.setRemoteName(entry.getKey());
+        target.setPeerName(entry.getKey());
         target.setRemoteIdentifier(entry.getValue());
         target.write(buffer);
 

http://git-wip-us.apache.org/repos/asf/accumulo/blob/baa7a4f7/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 89a1ddd..16e6757 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
@@ -22,7 +22,6 @@ 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;
@@ -76,7 +75,7 @@ public class ReplicationWorkAssignerTest {
   @Test
   public void workQueuedUsingFileName() throws Exception {
     ReplicationTarget target = new ReplicationTarget("cluster1", "table1"); 
-    Text serializedTarget = ReplicationTarget.toText(target);
+    Text serializedTarget = target.toText();
 
     DistributedWorkQueue workQueue = createMock(DistributedWorkQueue.class);
     Set<String> queuedWork = new HashSet<>();
@@ -118,7 +117,7 @@ public class ReplicationWorkAssignerTest {
   @Test
   public void createWorkForFilesNeedingIt() throws Exception {
     ReplicationTarget target1 = new ReplicationTarget("cluster1", "table1"), target2 = new
ReplicationTarget("cluster1", "table2"); 
-    Text serializedTarget1 = ReplicationTarget.toText(target1), serializedTarget2 = ReplicationTarget.toText(target2);
+    Text serializedTarget1 = target1.toText(), serializedTarget2 = target2.toText();
 
     MockInstance inst = new MockInstance(test.getMethodName());
     Credentials creds = new Credentials("root", new PasswordToken(""));
@@ -191,7 +190,7 @@ public class ReplicationWorkAssignerTest {
   @Test
   public void doNotCreateWorkForFilesNotNeedingIt() throws Exception {
     ReplicationTarget target1 = new ReplicationTarget("cluster1", "table1"), target2 = new
ReplicationTarget("cluster1", "table2"); 
-    Text serializedTarget1 = ReplicationTarget.toText(target1), serializedTarget2 = ReplicationTarget.toText(target2);
+    Text serializedTarget1 = target1.toText(), serializedTarget2 = target2.toText();
 
     MockInstance inst = new MockInstance(test.getMethodName());
     Credentials creds = new Credentials("root", new PasswordToken(""));
@@ -263,7 +262,7 @@ public class ReplicationWorkAssignerTest {
     assigner.setQueuedWork(queuedWork);
 
     ReplicationTarget target = new ReplicationTarget("cluster1", "table1"); 
-    Text serializedTarget = ReplicationTarget.toText(target);
+    Text serializedTarget = target.toText();
 
     queuedWork.add("wal1|" + serializedTarget.toString());
 

http://git-wip-us.apache.org/repos/asf/accumulo/blob/baa7a4f7/server/tserver/src/main/java/org/apache/accumulo/tserver/replication/ReplicationProcessor.java
----------------------------------------------------------------------
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/replication/ReplicationProcessor.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/replication/ReplicationProcessor.java
index 82d7ae9..c0aa31d 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/replication/ReplicationProcessor.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/replication/ReplicationProcessor.java
@@ -16,22 +16,146 @@
  */
 package org.apache.accumulo.tserver.replication;
 
+import java.io.IOException;
+import java.util.Map;
+
+import org.apache.accumulo.core.client.AccumuloException;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+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.TableNotFoundException;
+import org.apache.accumulo.core.client.replication.ReplicaSystem;
+import org.apache.accumulo.core.client.replication.ReplicaSystemFactory;
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.protobuf.ProtobufUtil;
+import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
+import org.apache.accumulo.core.replication.ReplicationTarget;
+import org.apache.accumulo.core.replication.StatusUtil;
+import org.apache.accumulo.core.replication.proto.Replication.Status;
+import org.apache.accumulo.core.security.Credentials;
+import org.apache.accumulo.server.fs.VolumeManager;
+import org.apache.accumulo.server.replication.ReplicationTable;
+import org.apache.accumulo.server.replication.ReplicationWorkAssignerHelper;
+import org.apache.accumulo.server.security.SystemCredentials;
 import org.apache.accumulo.server.zookeeper.DistributedWorkQueue.Processor;
+import org.apache.hadoop.fs.Path;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Iterables;
+import com.google.protobuf.InvalidProtocolBufferException;
 
 /**
- * 
+ * Transmit the given data to a peer
  */
 public class ReplicationProcessor implements Processor {
+  private static final Logger log = LoggerFactory.getLogger(ReplicationProcessor.class);
+
+  private Instance inst;
+  private AccumuloConfiguration conf;
+  private VolumeManager fs;
+  private Credentials creds;
+
+  public ReplicationProcessor(Instance inst, AccumuloConfiguration conf, VolumeManager fs)
{
+    this.conf = conf;
+    this.fs = fs;
+    creds = SystemCredentials.get();
+  }
 
   @Override
   public ReplicationProcessor newProcessor() {
-    return new ReplicationProcessor();
+    return new ReplicationProcessor(inst, conf, fs);
   }
 
   @Override
   public void process(String workID, byte[] data) {
-    // TODO Auto-generated method stub
+    ReplicationTarget target = ReplicationWorkAssignerHelper.fromQueueKey(workID).getValue();
+    String file = new String(data);
+
+    // Find the configured replication peer so we know how to replicate to it
+    Map<String,String> configuredPeers = conf.getAllPropertiesWithPrefix(Property.REPLICATION_PEERS);
+    String peerType = configuredPeers.get(target.getPeerName());
+    if (null == peerType) {
+      String msg = "Cannot process replication for unknown peer: " +  target.getPeerName();
+      log.warn(msg);
+      throw new IllegalArgumentException(msg);
+    }
+
+    // Get the peer that we're replicating to
+    ReplicaSystem replica = ReplicaSystemFactory.get(peerType);
+    Status status;
+    try {
+      status = getStatus(file, target);
+    } catch (TableNotFoundException | AccumuloException | AccumuloSecurityException e) {
+      log.error("Could not look for necessary replication record", e);
+      throw new IllegalStateException("Could not look for replication record", e);
+    } catch (InvalidProtocolBufferException e) {
+      log.error("Could not deserialize Status from Work section for {} and ", file, target);
+      throw new RuntimeException("Could not parse Status for work record", e);
+    }
 
+    // We don't need to do anything (shouldn't have gotten this work record in the first
place)
+    if (!StatusUtil.isWorkRequired(status)) {
+      log.info("Received work request for {} and {}, but it does not need replication. Ignoring...",
file, target);
+      return;
+    }
+
+    // Sanity check that nothing bad happened and our replication source still exists
+    Path filePath = new Path(file);
+    try {
+      if (!fs.exists(filePath)) {
+        log.warn("Received work request for {} and {}, but the file doesn't exist", filePath,
target);
+        return;
+      }
+    } catch (IOException e) {
+      log.error("Could not determine if file exists {}", filePath, e);
+      throw new RuntimeException(e);
+    }
+
+    // Replicate that sucker
+    Status replicatedStatus = replica.replicate(filePath, status, target);
+
+    // If we got a different status
+    if (!replicatedStatus.equals(status)) {
+      // We actually did some work!
+      recordNewStatus(filePath, replicatedStatus, target);
+    }
+
+    // otherwise, we didn't actually replicate because there was error sending the data
+    // we can just not record any updates, and it will be picked up again by the work assigner
   }
 
+  public Status getStatus(String file, ReplicationTarget target) throws TableNotFoundException,
AccumuloException, AccumuloSecurityException, InvalidProtocolBufferException {
+    Scanner s = ReplicationTable.getScanner(inst.getConnector(creds.getPrincipal(), creds.getToken()));
+    s.setRange(Range.exact(file));
+    s.fetchColumn(WorkSection.NAME, target.toText());
+    
+    return Status.parseFrom(Iterables.getOnlyElement(s).getValue().get());
+  }
+
+  /**
+   * Record the updated Status for this file and target
+   * @param filePath Path to file being replicated
+   * @param status Updated Status after replication
+   * @param target Peer that was replicated to
+   */
+  public void recordNewStatus(Path filePath, Status status, ReplicationTarget target) {
+    try {
+      Connector conn = inst.getConnector(creds.getPrincipal(), creds.getToken());
+      BatchWriter bw = ReplicationTable.getBatchWriter(conn);
+      Mutation m = new Mutation(filePath.toString());
+      WorkSection.add(m, target.toText(), ProtobufUtil.toValue(status));
+      bw.addMutation(m);
+      bw.close();
+    } catch (AccumuloException | AccumuloSecurityException | TableNotFoundException e) {
+      log.error("Error recording updated Status for {}", filePath, e);
+      throw new RuntimeException(e);
+    }
+    
+  }
 }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/baa7a4f7/server/tserver/src/main/java/org/apache/accumulo/tserver/replication/ReplicationWorker.java
----------------------------------------------------------------------
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/replication/ReplicationWorker.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/replication/ReplicationWorker.java
index 627e51a..e00f343 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/replication/ReplicationWorker.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/replication/ReplicationWorker.java
@@ -49,7 +49,7 @@ public class ReplicationWorker implements Runnable {
   @Override
   public void run() {
     try {
-      new DistributedWorkQueue(ZooUtil.getRoot(inst) + Constants.ZREPLICATION, conf).startProcessing(new
ReplicationProcessor(), executor);
+      new DistributedWorkQueue(ZooUtil.getRoot(inst) + Constants.ZREPLICATION, conf).startProcessing(new
ReplicationProcessor(inst, conf, fs), executor);
     } catch (KeeperException | InterruptedException e) {
       throw new RuntimeException(e);
     }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/baa7a4f7/test/src/test/java/org/apache/accumulo/test/replication/ReplicationWithGCIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/replication/ReplicationWithGCIT.java
b/test/src/test/java/org/apache/accumulo/test/replication/ReplicationWithGCIT.java
index 33da8a0..92f6fb3 100644
--- a/test/src/test/java/org/apache/accumulo/test/replication/ReplicationWithGCIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/replication/ReplicationWithGCIT.java
@@ -385,7 +385,7 @@ public class ReplicationWithGCIT extends ConfigurableMacIT {
         s = ReplicationTable.getScanner(conn);
         WorkSection.limit(s);
         Entry<Key,Value> e = Iterables.getOnlyElement(s);
-        Text expectedColqual = ReplicationTarget.toText(new ReplicationTarget("cluster1",
"4"));
+        Text expectedColqual = new ReplicationTarget("cluster1", "4").toText();
         Assert.assertEquals(expectedColqual, e.getKey().getColumnQualifier());
         notFound = false;
       } catch (NoSuchElementException e) {

http://git-wip-us.apache.org/repos/asf/accumulo/blob/baa7a4f7/test/src/test/java/org/apache/accumulo/test/replication/ReplicationWithMakerTest.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/replication/ReplicationWithMakerTest.java
b/test/src/test/java/org/apache/accumulo/test/replication/ReplicationWithMakerTest.java
index 613a01b..defff65 100644
--- a/test/src/test/java/org/apache/accumulo/test/replication/ReplicationWithMakerTest.java
+++ b/test/src/test/java/org/apache/accumulo/test/replication/ReplicationWithMakerTest.java
@@ -303,7 +303,7 @@ public class ReplicationWithMakerTest extends ConfigurableMacIT {
       WorkSection.limit(s);
       try {
         Entry<Key,Value> e = Iterables.getOnlyElement(s);
-        Text expectedColqual = ReplicationTarget.toText(new ReplicationTarget("cluster1",
"4"));
+        Text expectedColqual = new ReplicationTarget("cluster1", "4").toText();
         Assert.assertEquals(expectedColqual, e.getKey().getColumnQualifier());
         notFound = false;
       } catch (NoSuchElementException e) {} catch (IllegalArgumentException e) {


Mime
View raw message