accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From els...@apache.org
Subject [33/50] [abbrv] git commit: ACCUMULO-2583 Get compilation working with slf4j dependencies. Lift the retry count into a Property.
Date Fri, 09 May 2014 15:29:11 GMT
ACCUMULO-2583 Get compilation working with slf4j dependencies. Lift the retry count into a
Property.

Left a todo for myself as well.


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

Branch: refs/heads/ACCUMULO-378
Commit: c7b6062c315ecde8321f83e50afa757003c24a97
Parents: baa7a4f
Author: Josh Elser <elserj@apache.org>
Authored: Wed May 7 00:55:15 2014 -0400
Committer: Josh Elser <elserj@apache.org>
Committed: Wed May 7 00:55:15 2014 -0400

----------------------------------------------------------------------
 .../org/apache/accumulo/core/conf/Property.java |   2 +
 .../core/replication/AccumuloReplicaSystem.java | 124 -----------------
 server/base/pom.xml                             |  18 ++-
 .../replication/AccumuloReplicaSystem.java      | 137 +++++++++++++++++++
 server/tserver/pom.xml                          |  18 ++-
 5 files changed, 155 insertions(+), 144 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo/blob/c7b6062c/core/src/main/java/org/apache/accumulo/core/conf/Property.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/conf/Property.java b/core/src/main/java/org/apache/accumulo/core/conf/Property.java
