You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by el...@apache.org on 2014/05/09 17:29:06 UTC
[28/50] [abbrv] ACCUMULO-2583 First stab at setting up the actual
wire transfer to the replication peer
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) {