You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by ct...@apache.org on 2014/11/01 05:57:41 UTC
[47/50] [abbrv] Merge branch '1.6'
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/base/src/main/java/org/apache/accumulo/server/util/ChangeSecret.java
----------------------------------------------------------------------
diff --cc server/base/src/main/java/org/apache/accumulo/server/util/ChangeSecret.java
index 7f2048b,7a1e1ba..c93f680
--- a/server/base/src/main/java/org/apache/accumulo/server/util/ChangeSecret.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/ChangeSecret.java
@@@ -16,8 -16,9 +16,9 @@@
*/
package org.apache.accumulo.server.util;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+
import java.io.IOException;
- import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/base/src/main/java/org/apache/accumulo/server/util/CleanZookeeper.java
----------------------------------------------------------------------
diff --cc server/base/src/main/java/org/apache/accumulo/server/util/CleanZookeeper.java
index 3543a16,809a7c8..0652cf1
--- a/server/base/src/main/java/org/apache/accumulo/server/util/CleanZookeeper.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/CleanZookeeper.java
@@@ -16,9 -16,10 +16,9 @@@
*/
package org.apache.accumulo.server.util;
- import java.io.IOException;
- import java.nio.charset.StandardCharsets;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+ import java.io.IOException;
-
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.cli.Help;
import org.apache.accumulo.fate.zookeeper.IZooReaderWriter;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/base/src/main/java/org/apache/accumulo/server/util/DeleteZooInstance.java
----------------------------------------------------------------------
diff --cc server/base/src/main/java/org/apache/accumulo/server/util/DeleteZooInstance.java
index f728fec,d40830f..0793c6b
--- a/server/base/src/main/java/org/apache/accumulo/server/util/DeleteZooInstance.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/DeleteZooInstance.java
@@@ -16,7 -16,8 +16,8 @@@
*/
package org.apache.accumulo.server.util;
- import java.nio.charset.StandardCharsets;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+
import java.util.HashSet;
import java.util.Set;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/base/src/main/java/org/apache/accumulo/server/util/DumpZookeeper.java
----------------------------------------------------------------------
diff --cc server/base/src/main/java/org/apache/accumulo/server/util/DumpZookeeper.java
index 565eaba,43348df..478b3b5
--- a/server/base/src/main/java/org/apache/accumulo/server/util/DumpZookeeper.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/DumpZookeeper.java
@@@ -16,10 -16,11 +16,10 @@@
*/
package org.apache.accumulo.server.util;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+
import java.io.PrintStream;
import java.io.UnsupportedEncodingException;
- import java.nio.charset.StandardCharsets;
--
import org.apache.accumulo.core.cli.Help;
import org.apache.accumulo.core.util.Base64;
import org.apache.accumulo.fate.zookeeper.IZooReaderWriter;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/base/src/main/java/org/apache/accumulo/server/util/ListInstances.java
----------------------------------------------------------------------
diff --cc server/base/src/main/java/org/apache/accumulo/server/util/ListInstances.java
index f73747d,be85b1b..8550d0b
--- a/server/base/src/main/java/org/apache/accumulo/server/util/ListInstances.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/ListInstances.java
@@@ -16,7 -16,8 +16,8 @@@
*/
package org.apache.accumulo.server.util;
- import java.nio.charset.StandardCharsets;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+
import java.util.Formattable;
import java.util.Formatter;
import java.util.List;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/base/src/main/java/org/apache/accumulo/server/util/MasterMetadataUtil.java
----------------------------------------------------------------------
diff --cc server/base/src/main/java/org/apache/accumulo/server/util/MasterMetadataUtil.java
index 05f28d7,e936b97..01eb477
--- a/server/base/src/main/java/org/apache/accumulo/server/util/MasterMetadataUtil.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/MasterMetadataUtil.java
@@@ -16,8 -16,7 +16,9 @@@
*/
package org.apache.accumulo.server.util;
++import static java.nio.charset.StandardCharsets.UTF_8;
++
import java.io.IOException;
- import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
@@@ -70,8 -69,8 +71,8 @@@ public class MasterMetadataUtil
Map<FileRef,Long> bulkLoadedFiles, Credentials credentials, String time, long lastFlushID, long lastCompactID, ZooLock zooLock) {
Mutation m = extent.getPrevRowUpdateMutation();
- TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN.put(m, new Value(path.getBytes(StandardCharsets.UTF_8)));
- TabletsSection.ServerColumnFamily.TIME_COLUMN.put(m, new Value(time.getBytes(StandardCharsets.UTF_8)));
- TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN.put(m, new Value(path.getBytes()));
- TabletsSection.ServerColumnFamily.TIME_COLUMN.put(m, new Value(time.getBytes()));
++ TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN.put(m, new Value(path.getBytes(UTF_8)));
++ TabletsSection.ServerColumnFamily.TIME_COLUMN.put(m, new Value(time.getBytes(UTF_8)));
if (lastFlushID > 0)
TabletsSection.ServerColumnFamily.FLUSH_COLUMN.put(m, new Value(("" + lastFlushID).getBytes()));
if (lastCompactID > 0)
@@@ -104,7 -103,7 +105,7 @@@
throw new IllegalArgumentException("Metadata entry does not have split ratio (" + metadataEntry + ")");
}
- double splitRatio = Double.parseDouble(new String(columns.get(TabletsSection.TabletColumnFamily.SPLIT_RATIO_COLUMN).get(), StandardCharsets.UTF_8));
- double splitRatio = Double.parseDouble(new String(columns.get(TabletsSection.TabletColumnFamily.SPLIT_RATIO_COLUMN).get()));
++ double splitRatio = Double.parseDouble(new String(columns.get(TabletsSection.TabletColumnFamily.SPLIT_RATIO_COLUMN).get(), UTF_8));
Value prevEndRowIBW = columns.get(TabletsSection.TabletColumnFamily.PREV_ROW_COLUMN);
@@@ -245,62 -244,41 +246,62 @@@
Set<FileRef> filesInUseByScans, String address, ZooLock zooLock, Set<String> unusedWalLogs, TServerInstance lastLocation, long flushId) {
if (extent.isRootTablet()) {
if (unusedWalLogs != null) {
- IZooReaderWriter zk = ZooReaderWriter.getInstance();
- // unusedWalLogs will contain the location/name of each log in a log set
- // the log set is stored under one of the log names, but not both
- // find the entry under one of the names and delete it.
- String root = MetadataTableUtil.getZookeeperLogLocation();
- boolean foundEntry = false;
- for (String entry : unusedWalLogs) {
- String[] parts = entry.split("/");
- String zpath = root + "/" + parts[parts.length - 1];
- while (true) {
- try {
- if (zk.exists(zpath)) {
- zk.recursiveDelete(zpath, NodeMissingPolicy.SKIP);
- foundEntry = true;
- }
- break;
- } catch (KeeperException e) {
- log.error(e, e);
- } catch (InterruptedException e) {
- log.error(e, e);
- }
- UtilWaitThread.sleep(1000);
+ updateRootTabletDataFile(extent, path, mergeFile, dfv, time, filesInUseByScans, address, zooLock, unusedWalLogs, lastLocation, flushId);
+ }
+
+ return;
+ }
+
+ Mutation m = getUpdateForTabletDataFile(extent, path, mergeFile, dfv, time, filesInUseByScans, address, zooLock, unusedWalLogs, lastLocation, flushId);
+
+ MetadataTableUtil.update(credentials, zooLock, m, extent);
+
+ }
+
+ /**
+ * Update the data file for the root tablet
+ */
+ protected static void updateRootTabletDataFile(KeyExtent extent, FileRef path, FileRef mergeFile, DataFileValue dfv, String time,
+ Set<FileRef> filesInUseByScans, String address, ZooLock zooLock, Set<String> unusedWalLogs, TServerInstance lastLocation, long flushId) {
+ IZooReaderWriter zk = ZooReaderWriter.getInstance();
+ // unusedWalLogs will contain the location/name of each log in a log set
+ // the log set is stored under one of the log names, but not both
+ // find the entry under one of the names and delete it.
+ String root = MetadataTableUtil.getZookeeperLogLocation();
+ boolean foundEntry = false;
+ for (String entry : unusedWalLogs) {
+ String[] parts = entry.split("/");
+ String zpath = root + "/" + parts[parts.length - 1];
+ while (true) {
+ try {
+ if (zk.exists(zpath)) {
+ zk.recursiveDelete(zpath, NodeMissingPolicy.SKIP);
+ foundEntry = true;
}
+ break;
+ } catch (KeeperException e) {
+ log.error(e, e);
+ } catch (InterruptedException e) {
+ log.error(e, e);
}
- if (unusedWalLogs.size() > 0 && !foundEntry)
- log.warn("WALog entry for root tablet did not exist " + unusedWalLogs);
+ UtilWaitThread.sleep(1000);
}
- return;
}
-
+ if (unusedWalLogs.size() > 0 && !foundEntry)
+ log.warn("WALog entry for root tablet did not exist " + unusedWalLogs);
+ }
+
+ /**
+ * Create an update that updates a tablet
+ * @return A Mutation to update a tablet from the given information
+ */
+ protected static Mutation getUpdateForTabletDataFile(KeyExtent extent, FileRef path, FileRef mergeFile, DataFileValue dfv, String time,
+ Set<FileRef> filesInUseByScans, String address, ZooLock zooLock, Set<String> unusedWalLogs, TServerInstance lastLocation, long flushId) {
Mutation m = new Mutation(extent.getMetadataEntry());
-
+
if (dfv.getNumEntries() > 0) {
m.put(DataFileColumnFamily.NAME, path.meta(), new Value(dfv.encode()));
- TabletsSection.ServerColumnFamily.TIME_COLUMN.put(m, new Value(time.getBytes(StandardCharsets.UTF_8)));
- TabletsSection.ServerColumnFamily.TIME_COLUMN.put(m, new Value(time.getBytes()));
++ TabletsSection.ServerColumnFamily.TIME_COLUMN.put(m, new Value(time.getBytes(UTF_8)));
// stuff in this location
TServerInstance self = getTServerInstance(address, zooLock);
self.putLastLocation(m);
@@@ -313,15 -291,17 +314,15 @@@
m.putDelete(LogColumnFamily.NAME, new Text(entry));
}
}
-
+
for (FileRef scanFile : filesInUseByScans)
- m.put(ScanFileColumnFamily.NAME, scanFile.meta(), new Value("".getBytes()));
-
+ m.put(ScanFileColumnFamily.NAME, scanFile.meta(), new Value(new byte[0]));
+
if (mergeFile != null)
m.putDelete(DataFileColumnFamily.NAME, mergeFile.meta());
-
- TabletsSection.ServerColumnFamily.FLUSH_COLUMN.put(m, new Value((flushId + "").getBytes()));
-
- MetadataTableUtil.update(credentials, zooLock, m, extent);
-
+
- TabletsSection.ServerColumnFamily.FLUSH_COLUMN.put(m, new Value(Long.toString(flushId).getBytes(StandardCharsets.UTF_8)));
++ TabletsSection.ServerColumnFamily.FLUSH_COLUMN.put(m, new Value(Long.toString(flushId).getBytes(UTF_8)));
+
+ return m;
}
-
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java
----------------------------------------------------------------------
diff --cc server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java
index cb97d2a,9cd02eb..8f84169
--- a/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java
@@@ -16,8 -16,9 +16,9 @@@
*/
package org.apache.accumulo.server.util;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+
import java.io.IOException;
- import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/base/src/main/java/org/apache/accumulo/server/util/NamespacePropUtil.java
----------------------------------------------------------------------
diff --cc server/base/src/main/java/org/apache/accumulo/server/util/NamespacePropUtil.java
index 3bd7b8e,c0d6894..85cc484
--- a/server/base/src/main/java/org/apache/accumulo/server/util/NamespacePropUtil.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/NamespacePropUtil.java
@@@ -16,7 -16,7 +16,7 @@@
*/
package org.apache.accumulo.server.util;
- import java.nio.charset.StandardCharsets;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.conf.Property;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/base/src/main/java/org/apache/accumulo/server/util/RandomizeVolumes.java
----------------------------------------------------------------------
diff --cc server/base/src/main/java/org/apache/accumulo/server/util/RandomizeVolumes.java
index 9f298af,8924c85..d797392
--- a/server/base/src/main/java/org/apache/accumulo/server/util/RandomizeVolumes.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/RandomizeVolumes.java
@@@ -16,12 -16,12 +16,12 @@@
*/
package org.apache.accumulo.server.util;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN;
import java.io.IOException;
- import java.nio.charset.StandardCharsets;
import java.util.Map.Entry;
-import org.apache.accumulo.server.security.SystemCredentials;
import org.apache.accumulo.core.cli.ClientOnRequiredTable;
import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
@@@ -105,10 -105,9 +105,10 @@@ public class RandomizeVolumes
}
Key key = entry.getKey();
Mutation m = new Mutation(key.getRow());
-
- String newLocation = vm.choose(ServerConstants.getBaseUris()) + Path.SEPARATOR + ServerConstants.TABLE_DIR + Path.SEPARATOR + tableId + Path.SEPARATOR + directory;
+
+ String newLocation = vm.choose(ServerConstants.getBaseUris()) + Path.SEPARATOR + ServerConstants.TABLE_DIR + Path.SEPARATOR + tableId + Path.SEPARATOR
+ + directory;
- m.put(key.getColumnFamily(), key.getColumnQualifier(), new Value(newLocation.getBytes(StandardCharsets.UTF_8)));
+ m.put(key.getColumnFamily(), key.getColumnQualifier(), new Value(newLocation.getBytes(UTF_8)));
if (log.isTraceEnabled()) {
log.trace("Replacing " + oldLocation + " with " + newLocation);
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/base/src/main/java/org/apache/accumulo/server/util/RestoreZookeeper.java
----------------------------------------------------------------------
diff --cc server/base/src/main/java/org/apache/accumulo/server/util/RestoreZookeeper.java
index 3a8eb0b,a82fa14..b08bf90
--- a/server/base/src/main/java/org/apache/accumulo/server/util/RestoreZookeeper.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/RestoreZookeeper.java
@@@ -16,9 -16,10 +16,10 @@@
*/
package org.apache.accumulo.server.util;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+
import java.io.FileInputStream;
import java.io.InputStream;
- import java.nio.charset.StandardCharsets;
import java.util.Stack;
import javax.xml.parsers.SAXParser;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/base/src/main/java/org/apache/accumulo/server/util/SendLogToChainsaw.java
----------------------------------------------------------------------
diff --cc server/base/src/main/java/org/apache/accumulo/server/util/SendLogToChainsaw.java
index ee01bb2,b1a553d..63f1343
--- a/server/base/src/main/java/org/apache/accumulo/server/util/SendLogToChainsaw.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/SendLogToChainsaw.java
@@@ -16,6 -16,8 +16,8 @@@
*/
package org.apache.accumulo.server.util;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/base/src/main/java/org/apache/accumulo/server/util/SystemPropUtil.java
----------------------------------------------------------------------
diff --cc server/base/src/main/java/org/apache/accumulo/server/util/SystemPropUtil.java
index 9543440,401d98c..a376ed6
--- a/server/base/src/main/java/org/apache/accumulo/server/util/SystemPropUtil.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/SystemPropUtil.java
@@@ -16,7 -16,7 +16,7 @@@
*/
package org.apache.accumulo.server.util;
- import java.nio.charset.StandardCharsets;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.conf.Property;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/base/src/main/java/org/apache/accumulo/server/util/TablePropUtil.java
----------------------------------------------------------------------
diff --cc server/base/src/main/java/org/apache/accumulo/server/util/TablePropUtil.java
index a755c0a,6ee1593..5e6542d
--- a/server/base/src/main/java/org/apache/accumulo/server/util/TablePropUtil.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/TablePropUtil.java
@@@ -16,7 -16,7 +16,7 @@@
*/
package org.apache.accumulo.server.util;
- import java.nio.charset.StandardCharsets;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.conf.Property;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/base/src/main/java/org/apache/accumulo/server/util/TabletServerLocks.java
----------------------------------------------------------------------
diff --cc server/base/src/main/java/org/apache/accumulo/server/util/TabletServerLocks.java
index f334613,edbb103..307bb0c
--- a/server/base/src/main/java/org/apache/accumulo/server/util/TabletServerLocks.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/TabletServerLocks.java
@@@ -16,7 -16,8 +16,8 @@@
*/
package org.apache.accumulo.server.util;
- import java.nio.charset.StandardCharsets;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+
import java.util.List;
import org.apache.accumulo.core.Constants;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/base/src/main/java/org/apache/accumulo/server/watcher/MonitorLog4jWatcher.java
----------------------------------------------------------------------
diff --cc server/base/src/main/java/org/apache/accumulo/server/watcher/MonitorLog4jWatcher.java
index ee20006,7ae4d5a..103aefc
--- a/server/base/src/main/java/org/apache/accumulo/server/watcher/MonitorLog4jWatcher.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/watcher/MonitorLog4jWatcher.java
@@@ -16,7 -16,7 +16,7 @@@
*/
package org.apache.accumulo.server.watcher;
- import java.nio.charset.StandardCharsets;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.zookeeper.ZooUtil;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/base/src/main/java/org/apache/accumulo/server/zookeeper/DistributedWorkQueue.java
----------------------------------------------------------------------
diff --cc server/base/src/main/java/org/apache/accumulo/server/zookeeper/DistributedWorkQueue.java
index 0f1a005,c5a9528..1452aa7
--- a/server/base/src/main/java/org/apache/accumulo/server/zookeeper/DistributedWorkQueue.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/zookeeper/DistributedWorkQueue.java
@@@ -16,7 -16,6 +16,8 @@@
*/
package org.apache.accumulo.server.zookeeper;
- import java.nio.charset.StandardCharsets;
++import static java.nio.charset.StandardCharsets.UTF_8;
++
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@@ -212,14 -200,7 +213,14 @@@ public class DistributedWorkQueue
log.info("Interrupted looking for work", e);
}
}
- }, r.nextInt(60 * 1000), 60 * 1000);
+ }, timerInitialDelay, timerPeriod);
+ }
+
+ /**
- * Adds work to the queue, automatically converting the String to bytes using {@link StandardCharsets#UTF_8}
++ * Adds work to the queue, automatically converting the String to bytes using UTF-8
+ */
+ public void addWork(String workId, String data) throws KeeperException, InterruptedException {
- addWork(workId, data.getBytes(StandardCharsets.UTF_8));
++ addWork(workId, data.getBytes(UTF_8));
}
public void addWork(String workId, byte[] data) throws KeeperException, InterruptedException {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/base/src/main/java/org/apache/accumulo/server/zookeeper/ZooQueueLock.java
----------------------------------------------------------------------
diff --cc server/base/src/main/java/org/apache/accumulo/server/zookeeper/ZooQueueLock.java
index 9a968a9,1bbae03..e4ef1eb
--- a/server/base/src/main/java/org/apache/accumulo/server/zookeeper/ZooQueueLock.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/zookeeper/ZooQueueLock.java
@@@ -16,7 -16,8 +16,8 @@@
*/
package org.apache.accumulo.server.zookeeper;
- import java.nio.charset.StandardCharsets;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/base/src/main/java/org/apache/accumulo/server/zookeeper/ZooReaderWriter.java
----------------------------------------------------------------------
diff --cc server/base/src/main/java/org/apache/accumulo/server/zookeeper/ZooReaderWriter.java
index e6d2350,c5374a2..34c9070
--- a/server/base/src/main/java/org/apache/accumulo/server/zookeeper/ZooReaderWriter.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/zookeeper/ZooReaderWriter.java
@@@ -16,7 -16,7 +16,7 @@@
*/
package org.apache.accumulo.server.zookeeper;
- import java.nio.charset.StandardCharsets;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.Property;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/base/src/main/java/org/apache/accumulo/server/zookeeper/ZooReaderWriterFactory.java
----------------------------------------------------------------------
diff --cc server/base/src/main/java/org/apache/accumulo/server/zookeeper/ZooReaderWriterFactory.java
index e07c2b9,e15e63e..12bdeb1
--- a/server/base/src/main/java/org/apache/accumulo/server/zookeeper/ZooReaderWriterFactory.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/zookeeper/ZooReaderWriterFactory.java
@@@ -16,7 -16,7 +16,7 @@@
*/
package org.apache.accumulo.server.zookeeper;
- import java.nio.charset.StandardCharsets;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.Property;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/base/src/test/java/org/apache/accumulo/server/conf/NamespaceConfigurationTest.java
----------------------------------------------------------------------
diff --cc server/base/src/test/java/org/apache/accumulo/server/conf/NamespaceConfigurationTest.java
index 36ddd3b,3aed7ef..628b981
--- a/server/base/src/test/java/org/apache/accumulo/server/conf/NamespaceConfigurationTest.java
+++ b/server/base/src/test/java/org/apache/accumulo/server/conf/NamespaceConfigurationTest.java
@@@ -16,7 -16,8 +16,18 @@@
*/
package org.apache.accumulo.server.conf;
- import java.nio.charset.StandardCharsets;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
++import static org.junit.Assert.assertEquals;
++import static org.junit.Assert.assertNull;
++import static org.junit.Assert.assertTrue;
++import static org.easymock.EasyMock.anyObject;
++import static org.easymock.EasyMock.createMock;
++import static org.easymock.EasyMock.eq;
++import static org.easymock.EasyMock.expect;
++import static org.easymock.EasyMock.expectLastCall;
++import static org.easymock.EasyMock.replay;
++import static org.easymock.EasyMock.verify;
+
import java.util.Collection;
import java.util.List;
import java.util.Map;
@@@ -34,16 -35,16 +45,6 @@@ import org.apache.accumulo.fate.zookeep
import org.apache.accumulo.fate.zookeeper.ZooCacheFactory;
import org.junit.Before;
import org.junit.Test;
--import static org.junit.Assert.assertEquals;
--import static org.junit.Assert.assertNull;
--import static org.junit.Assert.assertTrue;
--import static org.easymock.EasyMock.anyObject;
--import static org.easymock.EasyMock.createMock;
--import static org.easymock.EasyMock.eq;
--import static org.easymock.EasyMock.expect;
--import static org.easymock.EasyMock.expectLastCall;
--import static org.easymock.EasyMock.replay;
--import static org.easymock.EasyMock.verify;
public class NamespaceConfigurationTest {
private static final String NSID = "namespace";
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/base/src/test/java/org/apache/accumulo/server/conf/ServerConfigurationFactoryTest.java
----------------------------------------------------------------------
diff --cc server/base/src/test/java/org/apache/accumulo/server/conf/ServerConfigurationFactoryTest.java
index 475afbb,81148a0..cec75c2
--- a/server/base/src/test/java/org/apache/accumulo/server/conf/ServerConfigurationFactoryTest.java
+++ b/server/base/src/test/java/org/apache/accumulo/server/conf/ServerConfigurationFactoryTest.java
@@@ -16,6 -16,7 +16,7 @@@
*/
package org.apache.accumulo.server.conf;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
import static org.easymock.EasyMock.anyObject;
import static org.easymock.EasyMock.createMock;
import static org.easymock.EasyMock.endsWith;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/base/src/test/java/org/apache/accumulo/server/conf/TableConfigurationTest.java
----------------------------------------------------------------------
diff --cc server/base/src/test/java/org/apache/accumulo/server/conf/TableConfigurationTest.java
index 12521f9,2f38e4c..e66ec98
--- a/server/base/src/test/java/org/apache/accumulo/server/conf/TableConfigurationTest.java
+++ b/server/base/src/test/java/org/apache/accumulo/server/conf/TableConfigurationTest.java
@@@ -16,6 -16,7 +16,7 @@@
*/
package org.apache.accumulo.server.conf;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
import static org.easymock.EasyMock.anyObject;
import static org.easymock.EasyMock.createMock;
import static org.easymock.EasyMock.eq;
@@@ -84,8 -84,8 +84,8 @@@ public class TableConfigurationTest
@Test
public void testGet_InZK() {
Property p = Property.INSTANCE_SECRET;
- expect(zc.get(ZooUtil.getRoot(iid) + Constants.ZTABLES + "/" + TID + Constants.ZTABLE_CONF + "/" + p.getKey()))
- .andReturn("sekrit".getBytes(UTF_8));
+ expect(zc.get(ZooUtil.getRoot(iid) + Constants.ZTABLES + "/" + TID + Constants.ZTABLE_CONF + "/" + p.getKey())).andReturn(
- "sekrit".getBytes(StandardCharsets.UTF_8));
++ "sekrit".getBytes(UTF_8));
replay(zc);
assertEquals("sekrit", c.get(Property.INSTANCE_SECRET));
}
@@@ -110,10 -110,8 +110,10 @@@
children.add("foo");
children.add("ding");
expect(zc.getChildren(ZooUtil.getRoot(iid) + Constants.ZTABLES + "/" + TID + Constants.ZTABLE_CONF)).andReturn(children);
- expect(zc.get(ZooUtil.getRoot(iid) + Constants.ZTABLES + "/" + TID + Constants.ZTABLE_CONF + "/" + "foo")).andReturn("bar".getBytes(UTF_8));
- expect(zc.get(ZooUtil.getRoot(iid) + Constants.ZTABLES + "/" + TID + Constants.ZTABLE_CONF + "/" + "ding")).andReturn("dong".getBytes(UTF_8));
+ expect(zc.get(ZooUtil.getRoot(iid) + Constants.ZTABLES + "/" + TID + Constants.ZTABLE_CONF + "/" + "foo"))
- .andReturn("bar".getBytes(StandardCharsets.UTF_8));
++ .andReturn("bar".getBytes(UTF_8));
+ expect(zc.get(ZooUtil.getRoot(iid) + Constants.ZTABLES + "/" + TID + Constants.ZTABLE_CONF + "/" + "ding")).andReturn(
- "dong".getBytes(StandardCharsets.UTF_8));
++ "dong".getBytes(UTF_8));
replay(zc);
c.getProperties(props, filter);
assertEquals(2, props.size());
@@@ -137,8 -135,8 +137,8 @@@
public void testInvalidateCache() {
// need to do a get so the accessor is created
Property p = Property.INSTANCE_SECRET;
- expect(zc.get(ZooUtil.getRoot(iid) + Constants.ZTABLES + "/" + TID + Constants.ZTABLE_CONF + "/" + p.getKey()))
- .andReturn("sekrit".getBytes(UTF_8));
+ expect(zc.get(ZooUtil.getRoot(iid) + Constants.ZTABLES + "/" + TID + Constants.ZTABLE_CONF + "/" + p.getKey())).andReturn(
- "sekrit".getBytes(StandardCharsets.UTF_8));
++ "sekrit".getBytes(UTF_8));
zc.clear();
replay(zc);
c.get(Property.INSTANCE_SECRET);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/base/src/test/java/org/apache/accumulo/server/conf/ZooCachePropertyAccessorTest.java
----------------------------------------------------------------------
diff --cc server/base/src/test/java/org/apache/accumulo/server/conf/ZooCachePropertyAccessorTest.java
index 742ccdd,4369119..210d240
--- a/server/base/src/test/java/org/apache/accumulo/server/conf/ZooCachePropertyAccessorTest.java
+++ b/server/base/src/test/java/org/apache/accumulo/server/conf/ZooCachePropertyAccessorTest.java
@@@ -16,6 -16,7 +16,7 @@@
*/
package org.apache.accumulo.server.conf;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
import static org.easymock.EasyMock.createMock;
import static org.easymock.EasyMock.expect;
import static org.easymock.EasyMock.replay;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/master/src/main/java/org/apache/accumulo/master/Master.java
----------------------------------------------------------------------
diff --cc server/master/src/main/java/org/apache/accumulo/master/Master.java
index e5f3955,83e83cf..e3fc69d
--- a/server/master/src/main/java/org/apache/accumulo/master/Master.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/Master.java
@@@ -16,8 -16,9 +16,9 @@@
*/
package org.apache.accumulo.master;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+
import java.io.IOException;
- import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@@ -301,53 -291,20 +302,53 @@@ public class Master implements LiveTSer
IZooReaderWriter zoo = ZooReaderWriter.getInstance();
final String zooRoot = ZooUtil.getRoot(instance);
- if (accumuloPersistentVersion == ServerConstants.TWO_DATA_VERSIONS_AGO) {
- log.debug("Handling updates for version " + ServerConstants.TWO_DATA_VERSIONS_AGO);
+ log.debug("Handling updates for version " + accumuloPersistentVersion);
- log.debug("Cleaning out remnants of logger role.");
- zoo.recursiveDelete(zooRoot + "/loggers", NodeMissingPolicy.SKIP);
- zoo.recursiveDelete(zooRoot + "/dead/loggers", NodeMissingPolicy.SKIP);
+ log.debug("Cleaning out remnants of logger role.");
+ zoo.recursiveDelete(zooRoot + "/loggers", NodeMissingPolicy.SKIP);
+ zoo.recursiveDelete(zooRoot + "/dead/loggers", NodeMissingPolicy.SKIP);
- final byte[] zero = new byte[] {'0'};
- log.debug("Initializing recovery area.");
- zoo.putPersistentData(zooRoot + Constants.ZRECOVERY, zero, NodeExistsPolicy.SKIP);
+ final byte[] zero = new byte[] {'0'};
+ log.debug("Initializing recovery area.");
+ zoo.putPersistentData(zooRoot + Constants.ZRECOVERY, zero, NodeExistsPolicy.SKIP);
- for (String id : zoo.getChildren(zooRoot + Constants.ZTABLES)) {
- log.debug("Prepping table " + id + " for compaction cancellations.");
- zoo.putPersistentData(zooRoot + Constants.ZTABLES + "/" + id + Constants.ZTABLE_COMPACT_CANCEL_ID, zero, NodeExistsPolicy.SKIP);
+ for (String id : zoo.getChildren(zooRoot + Constants.ZTABLES)) {
+ log.debug("Prepping table " + id + " for compaction cancellations.");
+ zoo.putPersistentData(zooRoot + Constants.ZTABLES + "/" + id + Constants.ZTABLE_COMPACT_CANCEL_ID, zero, NodeExistsPolicy.SKIP);
+ }
+
+ @SuppressWarnings("deprecation")
+ String zpath = zooRoot + Constants.ZCONFIG + "/" + Property.TSERV_WAL_SYNC_METHOD.getKey();
+ // is the entire instance set to use flushing vs sync?
+ boolean flushDefault = false;
+ try {
+ byte data[] = zoo.getData(zpath, null);
- if (new String(data, StandardCharsets.UTF_8).endsWith("flush")) {
++ if (new String(data, UTF_8).endsWith("flush")) {
+ flushDefault = true;
+ }
+ } catch (KeeperException.NoNodeException ex) {
+ // skip
+ }
+ for (String id : zoo.getChildren(zooRoot + Constants.ZTABLES)) {
+ log.debug("Converting table " + id + " WALog setting to Durability");
+ try {
+ @SuppressWarnings("deprecation")
+ String path = zooRoot + Constants.ZTABLES + "/" + id + Constants.ZTABLE_CONF + "/" + Property.TABLE_WALOG_ENABLED.getKey();
+ byte[] data = zoo.getData(path, null);
- boolean useWAL = Boolean.parseBoolean(new String(data, StandardCharsets.UTF_8));
++ boolean useWAL = Boolean.parseBoolean(new String(data, UTF_8));
+ zoo.recursiveDelete(path, NodeMissingPolicy.FAIL);
+ path = zooRoot + Constants.ZTABLES + "/" + id + Constants.ZTABLE_CONF + "/" + Property.TABLE_DURABILITY.getKey();
+ if (useWAL) {
+ if (flushDefault) {
+ zoo.putPersistentData(path, "flush".getBytes(), NodeExistsPolicy.SKIP);
+ } else {
+ zoo.putPersistentData(path, "sync".getBytes(), NodeExistsPolicy.SKIP);
+ }
+ } else {
+ zoo.putPersistentData(path, "none".getBytes(), NodeExistsPolicy.SKIP);
+ }
+ } catch (KeeperException.NoNodeException ex) {
+ // skip it
}
}
@@@ -385,8 -342,8 +386,8 @@@
// rename metadata table
log.debug("Upgrade renaming table " + MetadataTable.OLD_NAME + " (ID: " + MetadataTable.ID + ") to " + MetadataTable.NAME);
- zoo.putPersistentData(tables + "/" + MetadataTable.ID + Constants.ZTABLE_NAME, Tables.qualify(MetadataTable.NAME).getSecond().getBytes(UTF_8),
- NodeExistsPolicy.OVERWRITE);
+ zoo.putPersistentData(tables + "/" + MetadataTable.ID + Constants.ZTABLE_NAME,
- Tables.qualify(MetadataTable.NAME).getSecond().getBytes(StandardCharsets.UTF_8), NodeExistsPolicy.OVERWRITE);
++ Tables.qualify(MetadataTable.NAME).getSecond().getBytes(UTF_8), NodeExistsPolicy.OVERWRITE);
moveRootTabletToRootTable(zoo);
@@@ -1087,41 -1030,6 +1088,41 @@@
while (!clientService.isServing()) {
UtilWaitThread.sleep(100);
}
+
+ // Start the daemon to scan the replication table and make units of work
+ replicationWorkDriver = new ReplicationDriver(this);
+ replicationWorkDriver.start();
+
+ // Start the daemon to assign work to tservers to replicate to our peers
+ try {
+ replicationWorkAssigner = new WorkDriver(this, getConnector());
+ } catch (AccumuloException | AccumuloSecurityException e) {
+ log.error("Caught exception trying to initialize replication WorkDriver", e);
+ throw new RuntimeException(e);
+ }
+ replicationWorkAssigner.start();
+
+ // Start the replication coordinator which assigns tservers to service replication requests
+ ReplicationCoordinator.Processor<ReplicationCoordinator.Iface> replicationCoordinatorProcessor = new ReplicationCoordinator.Processor<ReplicationCoordinator.Iface>(
+ RpcWrapper.service(new MasterReplicationCoordinator(this)));
+ ServerAddress replAddress = TServerUtils.startServer(getConfiguration(), hostname, Property.MASTER_REPLICATION_COORDINATOR_PORT,
+ replicationCoordinatorProcessor, "Master Replication Coordinator", "Replication Coordinator", null, Property.MASTER_REPLICATION_COORDINATOR_MINTHREADS,
+ Property.MASTER_REPLICATION_COORDINATOR_THREADCHECK, Property.GENERAL_MAX_MESSAGE_SIZE);
+
+ log.info("Started replication coordinator service at " + replAddress.address);
+
+ // Advertise that port we used so peers don't have to be told what it is
+ ZooReaderWriter.getInstance().putPersistentData(ZooUtil.getRoot(instance) + Constants.ZMASTER_REPLICATION_COORDINATOR_ADDR,
- replAddress.address.toString().getBytes(StandardCharsets.UTF_8), NodeExistsPolicy.OVERWRITE);
++ replAddress.address.toString().getBytes(UTF_8), NodeExistsPolicy.OVERWRITE);
+
+ final SystemCredentials creds = SystemCredentials.get();
+ try {
+ ReplicationMetrics beanImpl = new ReplicationMetrics(this.instance.getConnector(creds.getPrincipal(), creds.getToken()));
+ beanImpl.register();
+ } catch (Exception e) {
+ log.error("Error registering Replication metrics with JMX", e);
+ }
+
while (clientService.isServing()) {
UtilWaitThread.sleep(500);
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/master/src/main/java/org/apache/accumulo/master/recovery/RecoveryManager.java
----------------------------------------------------------------------
diff --cc server/master/src/main/java/org/apache/accumulo/master/recovery/RecoveryManager.java
index ba7a52a,5e478cb..c636dbb
--- a/server/master/src/main/java/org/apache/accumulo/master/recovery/RecoveryManager.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/recovery/RecoveryManager.java
@@@ -16,9 -16,10 +16,10 @@@
*/
package org.apache.accumulo.master.recovery;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+
import java.io.FileNotFoundException;
import java.io.IOException;
- import java.nio.charset.StandardCharsets;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
@@@ -114,10 -114,9 +115,10 @@@ public class RecoveryManager
}
- private void initiateSort(String sortId, String source, final String destination) throws KeeperException, InterruptedException, IOException {
+ private void initiateSort(String sortId, String source, final String destination, AccumuloConfiguration aconf)
+ throws KeeperException, InterruptedException, IOException {
String work = source + "|" + destination;
- new DistributedWorkQueue(ZooUtil.getRoot(master.getInstance()) + Constants.ZRECOVERY, aconf).addWork(sortId, work.getBytes(StandardCharsets.UTF_8));
- new DistributedWorkQueue(ZooUtil.getRoot(master.getInstance()) + Constants.ZRECOVERY).addWork(sortId, work.getBytes(UTF_8));
++ new DistributedWorkQueue(ZooUtil.getRoot(master.getInstance()) + Constants.ZRECOVERY, aconf).addWork(sortId, work.getBytes(UTF_8));
synchronized (this) {
sortsQueued.add(sortId);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/master/src/main/java/org/apache/accumulo/master/replication/MasterReplicationCoordinator.java
----------------------------------------------------------------------
diff --cc server/master/src/main/java/org/apache/accumulo/master/replication/MasterReplicationCoordinator.java
index 795fabb,0000000..8218c8a
mode 100644,000000..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
@@@ -1,107 -1,0 +1,108 @@@
+/*
+ * 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.master.replication;
+
- import java.nio.charset.StandardCharsets;
++import static java.nio.charset.StandardCharsets.UTF_8;
++
+import java.util.Iterator;
+import java.util.Random;
+import java.util.Set;
+
+import org.apache.accumulo.core.client.Instance;
+import org.apache.accumulo.core.client.impl.thrift.ThriftSecurityException;
+import org.apache.accumulo.core.replication.ReplicationConstants;
+import org.apache.accumulo.core.replication.thrift.ReplicationCoordinator;
+import org.apache.accumulo.core.replication.thrift.ReplicationCoordinatorErrorCode;
+import org.apache.accumulo.core.replication.thrift.ReplicationCoordinatorException;
+import org.apache.accumulo.core.security.thrift.TCredentials;
+import org.apache.accumulo.core.zookeeper.ZooUtil;
+import org.apache.accumulo.fate.zookeeper.ZooReader;
+import org.apache.accumulo.master.Master;
+import org.apache.accumulo.server.master.state.TServerInstance;
+import org.apache.accumulo.server.security.SecurityOperation;
+import org.apache.accumulo.server.security.SystemCredentials;
+import org.apache.thrift.TException;
+import org.apache.zookeeper.KeeperException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * Choose a tserver to service a replication task
+ */
+public class MasterReplicationCoordinator implements ReplicationCoordinator.Iface {
+ private static final Logger log = LoggerFactory.getLogger(MasterReplicationCoordinator.class);
+
+ private final Master master;
+ private final Instance inst;
+ private final Random rand;
+ private final ZooReader reader;
+ private final SecurityOperation security;
+
+ public MasterReplicationCoordinator(Master master) {
+ this(master, new ZooReader(master.getInstance().getZooKeepers(), master.getInstance().getZooKeepersSessionTimeOut()));
+ }
+
+ protected MasterReplicationCoordinator(Master master, ZooReader reader) {
+ this.master = master;
+ this.rand = new Random(358923462l);
+ this.inst = master.getInstance();
+ this.reader = reader;
+ this.security = SecurityOperation.getInstance(inst.getInstanceID(), false);
+ }
+
+
+ @Override
+ public String getServicerAddress(String remoteTableId, TCredentials creds) throws ReplicationCoordinatorException, TException {
+ try {
+ security.authenticateUser(SystemCredentials.get().toThrift(inst), creds);
+ } catch (ThriftSecurityException e) {
+ log.error("{} failed to authenticate for replication to {}", creds.getPrincipal(), remoteTableId);
+ throw new ReplicationCoordinatorException(ReplicationCoordinatorErrorCode.CANNOT_AUTHENTICATE, "Could not authenticate " + creds.getPrincipal());
+ }
+
+ Set<TServerInstance> tservers = master.onlineTabletServers();
+ if (tservers.isEmpty()) {
+ throw new ReplicationCoordinatorException(ReplicationCoordinatorErrorCode.NO_AVAILABLE_SERVERS, "No tservers are available for replication");
+ }
+
+ TServerInstance tserver = getRandomTServer(tservers, rand.nextInt(tservers.size()));
+ String replServiceAddr;
+ try {
- replServiceAddr = new String(reader.getData(ZooUtil.getRoot(inst) + ReplicationConstants.ZOO_TSERVERS + "/" + tserver.hostPort(), null), StandardCharsets.UTF_8);
++ replServiceAddr = new String(reader.getData(ZooUtil.getRoot(inst) + ReplicationConstants.ZOO_TSERVERS + "/" + tserver.hostPort(), null), UTF_8);
+ } catch (KeeperException | InterruptedException e) {
+ log.error("Could not fetch repliation service port for tserver", e);
+ throw new ReplicationCoordinatorException(ReplicationCoordinatorErrorCode.SERVICE_CONFIGURATION_UNAVAILABLE,
+ "Could not determine port for replication service running at " + tserver.hostPort());
+ }
+
+ return replServiceAddr;
+ }
+
+ protected TServerInstance getRandomTServer(Set<TServerInstance> tservers, int offset) {
+ Preconditions.checkArgument(tservers.size() > offset, "Must provide an offset less than the size of the set");
+ Iterator<TServerInstance> iter = tservers.iterator();
+ while (offset > 0) {
+ iter.next();
+ offset--;
+ }
+
+ return iter.next();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/master/src/main/java/org/apache/accumulo/master/state/SetGoalState.java
----------------------------------------------------------------------
diff --cc server/master/src/main/java/org/apache/accumulo/master/state/SetGoalState.java
index 3da225d,a674557..2bbbba3
--- a/server/master/src/main/java/org/apache/accumulo/master/state/SetGoalState.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/state/SetGoalState.java
@@@ -16,10 -16,9 +16,10 @@@
*/
package org.apache.accumulo.master.state;
- import java.nio.charset.StandardCharsets;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.conf.SiteConfiguration;
import org.apache.accumulo.core.master.thrift.MasterGoalState;
import org.apache.accumulo.core.security.SecurityUtil;
import org.apache.accumulo.core.zookeeper.ZooUtil;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/master/src/main/java/org/apache/accumulo/master/tableOps/BulkImport.java
----------------------------------------------------------------------
diff --cc server/master/src/main/java/org/apache/accumulo/master/tableOps/BulkImport.java
index b55b315,10cf224..ce25310
--- a/server/master/src/main/java/org/apache/accumulo/master/tableOps/BulkImport.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/tableOps/BulkImport.java
@@@ -16,6 -16,8 +16,8 @@@
*/
package org.apache.accumulo.master.tableOps;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.FileNotFoundException;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/master/src/main/java/org/apache/accumulo/master/tableOps/CancelCompactions.java
----------------------------------------------------------------------
diff --cc server/master/src/main/java/org/apache/accumulo/master/tableOps/CancelCompactions.java
index eb49a35,fd2441c..374fc24
--- a/server/master/src/main/java/org/apache/accumulo/master/tableOps/CancelCompactions.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/tableOps/CancelCompactions.java
@@@ -16,7 -16,7 +16,7 @@@
*/
package org.apache.accumulo.master.tableOps;
- import java.nio.charset.StandardCharsets;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.Instance;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/master/src/main/java/org/apache/accumulo/master/tableOps/CompactRange.java
----------------------------------------------------------------------
diff --cc server/master/src/main/java/org/apache/accumulo/master/tableOps/CompactRange.java
index 2673f03,fbaa70d..13ef68e
--- a/server/master/src/main/java/org/apache/accumulo/master/tableOps/CompactRange.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/tableOps/CompactRange.java
@@@ -16,6 -16,8 +16,8 @@@
*/
package org.apache.accumulo.master.tableOps;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/master/src/main/java/org/apache/accumulo/master/tableOps/ExportTable.java
----------------------------------------------------------------------
diff --cc server/master/src/main/java/org/apache/accumulo/master/tableOps/ExportTable.java
index 5d032d1,037455c..9c397db
--- a/server/master/src/main/java/org/apache/accumulo/master/tableOps/ExportTable.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/tableOps/ExportTable.java
@@@ -16,6 -16,8 +16,8 @@@
*/
package org.apache.accumulo.master.tableOps;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+
import java.io.BufferedOutputStream;
import java.io.BufferedWriter;
import java.io.DataOutputStream;
@@@ -258,10 -259,9 +259,10 @@@ class WriteExportFiles extends MasterRe
Map<String,String> siteConfig = conn.instanceOperations().getSiteConfiguration();
Map<String,String> systemConfig = conn.instanceOperations().getSystemConfiguration();
- TableConfiguration tableConfig = ServerConfiguration.getTableConfiguration(conn.getInstance(), tableID);
+ ServerConfigurationFactory factory = new ServerConfigurationFactory(conn.getInstance());
+ TableConfiguration tableConfig = factory.getTableConfiguration(tableID);
- OutputStreamWriter osw = new OutputStreamWriter(dataOut, StandardCharsets.UTF_8);
+ OutputStreamWriter osw = new OutputStreamWriter(dataOut, UTF_8);
// only put props that are different than defaults and higher level configurations
zipOut.putNextEntry(new ZipEntry(Constants.EXPORT_TABLE_CONFIG_FILE));
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportTable.java
----------------------------------------------------------------------
diff --cc server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportTable.java
index 44ca898,05ec263..35067ce
--- a/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportTable.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/tableOps/ImportTable.java
@@@ -16,6 -16,8 +16,8 @@@
*/
package org.apache.accumulo.master.tableOps;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+
import java.io.BufferedInputStream;
import java.io.BufferedReader;
import java.io.BufferedWriter;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/master/src/main/java/org/apache/accumulo/master/tableOps/RenameTable.java
----------------------------------------------------------------------
diff --cc server/master/src/main/java/org/apache/accumulo/master/tableOps/RenameTable.java
index 7965617,4f381ef..5d511ac
--- a/server/master/src/main/java/org/apache/accumulo/master/tableOps/RenameTable.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/tableOps/RenameTable.java
@@@ -16,7 -16,7 +16,7 @@@
*/
package org.apache.accumulo.master.tableOps;
- import java.nio.charset.StandardCharsets;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.Instance;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/master/src/main/java/org/apache/accumulo/master/tableOps/Utils.java
----------------------------------------------------------------------
diff --cc server/master/src/main/java/org/apache/accumulo/master/tableOps/Utils.java
index 6961776,3e8704e..64947fb
--- a/server/master/src/main/java/org/apache/accumulo/master/tableOps/Utils.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/tableOps/Utils.java
@@@ -16,8 -16,9 +16,9 @@@
*/
package org.apache.accumulo.master.tableOps;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+
import java.math.BigInteger;
- import java.nio.charset.StandardCharsets;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/master/src/main/java/org/apache/accumulo/master/tserverOps/ShutdownTServer.java
----------------------------------------------------------------------
diff --cc server/master/src/main/java/org/apache/accumulo/master/tserverOps/ShutdownTServer.java
index aeb07e2,a9058b7..53c7c6f
--- a/server/master/src/main/java/org/apache/accumulo/master/tserverOps/ShutdownTServer.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/tserverOps/ShutdownTServer.java
@@@ -16,7 -16,7 +16,7 @@@
*/
package org.apache.accumulo.master.tserverOps;
- import java.nio.charset.StandardCharsets;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.master.thrift.TabletServerStatus;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/monitor/src/main/java/org/apache/accumulo/monitor/Monitor.java
----------------------------------------------------------------------
diff --cc server/monitor/src/main/java/org/apache/accumulo/monitor/Monitor.java
index 6135fb0,75d1e9e..49bb56d
--- a/server/monitor/src/main/java/org/apache/accumulo/monitor/Monitor.java
+++ b/server/monitor/src/main/java/org/apache/accumulo/monitor/Monitor.java
@@@ -16,8 -16,9 +16,9 @@@
*/
package org.apache.accumulo.monitor;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+
import java.net.InetAddress;
- import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/monitor/src/main/java/org/apache/accumulo/monitor/ZooKeeperStatus.java
----------------------------------------------------------------------
diff --cc server/monitor/src/main/java/org/apache/accumulo/monitor/ZooKeeperStatus.java
index b0c20ec,78dd95e..7aefa1d
--- a/server/monitor/src/main/java/org/apache/accumulo/monitor/ZooKeeperStatus.java
+++ b/server/monitor/src/main/java/org/apache/accumulo/monitor/ZooKeeperStatus.java
@@@ -16,7 -16,8 +16,8 @@@
*/
package org.apache.accumulo.monitor;
- import java.nio.charset.StandardCharsets;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+
import java.util.Collection;
import java.util.SortedSet;
import java.util.TreeSet;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/BasicServlet.java
----------------------------------------------------------------------
diff --cc server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/BasicServlet.java
index b631320,7812603..2feb804
--- a/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/BasicServlet.java
+++ b/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/BasicServlet.java
@@@ -16,6 -16,8 +16,8 @@@
*/
package org.apache.accumulo.monitor.servlets;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+
import java.io.IOException;
import java.io.PrintWriter;
import java.io.StringWriter;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/DefaultServlet.java
----------------------------------------------------------------------
diff --cc server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/DefaultServlet.java
index 2395335,bb7e690..3347647
--- a/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/DefaultServlet.java
+++ b/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/DefaultServlet.java
@@@ -16,9 -16,10 +16,10 @@@
*/
package org.apache.accumulo.monitor.servlets;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+
import java.io.IOException;
import java.io.InputStream;
- import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/ShellServlet.java
----------------------------------------------------------------------
diff --cc server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/ShellServlet.java
index bf38064,4975428..6b2b31f
--- a/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/ShellServlet.java
+++ b/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/ShellServlet.java
@@@ -16,6 -16,8 +16,8 @@@
*/
package org.apache.accumulo.monitor.servlets;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/XMLServlet.java
----------------------------------------------------------------------
diff --cc server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/XMLServlet.java
index b639226,3cd635e..405d6df
--- a/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/XMLServlet.java
+++ b/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/XMLServlet.java
@@@ -16,7 -16,8 +16,8 @@@
*/
package org.apache.accumulo.monitor.servlets;
- import java.nio.charset.StandardCharsets;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+
import java.util.Map.Entry;
import java.util.SortedMap;
import java.util.TreeMap;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/trace/Basic.java
----------------------------------------------------------------------
diff --cc server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/trace/Basic.java
index c1531cb,1c464d4..475cae1
--- a/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/trace/Basic.java
+++ b/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/trace/Basic.java
@@@ -16,7 -16,8 +16,8 @@@
*/
package org.apache.accumulo.monitor.servlets.trace;
- import java.nio.charset.StandardCharsets;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+
import java.util.Date;
import java.util.Map;
import java.util.Map.Entry;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/tracer/src/main/java/org/apache/accumulo/tracer/TraceServer.java
----------------------------------------------------------------------
diff --cc server/tracer/src/main/java/org/apache/accumulo/tracer/TraceServer.java
index e7e046e,6094911..4858b8a
--- a/server/tracer/src/main/java/org/apache/accumulo/tracer/TraceServer.java
+++ b/server/tracer/src/main/java/org/apache/accumulo/tracer/TraceServer.java
@@@ -16,6 -16,8 +16,8 @@@
*/
package org.apache.accumulo.tracer;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.channels.ServerSocketChannel;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/tserver/src/main/java/org/apache/accumulo/tserver/BulkFailedCopyProcessor.java
----------------------------------------------------------------------
diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/BulkFailedCopyProcessor.java
index ff0097a,4d9a6c2..f7bda49
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/BulkFailedCopyProcessor.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/BulkFailedCopyProcessor.java
@@@ -16,11 -16,12 +16,11 @@@
*/
package org.apache.accumulo.tserver;
- import java.io.IOException;
- import java.nio.charset.StandardCharsets;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+ import java.io.IOException;
-
+import org.apache.accumulo.core.conf.SiteConfiguration;
import org.apache.accumulo.core.util.CachedConfiguration;
-import org.apache.accumulo.server.conf.ServerConfiguration;
import org.apache.accumulo.server.fs.VolumeManager;
import org.apache.accumulo.server.fs.VolumeManagerImpl;
import org.apache.accumulo.server.trace.TraceFileSystem;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
----------------------------------------------------------------------
diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
index 29ed8d7,94be0bb..54c75f8
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
@@@ -16,19 -16,22 +16,19 @@@
*/
package org.apache.accumulo.tserver;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
import static org.apache.accumulo.server.problems.ProblemType.TABLET_LOAD;
import java.io.FileNotFoundException;
import java.io.IOException;
-import java.lang.management.GarbageCollectorMXBean;
import java.lang.management.ManagementFactory;
-import java.net.Socket;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
- import java.nio.charset.StandardCharsets;
-import java.security.SecureRandom;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
-import java.util.EnumMap;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@@@ -412,8 -1198,8 +412,8 @@@ public class TabletServer implements Ru
List<IterInfo> ssiList, Map<String,Map<String,String>> ssio, List<ByteBuffer> authorizations, boolean waitForWrites, boolean isolated,
long readaheadThreshold) throws NotServingTabletException, ThriftSecurityException, org.apache.accumulo.core.tabletserver.thrift.TooManyFilesException {
- String tableId = new String(textent.getTable(), StandardCharsets.UTF_8);
+ String tableId = new String(textent.getTable(), UTF_8);
- if (!security.canScan(credentials, tableId, Tables.getNamespaceId(instance, tableId), range, columns, ssiList, ssio, authorizations))
+ if (!security.canScan(credentials, tableId, Tables.getNamespaceId(getInstance(), tableId), range, columns, ssiList, ssio, authorizations))
throw new ThriftSecurityException(credentials.getPrincipal(), SecurityErrorCode.PERMISSION_DENIED);
if (!security.userHasAuthorizations(credentials, authorizations))
@@@ -552,9 -1347,9 +552,9 @@@
public InitialMultiScan startMultiScan(TInfo tinfo, TCredentials credentials, Map<TKeyExtent,List<TRange>> tbatch, List<TColumn> tcolumns,
List<IterInfo> ssiList, Map<String,Map<String,String>> ssio, List<ByteBuffer> authorizations, boolean waitForWrites) throws ThriftSecurityException {
// find all of the tables that need to be scanned
- HashSet<String> tables = new HashSet<String>();
+ final HashSet<String> tables = new HashSet<String>();
for (TKeyExtent keyExtent : tbatch.keySet()) {
- tables.add(new String(keyExtent.getTable(), StandardCharsets.UTF_8));
+ tables.add(new String(keyExtent.getTable(), UTF_8));
}
if (tables.size() != 1)
@@@ -950,14 -1755,14 +950,14 @@@
}
@Override
- public void update(TInfo tinfo, TCredentials credentials, TKeyExtent tkeyExtent, TMutation tmutation) throws NotServingTabletException,
+ public void update(TInfo tinfo, TCredentials credentials, TKeyExtent tkeyExtent, TMutation tmutation, TDurability tdurability) throws NotServingTabletException,
ConstraintViolationException, ThriftSecurityException {
- final String tableId = new String(tkeyExtent.getTable(), StandardCharsets.UTF_8);
- String tableId = new String(tkeyExtent.getTable(), UTF_8);
- if (!security.canWrite(credentials, tableId, Tables.getNamespaceId(instance, tableId)))
++ final String tableId = new String(tkeyExtent.getTable(), UTF_8);
+ if (!security.canWrite(credentials, tableId, Tables.getNamespaceId(getInstance(), tableId)))
throw new ThriftSecurityException(credentials.getPrincipal(), SecurityErrorCode.PERMISSION_DENIED);
- KeyExtent keyExtent = new KeyExtent(tkeyExtent);
- Tablet tablet = onlineTablets.get(new KeyExtent(keyExtent));
+ final KeyExtent keyExtent = new KeyExtent(tkeyExtent);
+ final Tablet tablet = onlineTablets.get(new KeyExtent(keyExtent));
if (tablet == null) {
throw new NotServingTabletException(tkeyExtent);
}
@@@ -2260,30 -3106,7 +2260,30 @@@
return address;
}
- ZooLock getLock() {
+ private HostAndPort startReplicationService() throws UnknownHostException {
+ ReplicationServicer.Iface repl = RpcWrapper.service(new ReplicationServicerHandler(HdfsZooInstance.getInstance()));
+ ReplicationServicer.Processor<ReplicationServicer.Iface> processor = new ReplicationServicer.Processor<ReplicationServicer.Iface>(repl);
+ AccumuloConfiguration conf = getConfiguration();
+ Property maxMessageSizeProperty = (conf.get(Property.TSERV_MAX_MESSAGE_SIZE) != null ? Property.TSERV_MAX_MESSAGE_SIZE : Property.GENERAL_MAX_MESSAGE_SIZE);
+ ServerAddress sp = TServerUtils.startServer(conf, clientAddress.getHostText(), Property.REPLICATION_RECEIPT_SERVICE_PORT, processor,
+ "ReplicationServicerHandler", "Replication Servicer", null, Property.REPLICATION_MIN_THREADS, Property.REPLICATION_THREADCHECK, maxMessageSizeProperty);
+ this.replServer = sp.server;
+ log.info("Started replication service on " + sp.address);
+
+ try {
+ // The replication service is unique to the thrift service for a tserver, not just a host.
+ // Advertise the host and port for replication service given the host and port for the tserver.
+ ZooReaderWriter.getInstance().putPersistentData(ZooUtil.getRoot(getInstance()) + ReplicationConstants.ZOO_TSERVERS + "/" + clientAddress.toString(),
- sp.address.toString().getBytes(StandardCharsets.UTF_8), NodeExistsPolicy.OVERWRITE);
++ sp.address.toString().getBytes(UTF_8), NodeExistsPolicy.OVERWRITE);
+ } catch (Exception e) {
+ log.error("Could not advertise replication service port", e);
+ throw new RuntimeException(e);
+ }
+
+ return sp.address;
+ }
+
+ public ZooLock getLock() {
return tabletServerLock;
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/tserver/src/main/java/org/apache/accumulo/tserver/log/DfsLogger.java
----------------------------------------------------------------------
diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/log/DfsLogger.java
index 0ea455c,7158ea8..b7d5b0e
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/DfsLogger.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/DfsLogger.java
@@@ -16,14 -16,21 +16,20 @@@
*/
package org.apache.accumulo.tserver.log;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+ import static org.apache.accumulo.tserver.logger.LogEvents.COMPACTION_FINISH;
+ import static org.apache.accumulo.tserver.logger.LogEvents.COMPACTION_START;
+ import static org.apache.accumulo.tserver.logger.LogEvents.DEFINE_TABLET;
+ import static org.apache.accumulo.tserver.logger.LogEvents.MANY_MUTATIONS;
+ import static org.apache.accumulo.tserver.logger.LogEvents.OPEN;
+
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.EOFException;
import java.io.IOException;
import java.io.OutputStream;
-import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.nio.channels.ClosedChannelException;
- import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogFileValue.java
----------------------------------------------------------------------
diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogFileValue.java
index 1468fec,2c807fd..81cd593
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogFileValue.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogFileValue.java
@@@ -16,6 -16,8 +16,8 @@@
*/
package org.apache.accumulo.tserver.logger;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java
----------------------------------------------------------------------
diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java
index a444b45,29eefc8..df2f831
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java
@@@ -16,6 -16,8 +16,8 @@@
*/
package org.apache.accumulo.tserver.logger;
-import static com.google.common.base.Charsets.UTF_8;
++import static java.nio.charset.StandardCharsets.UTF_8;
+
import java.io.DataInputStream;
import java.io.EOFException;
import java.io.IOException;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/cfbdef13/server/tserver/src/main/java/org/apache/accumulo/tserver/replication/ReplicationProcessor.java
----------------------------------------------------------------------
diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/replication/ReplicationProcessor.java
index fbae6f2,0000000..4b57813
mode 100644,000000..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
@@@ -1,181 -1,0 +1,182 @@@
+/*
+ * 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.tserver.replication;
+
++import static java.nio.charset.StandardCharsets.UTF_8;
++
+import java.io.IOException;
- import java.nio.charset.StandardCharsets;
+import java.util.Map;
+import java.util.NoSuchElementException;
+
+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.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.Range;
+import org.apache.accumulo.core.protobuf.ProtobufUtil;
+import org.apache.accumulo.core.replication.ReplicaSystemHelper;
+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.conf.ServerConfigurationFactory;
+import org.apache.accumulo.server.fs.VolumeManager;
+import org.apache.accumulo.server.replication.DistributedWorkQueueWorkAssignerHelper;
+import org.apache.accumulo.server.replication.ReplicationTable;
+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 final Instance inst;
+ private final AccumuloConfiguration conf;
+ private final VolumeManager fs;
+ private final Credentials creds;
+ private final ReplicaSystemHelper helper;
+
+ public ReplicationProcessor(Instance inst, AccumuloConfiguration conf, VolumeManager fs, Credentials creds) {
+ this.inst = inst;
+ this.conf = conf;
+ this.fs = fs;
+ this.creds = creds;
+ this.helper = new ReplicaSystemHelper(inst, creds);
+ }
+
+ @Override
+ public ReplicationProcessor newProcessor() {
+ return new ReplicationProcessor(inst, new ServerConfigurationFactory(inst).getConfiguration(), fs, creds);
+ }
+
+ @Override
+ public void process(String workID, byte[] data) {
+ ReplicationTarget target = DistributedWorkQueueWorkAssignerHelper.fromQueueKey(workID).getValue();
- String file = new String(data, StandardCharsets.UTF_8);
++ String file = new String(data, UTF_8);
+
+ log.debug("Received replication work for {} to {}", file, target);
+
+ ReplicaSystem replica;
+ try {
+ replica = getReplicaSystem(target);
+ } catch (Exception e) {
+ log.error("Could not instantiate ReplicaSystem for {}, waiting before returning the work", target, e);
+ try {
+ // TODO configurable
+ Thread.sleep(5000);
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ }
+
+ return;
+ }
+
+ Status status;
+ try {
+ status = getStatus(file, target);
+ } catch (TableNotFoundException | AccumuloException | AccumuloSecurityException e) {
+ log.error("Could not look for 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);
+ } catch (NoSuchElementException e) {
+ log.error("Assigned work for {} to {} but could not find work record", file, target);
+ return;
+ }
+
+ log.debug("Current status for {} replicating to {}: {}", file, target, ProtobufUtil.toString(status));
+
+ // 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 (!doesFileExist(filePath, target)) {
+ return;
+ }
+ } catch (IOException e) {
+ log.error("Could not determine if file exists {}", filePath, e);
+ throw new RuntimeException(e);
+ }
+
+ log.debug("Replicating {} to {} using {}", filePath, target, replica.getClass().getName());
+
+ replica.replicate(filePath, status, target, getHelper());
+ }
+
+ protected ReplicaSystemHelper getHelper() {
+ return helper;
+ }
+
+ protected ReplicaSystem getReplicaSystem(ReplicationTarget target) {
+ // Find the configured replication peer so we know how to replicate to it
+ // Classname,Configuration
+ String peerType = getPeerType(target.getPeerName());
+
+ // Get the peer that we're replicating to
+ return ReplicaSystemFactory.get(peerType);
+ }
+
+ protected String getPeerType(String peerName) {
+ // 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(Property.REPLICATION_PEERS.getKey() + peerName);
+ if (null == peerType) {
+ String msg = "Cannot process replication for unknown peer: " + peerName;
+ log.warn(msg);
+ throw new IllegalArgumentException(msg);
+ }
+
+ return peerType;
+ }
+
+ protected boolean doesFileExist(Path filePath, ReplicationTarget target) throws IOException {
+ if (!fs.exists(filePath)) {
+ log.warn("Received work request for {} and {}, but the file doesn't exist", filePath, target);
+ return false;
+ }
+
+ return true;
+ }
+
+ protected 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());
+ }
+}