index f31cd46..6d3b00c 100644
--- a/core/src/main/java/org/apache/accumulo/core/conf/Property.java
+++ b/core/src/main/java/org/apache/accumulo/core/conf/Property.java
@@ -462,6 +462,8 @@ public enum Property {
   REPLICATION_WORKER_THREADS("replication.worker.threads", "4", PropertyType.COUNT, "Size
of the threadpool that each tabletserver devotes to replicating data"),
   @Experimental
   REPLICATION_RECEIPT_SERVICE_PORT("replication.receipt.service.port", "10001", PropertyType.PORT,
"Listen port used by thrift service in tserver listening for replication"),
+  @Experimental
+  REPLICATION_WORK_ATTEMPTS("replication.work.attempts", "10", PropertyType.COUNT, "Number
of attempts to try to replicate some data before giving up and letting it naturally be retried
later"),
 
   ;
 

http://git-wip-us.apache.org/repos/asf/accumulo/blob/c7b6062c/core/src/test/java/org/apache/accumulo/core/replication/AccumuloReplicaSystem.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/accumulo/core/replication/AccumuloReplicaSystem.java
b/core/src/test/java/org/apache/accumulo/core/replication/AccumuloReplicaSystem.java
deleted file mode 100644
index ca66b93..0000000
--- a/core/src/test/java/org/apache/accumulo/core/replication/AccumuloReplicaSystem.java
+++ /dev/null
@@ -1,124 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.accumulo.core.replication;
-
-import java.nio.ByteBuffer;
-
-import org.apache.accumulo.core.client.AccumuloException;
-import org.apache.accumulo.core.client.AccumuloSecurityException;
-import org.apache.accumulo.core.client.Instance;
-import org.apache.accumulo.core.client.ZooKeeperInstance;
-import org.apache.accumulo.core.client.impl.ClientExecReturn;
-import org.apache.accumulo.core.client.impl.ReplicationClient;
-import org.apache.accumulo.core.client.replication.ReplicaSystem;
-import org.apache.accumulo.core.replication.proto.Replication.Status;
-import org.apache.accumulo.core.replication.thrift.ReplicationCoordinator;
-import org.apache.accumulo.core.replication.thrift.ReplicationServicer;
-import org.apache.accumulo.core.replication.thrift.ReplicationServicer.Client;
-import org.apache.accumulo.core.util.ByteBufferUtil;
-import org.apache.accumulo.core.util.UtilWaitThread;
-import org.apache.hadoop.fs.Path;
-import org.apache.thrift.transport.TTransportException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.base.Preconditions;
-import com.google.protobuf.InvalidProtocolBufferException;
-
-/**
- * 
- */
-public class AccumuloReplicaSystem implements ReplicaSystem {
-  private static final Logger log = LoggerFactory.getLogger(AccumuloReplicaSystem.class);
-
-  private String instanceName, zookeepers;
-
-  @Override
-  public void configure(String configuration) {
-    Preconditions.checkNotNull(configuration);
-
-    int index = configuration.indexOf(',');
-    if (-1 == index) {
-      throw new IllegalArgumentException("Expected comma in configuration string");
-    }
-
-    instanceName = configuration.substring(0, index);
-    zookeepers = configuration.substring(index + 1);
-  }
-
-  @Override
-  public Status replicate(Path p, Status status, ReplicationTarget target) {
-    Instance peerInstance = getPeerInstance(target);
-    // Remote identifier is an integer (table id) in this case.
-    final int remoteTableId = Integer.parseInt(target.getRemoteIdentifier());
-
-    for (int i = 0; i < 10; i++) {
-      String peerTserver;
-      try {
-        // Ask the master on the remote what TServer we should talk with to replicate the
data
-        peerTserver = ReplicationClient.executeCoordinatorWithReturn(peerInstance, new ClientExecReturn<String,ReplicationCoordinator.Client>()
{
-  
-          @Override
-          public String execute(ReplicationCoordinator.Client client) throws Exception {
-            return client.getServicerAddress(remoteTableId);
-          }
-          
-        });
-      } catch (AccumuloException | AccumuloSecurityException e) {
-        // No progress is made
-        log.error("Could not connect to master at {}, cannot proceed with replication. Will
retry", target, e);
-        continue;
-      }
-  
-      if (null == peerTserver) {
-        // Something went wrong, and we didn't get a valid tserver from the remote for some
reason
-        log.warn("Did not receive tserver from master at {}, cannot proceed with replication.
Will retry.", target);
-        continue;
-      }
-  
-      // We have a tserver on the remote -- send the data its way.
-      ByteBuffer result;
-      try {
-        result = ReplicationClient.executeServicerWithReturn(peerInstance, peerTserver, new
ClientExecReturn<ByteBuffer,ReplicationServicer.Client>() {
-          @Override
-          public ByteBuffer execute(Client client) throws Exception {
-            return client.replicateLog(remoteTableId, null);
-          }
-        });
-
-        // We need to be able to parse the returned Status,
-        // if we can't, we don't know what the server actually parsed.
-        try {
-          return Status.parseFrom(ByteBufferUtil.toBytes(result));
-        } catch (InvalidProtocolBufferException e) {
-          log.error("Could not parse return Status from {}", peerTserver, e);
-          throw new RuntimeException("Could not parse returned Status from " + peerTserver,
e);
-        }
-      } catch (TTransportException | AccumuloException | AccumuloSecurityException e) {
-        log.warn("Could not connect to remote server {}, will retry", peerTserver, e);
-        UtilWaitThread.sleep(250);
-      }
-    }
-
-    // We made no status, punt on it for now, and let it re-queue itself for work
-    return status;
-  }
-
-  public Instance getPeerInstance(ReplicationTarget target) {
-    return new ZooKeeperInstance(instanceName, zookeepers);
-  }
-}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/c7b6062c/server/base/pom.xml
----------------------------------------------------------------------
diff --git a/server/base/pom.xml b/server/base/pom.xml
index f698621..672fe91 100644
--- a/server/base/pom.xml
+++ b/server/base/pom.xml
@@ -92,23 +92,21 @@
       <artifactId>zookeeper</artifactId>
     </dependency>
     <dependency>
-      <groupId>junit</groupId>
-      <artifactId>junit</artifactId>
-      <scope>test</scope>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-api</artifactId>
     </dependency>
     <dependency>
-      <groupId>org.easymock</groupId>
-      <artifactId>easymock</artifactId>
-      <scope>test</scope>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-log4j12</artifactId>
     </dependency>
     <dependency>
-      <groupId>org.slf4j</groupId>
-      <artifactId>slf4j-api</artifactId>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
       <scope>test</scope>
     </dependency>
     <dependency>
-      <groupId>org.slf4j</groupId>
-      <artifactId>slf4j-log4j12</artifactId>
+      <groupId>org.easymock</groupId>
+      <artifactId>easymock</artifactId>
       <scope>test</scope>
     </dependency>
   </dependencies>

http://git-wip-us.apache.org/repos/asf/accumulo/blob/c7b6062c/server/base/src/main/java/org/apache/accumulo/server/replication/AccumuloReplicaSystem.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/replication/AccumuloReplicaSystem.java
b/server/base/src/main/java/org/apache/accumulo/server/replication/AccumuloReplicaSystem.java
new file mode 100644
index 0000000..138eb71
--- /dev/null
+++ b/server/base/src/main/java/org/apache/accumulo/server/replication/AccumuloReplicaSystem.java
@@ -0,0 +1,137 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.accumulo.server.replication;
+
+import java.nio.ByteBuffer;
+
+import org.apache.accumulo.core.client.AccumuloException;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.Instance;
+import org.apache.accumulo.core.client.ZooKeeperInstance;
+import org.apache.accumulo.core.client.impl.ClientExecReturn;
+import org.apache.accumulo.core.client.impl.ReplicationClient;
+import org.apache.accumulo.core.client.impl.ServerConfigurationUtil;
+import org.apache.accumulo.core.client.replication.ReplicaSystem;
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.replication.ReplicationTarget;
+import org.apache.accumulo.core.replication.proto.Replication.Status;
+import org.apache.accumulo.core.replication.thrift.ReplicationCoordinator;
+import org.apache.accumulo.core.replication.thrift.ReplicationServicer;
+import org.apache.accumulo.core.replication.thrift.ReplicationServicer.Client;
+import org.apache.accumulo.core.util.ByteBufferUtil;
+import org.apache.accumulo.core.util.UtilWaitThread;
+import org.apache.accumulo.server.client.HdfsZooInstance;
+import org.apache.hadoop.fs.Path;
+import org.apache.thrift.transport.TTransportException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Preconditions;
+import com.google.protobuf.InvalidProtocolBufferException;
+
+/**
+ * 
+ */
+public class AccumuloReplicaSystem implements ReplicaSystem {
+  private static final Logger log = LoggerFactory.getLogger(AccumuloReplicaSystem.class);
+
+  private String instanceName, zookeepers;
+
+  @Override
+  public void configure(String configuration) {
+    Preconditions.checkNotNull(configuration);
+
+    int index = configuration.indexOf(',');
+    if (-1 == index) {
+      throw new IllegalArgumentException("Expected comma in configuration string");
+    }
+
+    instanceName = configuration.substring(0, index);
+    zookeepers = configuration.substring(index + 1);
+  }
+
+  @Override
+  public Status replicate(Path p, Status status, ReplicationTarget target) {
+    Instance localInstance = HdfsZooInstance.getInstance();
+    AccumuloConfiguration localConf = ServerConfigurationUtil.getConfiguration(localInstance);
+    
+    Instance peerInstance = getPeerInstance(target);
+    // Remote identifier is an integer (table id) in this case.
+    final int remoteTableId = Integer.parseInt(target.getRemoteIdentifier());
+
+    // Attempt the replication of this status a number of times before giving up and
+    // trying to replicate it again later some other time.
+    for (int i = 0; i < localConf.getCount(Property.REPLICATION_WORK_ATTEMPTS); i++) {
+      String peerTserver;
+      try {
+        // Ask the master on the remote what TServer we should talk with to replicate the
data
+        peerTserver = ReplicationClient.executeCoordinatorWithReturn(peerInstance, new ClientExecReturn<String,ReplicationCoordinator.Client>()
{
+  
+          @Override
+          public String execute(ReplicationCoordinator.Client client) throws Exception {
+            return client.getServicerAddress(remoteTableId);
+          }
+          
+        });
+      } catch (AccumuloException | AccumuloSecurityException e) {
+        // No progress is made
+        log.error("Could not connect to master at {}, cannot proceed with replication. Will
retry", target, e);
+        continue;
+      }
+  
+      if (null == peerTserver) {
+        // Something went wrong, and we didn't get a valid tserver from the remote for some
reason
+        log.warn("Did not receive tserver from master at {}, cannot proceed with replication.
Will retry.", target);
+        continue;
+      }
+  
+      // We have a tserver on the remote -- send the data its way.
+      ByteBuffer result;
+      //TODO should chunk up the given file into some configurable sizes instead of just
sending the entire file all at once
+      //     configuration should probably just be size based.
+      try {
+        result = ReplicationClient.executeServicerWithReturn(peerInstance, peerTserver, new
ClientExecReturn<ByteBuffer,ReplicationServicer.Client>() {
+          @Override
+          public ByteBuffer execute(Client client) throws Exception {
+            //TODO This needs to actually send the appropriate data, and choose replicateLog
or replicateKeyValues
+            return client.replicateLog(remoteTableId, null);
+          }
+        });
+
+        // We need to be able to parse the returned Status,
+        // if we can't, we don't know what the server actually parsed.
+        try {
+          return Status.parseFrom(ByteBufferUtil.toBytes(result));
+        } catch (InvalidProtocolBufferException e) {
+          log.error("Could not parse return Status from {}", peerTserver, e);
+          throw new RuntimeException("Could not parse returned Status from " + peerTserver,
e);
+        }
+      } catch (TTransportException | AccumuloException | AccumuloSecurityException e) {
+        log.warn("Could not connect to remote server {}, will retry", peerTserver, e);
+        UtilWaitThread.sleep(250);
+      }
+    }
+
+    // We made no status, punt on it for now, and let it re-queue itself for work
+    return status;
+  }
+
+  public Instance getPeerInstance(ReplicationTarget target) {
+    return new ZooKeeperInstance(instanceName, zookeepers);
+  }
+}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/c7b6062c/server/tserver/pom.xml
----------------------------------------------------------------------
diff --git a/server/tserver/pom.xml b/server/tserver/pom.xml
index 100bc2d..e09f348 100644
--- a/server/tserver/pom.xml
+++ b/server/tserver/pom.xml
@@ -88,23 +88,21 @@
       <artifactId>zookeeper</artifactId>
     </dependency>
     <dependency>
-      <groupId>junit</groupId>
-      <artifactId>junit</artifactId>
-      <scope>test</scope>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-api</artifactId>
     </dependency>
     <dependency>
-      <groupId>org.easymock</groupId>
-      <artifactId>easymock</artifactId>
-      <scope>test</scope>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-log4j12</artifactId>
     </dependency>
     <dependency>
-      <groupId>org.slf4j</groupId>
-      <artifactId>slf4j-api</artifactId>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
       <scope>test</scope>
     </dependency>
     <dependency>
-      <groupId>org.slf4j</groupId>
-      <artifactId>slf4j-log4j12</artifactId>
+      <groupId>org.easymock</groupId>
+      <artifactId>easymock</artifactId>
       <scope>test</scope>
     </dependency>
   </dependencies>


Mime
View raw message