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/07 23:31:16 UTC
[01/10] accumulo git commit: ACCUMULO-3147 Refactor ReplicationTable
constants
Repository: accumulo
Updated Branches:
refs/heads/master 5aff1fb7b -> 34ca056b3
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/server/master/src/test/java/org/apache/accumulo/master/replication/UnorderedWorkAssignerTest.java
----------------------------------------------------------------------
diff --git a/server/master/src/test/java/org/apache/accumulo/master/replication/UnorderedWorkAssignerTest.java b/server/master/src/test/java/org/apache/accumulo/master/replication/UnorderedWorkAssignerTest.java
index 2199808..5d69216 100644
--- a/server/master/src/test/java/org/apache/accumulo/master/replication/UnorderedWorkAssignerTest.java
+++ b/server/master/src/test/java/org/apache/accumulo/master/replication/UnorderedWorkAssignerTest.java
@@ -41,13 +41,14 @@ import org.apache.accumulo.core.protobuf.ProtobufUtil;
import org.apache.accumulo.core.replication.ReplicationConstants;
import org.apache.accumulo.core.replication.ReplicationSchema.OrderSection;
import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
+import org.apache.accumulo.core.replication.ReplicationTable;
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.core.security.TablePermission;
import org.apache.accumulo.server.replication.DistributedWorkQueueWorkAssignerHelper;
-import org.apache.accumulo.server.replication.ReplicationTable;
+import org.apache.accumulo.server.replication.ReplicationUtil;
import org.apache.accumulo.server.zookeeper.DistributedWorkQueue;
import org.apache.accumulo.server.zookeeper.ZooCache;
import org.apache.hadoop.fs.Path;
@@ -85,9 +86,10 @@ public class UnorderedWorkAssignerTest {
Path p = new Path("/accumulo/wal/tserver+port/" + UUID.randomUUID());
- String expectedQueueKey = p.getName() + DistributedWorkQueueWorkAssignerHelper.KEY_SEPARATOR + target.getPeerName() + DistributedWorkQueueWorkAssignerHelper.KEY_SEPARATOR
- + target.getRemoteIdentifier() + DistributedWorkQueueWorkAssignerHelper.KEY_SEPARATOR + target.getSourceTableId();
-
+ String expectedQueueKey = p.getName() + DistributedWorkQueueWorkAssignerHelper.KEY_SEPARATOR + target.getPeerName()
+ + DistributedWorkQueueWorkAssignerHelper.KEY_SEPARATOR + target.getRemoteIdentifier() + DistributedWorkQueueWorkAssignerHelper.KEY_SEPARATOR
+ + target.getSourceTableId();
+
workQueue.addWork(expectedQueueKey, p.toString());
expectLastCall().once();
@@ -135,7 +137,7 @@ public class UnorderedWorkAssignerTest {
assigner.setConnector(conn);
// Create and grant ourselves write to the replication table
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.WRITE);
Status.Builder builder = Status.newBuilder().setBegin(0).setEnd(0).setInfiniteEnd(true).setClosed(false).setCreatedTime(5l);
@@ -198,7 +200,7 @@ public class UnorderedWorkAssignerTest {
assigner.setConnector(conn);
// Create and grant ourselves write to the replication table
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.WRITE);
// Create two mutations, both of which need replication work done
@@ -270,7 +272,7 @@ public class UnorderedWorkAssignerTest {
assigner.setConnector(conn);
// Create and grant ourselves write to the replication table
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.WRITE);
// Create two mutations, both of which need replication work done
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/server/master/src/test/java/org/apache/accumulo/master/replication/WorkMakerTest.java
----------------------------------------------------------------------
diff --git a/server/master/src/test/java/org/apache/accumulo/master/replication/WorkMakerTest.java b/server/master/src/test/java/org/apache/accumulo/master/replication/WorkMakerTest.java
index d486ded..280ca82 100644
--- a/server/master/src/test/java/org/apache/accumulo/master/replication/WorkMakerTest.java
+++ b/server/master/src/test/java/org/apache/accumulo/master/replication/WorkMakerTest.java
@@ -33,11 +33,11 @@ import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
+import org.apache.accumulo.core.replication.ReplicationTable;
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.server.conf.TableConfiguration;
-import org.apache.accumulo.server.replication.ReplicationTable;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.junit.Assert;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/ReplicationServlet.java
----------------------------------------------------------------------
diff --git a/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/ReplicationServlet.java b/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/ReplicationServlet.java
index 1cb6abf..4c21c39 100644
--- a/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/ReplicationServlet.java
+++ b/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/ReplicationServlet.java
@@ -28,6 +28,7 @@ import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.admin.TableOperations;
import org.apache.accumulo.core.master.thrift.MasterMonitorInfo;
import org.apache.accumulo.core.replication.ReplicationConstants;
+import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.ReplicationTarget;
import org.apache.accumulo.core.security.Credentials;
import org.apache.accumulo.core.zookeeper.ZooUtil;
@@ -75,7 +76,7 @@ public class ReplicationServlet extends BasicServlet {
int totalWorkQueueSize = replicationUtil.getMaxReplicationThreads(systemProps, mmi);
TableOperations tops = conn.tableOperations();
- if (!tops.exists(ReplicationConstants.TABLE_NAME)) {
+ if (!tops.exists(ReplicationTable.NAME)) {
banner(sb, "", "Replication table does not yet exist");
return;
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/server/tserver/src/main/java/org/apache/accumulo/tserver/replication/ReplicationProcessor.java
----------------------------------------------------------------------
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/replication/ReplicationProcessor.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/replication/ReplicationProcessor.java
index 4b57813..1843acb 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
@@ -35,6 +35,7 @@ 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.ReplicationTable;
import org.apache.accumulo.core.replication.ReplicationTarget;
import org.apache.accumulo.core.replication.StatusUtil;
import org.apache.accumulo.core.replication.proto.Replication.Status;
@@ -42,7 +43,6 @@ 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;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/test/src/test/java/org/apache/accumulo/test/replication/MultiInstanceReplicationIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/replication/MultiInstanceReplicationIT.java b/test/src/test/java/org/apache/accumulo/test/replication/MultiInstanceReplicationIT.java
index 9e571cd..0000e7f 100644
--- a/test/src/test/java/org/apache/accumulo/test/replication/MultiInstanceReplicationIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/replication/MultiInstanceReplicationIT.java
@@ -43,6 +43,7 @@ import org.apache.accumulo.core.metadata.MetadataTable;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.ReplicationSection;
import org.apache.accumulo.core.protobuf.ProtobufUtil;
import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
+import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.StatusUtil;
import org.apache.accumulo.core.replication.proto.Replication.Status;
import org.apache.accumulo.core.security.Authorizations;
@@ -53,7 +54,7 @@ import org.apache.accumulo.minicluster.ServerType;
import org.apache.accumulo.minicluster.impl.MiniAccumuloClusterImpl;
import org.apache.accumulo.minicluster.impl.MiniAccumuloConfigImpl;
import org.apache.accumulo.minicluster.impl.ProcessReference;
-import org.apache.accumulo.server.replication.ReplicationTable;
+import org.apache.accumulo.server.replication.ReplicationUtil;
import org.apache.accumulo.test.functional.ConfigurableMacIT;
import org.apache.accumulo.tserver.TabletServer;
import org.apache.accumulo.tserver.replication.AccumuloReplicaSystem;
@@ -163,7 +164,7 @@ public class MultiInstanceReplicationIT extends ConfigurableMacIT {
final Connector connMaster = getConnector();
final Connector connPeer = peerCluster.getConnector("root", ROOT_PASSWORD);
- ReplicationTable.create(connMaster);
+ ReplicationUtil.createReplicationTable(connMaster);
String peerUserName = "peer", peerPassword = "foo";
@@ -640,7 +641,6 @@ public class MultiInstanceReplicationIT extends ConfigurableMacIT {
Thread.sleep(500);
}
-
for (ProcessReference proc : cluster.getProcesses().get(ServerType.TABLET_SERVER)) {
cluster.killProcess(ServerType.TABLET_SERVER, proc);
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/test/src/test/java/org/apache/accumulo/test/replication/ReplicationIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/replication/ReplicationIT.java b/test/src/test/java/org/apache/accumulo/test/replication/ReplicationIT.java
index fb96d7f..69cd195 100644
--- a/test/src/test/java/org/apache/accumulo/test/replication/ReplicationIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/replication/ReplicationIT.java
@@ -49,6 +49,7 @@ import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.Lo
import org.apache.accumulo.core.protobuf.ProtobufUtil;
import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
+import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.ReplicationTarget;
import org.apache.accumulo.core.replication.StatusUtil;
import org.apache.accumulo.core.replication.proto.Replication.Status;
@@ -60,7 +61,7 @@ import org.apache.accumulo.gc.SimpleGarbageCollector;
import org.apache.accumulo.minicluster.ServerType;
import org.apache.accumulo.minicluster.impl.MiniAccumuloConfigImpl;
import org.apache.accumulo.minicluster.impl.ProcessReference;
-import org.apache.accumulo.server.replication.ReplicationTable;
+import org.apache.accumulo.server.replication.ReplicationUtil;
import org.apache.accumulo.server.util.ReplicationTableUtil;
import org.apache.accumulo.test.functional.ConfigurableMacIT;
import org.apache.accumulo.tserver.TabletServer;
@@ -81,9 +82,8 @@ import com.google.common.collect.Sets;
import com.google.protobuf.TextFormat;
/**
- * Replication tests which verify expected functionality using a single MAC instance. A MockReplicaSystem
- * is used to "fake" the peer instance that we're replicating to. This lets us test replication in a functional
- * way without having to worry about two real systems.
+ * Replication tests which verify expected functionality using a single MAC instance. A MockReplicaSystem is used to "fake" the peer instance that we're
+ * replicating to. This lets us test replication in a functional way without having to worry about two real systems.
*/
public class ReplicationIT extends ConfigurableMacIT {
private static final Logger log = LoggerFactory.getLogger(ReplicationIT.class);
@@ -311,7 +311,7 @@ public class ReplicationIT extends ConfigurableMacIT {
conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.READ);
// Verify that we found a single replication record that's for table1
- Scanner s = ReplicationTable.getScanner(conn, new Authorizations());
+ Scanner s = ReplicationTable.getScanner(conn);
StatusSection.limit(s);
Iterator<Entry<Key,Value>> iter = s.iterator();
attempts = 5;
@@ -319,7 +319,7 @@ public class ReplicationIT extends ConfigurableMacIT {
if (!iter.hasNext()) {
s.close();
Thread.sleep(1000);
- s = ReplicationTable.getScanner(conn, new Authorizations());
+ s = ReplicationTable.getScanner(conn);
iter = s.iterator();
attempts--;
} else {
@@ -384,7 +384,7 @@ public class ReplicationIT extends ConfigurableMacIT {
Thread.sleep(5000);
// Verify that we found two replication records: one for table1 and one for table2
- s = ReplicationTable.getScanner(conn, new Authorizations());
+ s = ReplicationTable.getScanner(conn);
StatusSection.limit(s);
iter = s.iterator();
Assert.assertTrue("Found no records in replication table", iter.hasNext());
@@ -562,7 +562,7 @@ public class ReplicationIT extends ConfigurableMacIT {
conn.tableOperations().delete(ReplicationTable.NAME);
}
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.WRITE);
final AtomicBoolean keepRunning = new AtomicBoolean(true);
@@ -1063,7 +1063,7 @@ public class ReplicationIT extends ConfigurableMacIT {
conn.tableOperations().delete(ReplicationTable.NAME);
}
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.WRITE);
final AtomicBoolean keepRunning = new AtomicBoolean(true);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/test/src/test/java/org/apache/accumulo/test/replication/StatusCombinerMacTest.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/replication/StatusCombinerMacTest.java b/test/src/test/java/org/apache/accumulo/test/replication/StatusCombinerMacTest.java
index 396df20..ff64d0b 100644
--- a/test/src/test/java/org/apache/accumulo/test/replication/StatusCombinerMacTest.java
+++ b/test/src/test/java/org/apache/accumulo/test/replication/StatusCombinerMacTest.java
@@ -35,10 +35,11 @@ import org.apache.accumulo.core.metadata.MetadataTable;
import org.apache.accumulo.core.metadata.schema.MetadataSchema;
import org.apache.accumulo.core.protobuf.ProtobufUtil;
import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
+import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.StatusUtil;
import org.apache.accumulo.core.replication.proto.Replication.Status;
import org.apache.accumulo.core.security.Authorizations;
-import org.apache.accumulo.server.replication.ReplicationTable;
+import org.apache.accumulo.server.replication.ReplicationUtil;
import org.apache.accumulo.server.util.ReplicationTableUtil;
import org.apache.accumulo.test.functional.SimpleMacIT;
import org.apache.hadoop.io.Text;
@@ -84,7 +85,7 @@ public class StatusCombinerMacTest extends SimpleMacIT {
conn.tableOperations().delete(ReplicationTable.NAME);
}
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
BatchWriter bw = conn.createBatchWriter(ReplicationTable.NAME, new BatchWriterConfig());
long createTime = System.currentTimeMillis();
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/test/src/test/java/org/apache/accumulo/test/replication/UnorderedWorkAssignerReplicationIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/replication/UnorderedWorkAssignerReplicationIT.java b/test/src/test/java/org/apache/accumulo/test/replication/UnorderedWorkAssignerReplicationIT.java
index a4d6179..edd49ab 100644
--- a/test/src/test/java/org/apache/accumulo/test/replication/UnorderedWorkAssignerReplicationIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/replication/UnorderedWorkAssignerReplicationIT.java
@@ -43,6 +43,7 @@ import org.apache.accumulo.core.metadata.MetadataTable;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.ReplicationSection;
import org.apache.accumulo.core.protobuf.ProtobufUtil;
import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
+import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.StatusUtil;
import org.apache.accumulo.core.replication.proto.Replication.Status;
import org.apache.accumulo.core.security.Authorizations;
@@ -53,7 +54,7 @@ import org.apache.accumulo.minicluster.ServerType;
import org.apache.accumulo.minicluster.impl.MiniAccumuloClusterImpl;
import org.apache.accumulo.minicluster.impl.MiniAccumuloConfigImpl;
import org.apache.accumulo.minicluster.impl.ProcessReference;
-import org.apache.accumulo.server.replication.ReplicationTable;
+import org.apache.accumulo.server.replication.ReplicationUtil;
import org.apache.accumulo.test.functional.ConfigurableMacIT;
import org.apache.accumulo.tserver.TabletServer;
import org.apache.accumulo.tserver.replication.AccumuloReplicaSystem;
@@ -167,7 +168,7 @@ public class UnorderedWorkAssignerReplicationIT extends ConfigurableMacIT {
final Connector connMaster = getConnector();
final Connector connPeer = peerCluster.getConnector("root", ROOT_PASSWORD);
- ReplicationTable.create(connMaster);
+ ReplicationUtil.createReplicationTable(connMaster);
String peerUserName = "peer", peerPassword = "foo";
@@ -703,7 +704,7 @@ public class UnorderedWorkAssignerReplicationIT extends ConfigurableMacIT {
Assert.assertTrue("Did not find any records in " + peerTable1 + " on peer", countTable > 0);
- for (int i = 0; i < 10; i++ ) {
+ for (int i = 0; i < 10; i++) {
countTable = 0l;
for (Entry<Key,Value> entry : connPeer.createScanner(peerTable2, Authorizations.EMPTY)) {
countTable++;
[03/10] accumulo git commit: ACCUMULO-3147 Create replication table
at initialize, upgrade
Posted by ct...@apache.org.
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/test/src/test/java/org/apache/accumulo/test/replication/UnorderedWorkAssignerReplicationIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/replication/UnorderedWorkAssignerReplicationIT.java b/test/src/test/java/org/apache/accumulo/test/replication/UnorderedWorkAssignerReplicationIT.java
index edd49ab..ad9c9fc 100644
--- a/test/src/test/java/org/apache/accumulo/test/replication/UnorderedWorkAssignerReplicationIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/replication/UnorderedWorkAssignerReplicationIT.java
@@ -54,7 +54,6 @@ import org.apache.accumulo.minicluster.ServerType;
import org.apache.accumulo.minicluster.impl.MiniAccumuloClusterImpl;
import org.apache.accumulo.minicluster.impl.MiniAccumuloConfigImpl;
import org.apache.accumulo.minicluster.impl.ProcessReference;
-import org.apache.accumulo.server.replication.ReplicationUtil;
import org.apache.accumulo.test.functional.ConfigurableMacIT;
import org.apache.accumulo.tserver.TabletServer;
import org.apache.accumulo.tserver.replication.AccumuloReplicaSystem;
@@ -168,7 +167,7 @@ public class UnorderedWorkAssignerReplicationIT extends ConfigurableMacIT {
final Connector connMaster = getConnector();
final Connector connPeer = peerCluster.getConnector("root", ROOT_PASSWORD);
- ReplicationUtil.createReplicationTable(connMaster);
+ ReplicationTable.setOnline(connMaster);
String peerUserName = "peer", peerPassword = "foo";
@@ -243,7 +242,7 @@ public class UnorderedWorkAssignerReplicationIT extends ConfigurableMacIT {
log.info("");
log.info("Fetching replication records:");
- for (Entry<Key,Value> kv : connMaster.createScanner(ReplicationTable.NAME, Authorizations.EMPTY)) {
+ for (Entry<Key,Value> kv : ReplicationTable.getScanner(connMaster)) {
log.info(kv.getKey().toStringNoTruncate() + " " + ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
}
@@ -280,7 +279,7 @@ public class UnorderedWorkAssignerReplicationIT extends ConfigurableMacIT {
log.info("");
log.info("Fetching replication records:");
- for (Entry<Key,Value> kv : connMaster.createScanner(ReplicationTable.NAME, Authorizations.EMPTY)) {
+ for (Entry<Key,Value> kv : ReplicationTable.getScanner(connMaster)) {
log.info(kv.getKey().toStringNoTruncate() + " " + ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
}
@@ -404,7 +403,7 @@ public class UnorderedWorkAssignerReplicationIT extends ConfigurableMacIT {
Set<String> filesFor1 = connMaster.replicationOperations().referencedFiles(masterTable1), filesFor2 = connMaster.replicationOperations().referencedFiles(
masterTable2);
- while (!connMaster.tableOperations().exists(ReplicationTable.NAME)) {
+ while (!ReplicationTable.isOnline(connMaster)) {
Thread.sleep(500);
}
@@ -658,7 +657,7 @@ public class UnorderedWorkAssignerReplicationIT extends ConfigurableMacIT {
log.info("Wrote all data to master cluster");
- while (!connMaster.tableOperations().exists(ReplicationTable.NAME)) {
+ while (!ReplicationTable.isOnline(connMaster)) {
Thread.sleep(500);
}
[10/10] accumulo git commit: Merge branch 'ACCUMULO-3147'
Posted by ct...@apache.org.
Merge branch 'ACCUMULO-3147'
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/34ca056b
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/34ca056b
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/34ca056b
Branch: refs/heads/master
Commit: 34ca056b35fdfe60d782241ef72a44d04986080c
Parents: 5aff1fb bc62474
Author: Christopher Tubbs <ct...@apache.org>
Authored: Fri Nov 7 17:29:26 2014 -0500
Committer: Christopher Tubbs <ct...@apache.org>
Committed: Fri Nov 7 17:29:26 2014 -0500
----------------------------------------------------------------------
.../client/impl/ReplicationOperationsImpl.java | 11 +-
.../accumulo/core/client/mock/MockAccumulo.java | 2 +
.../core/client/mock/MockTableOperations.java | 8 +-
.../replication/PrintReplicationRecords.java | 4 +-
.../core/replication/ReplicaSystemHelper.java | 11 +-
.../core/replication/ReplicationConstants.java | 1 -
.../core/replication/ReplicationTable.java | 99 ++++++++
.../ReplicationTableOfflineException.java | 32 +++
.../ReplicationOperationsImplTest.java | 22 +-
.../apache/accumulo/server/ServerConstants.java | 8 +-
.../server/constraints/MetadataConstraints.java | 119 +++++-----
.../apache/accumulo/server/init/Initialize.java | 174 +++++++++-----
.../server/replication/ReplicationTable.java | 227 -------------------
.../server/replication/ReplicationUtil.java | 47 ++--
.../server/security/SecurityOperation.java | 3 +-
.../accumulo/server/util/MetadataTableUtil.java | 24 +-
.../replication/ReplicationTableTest.java | 132 -----------
.../gc/GarbageCollectWriteAheadLogs.java | 42 ++--
.../accumulo/gc/SimpleGarbageCollector.java | 7 +-
.../CloseWriteAheadLogReferences.java | 16 +-
.../gc/GarbageCollectWriteAheadLogsTest.java | 37 ++-
.../CloseWriteAheadLogReferencesTest.java | 15 +-
.../accumulo/master/FateServiceHandler.java | 3 +-
.../java/org/apache/accumulo/master/Master.java | 21 +-
.../master/metrics/ReplicationMetrics.java | 7 +-
.../DistributedWorkQueueWorkAssigner.java | 20 +-
.../master/replication/FinishedWorkUpdater.java | 15 +-
.../RemoveCompleteReplicationRecords.java | 12 +-
.../master/replication/StatusMaker.java | 57 ++---
.../accumulo/master/replication/WorkMaker.java | 12 +-
.../accumulo/master/util/TableValidators.java | 26 ++-
.../replication/FinishedWorkUpdaterTest.java | 16 +-
.../RemoveCompleteReplicationRecordsTest.java | 9 +-
.../replication/SequentialWorkAssignerTest.java | 16 +-
.../master/replication/StatusMakerTest.java | 2 +-
.../replication/UnorderedWorkAssignerTest.java | 18 +-
.../master/replication/WorkMakerTest.java | 17 +-
.../monitor/servlets/ReplicationServlet.java | 5 +-
.../replication/ReplicationProcessor.java | 8 +-
.../apache/accumulo/test/LargeSplitRowIT.java | 5 +-
.../test/MasterRepairsDualAssignmentIT.java | 8 +-
.../accumulo/test/MetaConstraintRetryIT.java | 7 +-
.../org/apache/accumulo/test/NamespacesIT.java | 6 +-
.../org/apache/accumulo/test/ShellServerIT.java | 17 +-
.../apache/accumulo/test/WaitForBalanceIT.java | 14 +-
.../replication/MultiInstanceReplicationIT.java | 21 +-
.../test/replication/ReplicationIT.java | 210 +++++++++++------
.../test/replication/StatusCombinerMacTest.java | 20 +-
.../UnorderedWorkAssignerReplicationIT.java | 14 +-
49 files changed, 800 insertions(+), 827 deletions(-)
----------------------------------------------------------------------
[06/10] accumulo git commit: Merge remote-tracking branch
'origin/master' into ACCUMULO-3147
Posted by ct...@apache.org.
Merge remote-tracking branch 'origin/master' into ACCUMULO-3147
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/fa717c91
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/fa717c91
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/fa717c91
Branch: refs/heads/master
Commit: fa717c916201f94801b4d2d1d0f0695f14d1843b
Parents: e840549 b5d7b94
Author: Christopher Tubbs <ct...@apache.org>
Authored: Fri Nov 7 12:40:31 2014 -0500
Committer: Christopher Tubbs <ct...@apache.org>
Committed: Fri Nov 7 12:40:31 2014 -0500
----------------------------------------------------------------------
.../accumulo/server/util/MetadataTableUtil.java | 2 +-
.../accumulo/test/functional/CloneTestIT.java | 90 +++++++++++++++++++-
2 files changed, 90 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/fa717c91/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java
----------------------------------------------------------------------
[07/10] accumulo git commit: ACCUMULO-3147 Fix replication test bug
Posted by ct...@apache.org.
ACCUMULO-3147 Fix replication test bug
Test was trying to talk to ZooKeeper for MockInstance
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/5753e168
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/5753e168
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/5753e168
Branch: refs/heads/master
Commit: 5753e16839cf149110372bd95e26e92208d6c266
Parents: fa717c9
Author: Christopher Tubbs <ct...@apache.org>
Authored: Fri Nov 7 13:03:08 2014 -0500
Committer: Christopher Tubbs <ct...@apache.org>
Committed: Fri Nov 7 13:03:08 2014 -0500
----------------------------------------------------------------------
.../org/apache/accumulo/core/replication/ReplicationTable.java | 3 ++-
.../accumulo/master/replication/FinishedWorkUpdaterTest.java | 2 +-
2 files changed, 3 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/5753e168/core/src/main/java/org/apache/accumulo/core/replication/ReplicationTable.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/replication/ReplicationTable.java b/core/src/main/java/org/apache/accumulo/core/replication/ReplicationTable.java
index b6d343f..ec7c202 100644
--- a/core/src/main/java/org/apache/accumulo/core/replication/ReplicationTable.java
+++ b/core/src/main/java/org/apache/accumulo/core/replication/ReplicationTable.java
@@ -31,6 +31,7 @@ import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.client.TableOfflineException;
import org.apache.accumulo.core.client.impl.Namespaces;
import org.apache.accumulo.core.client.impl.Tables;
+import org.apache.accumulo.core.client.mock.MockInstance;
import org.apache.accumulo.core.master.state.tables.TableState;
import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
@@ -84,7 +85,7 @@ public class ReplicationTable {
}
public static boolean isOnline(Connector conn) {
- return TableState.ONLINE == Tables.getTableState(conn.getInstance(), ID);
+ return conn.getInstance() instanceof MockInstance || TableState.ONLINE == Tables.getTableState(conn.getInstance(), ID);
}
public static void setOnline(Connector conn) throws AccumuloSecurityException, AccumuloException {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/5753e168/server/master/src/test/java/org/apache/accumulo/master/replication/FinishedWorkUpdaterTest.java
----------------------------------------------------------------------
diff --git a/server/master/src/test/java/org/apache/accumulo/master/replication/FinishedWorkUpdaterTest.java b/server/master/src/test/java/org/apache/accumulo/master/replication/FinishedWorkUpdaterTest.java
index 7e8f762..ae84472 100644
--- a/server/master/src/test/java/org/apache/accumulo/master/replication/FinishedWorkUpdaterTest.java
+++ b/server/master/src/test/java/org/apache/accumulo/master/replication/FinishedWorkUpdaterTest.java
@@ -61,7 +61,7 @@ public class FinishedWorkUpdaterTest {
}
@Test
- public void noReplicationTableFailsGracefully() {
+ public void offlineReplicationTableFailsGracefully() {
updater.run();
}
[05/10] accumulo git commit: ACCUMULO-3147 Create replication table
at initialize, upgrade
Posted by ct...@apache.org.
ACCUMULO-3147 Create replication table at initialize, upgrade
Migrate to all replication code to a built-in system-managed replication
table instead of the user-space one. Ensure all replication-related tests
pass. Also fix a small bug in Mock which threw an exception when using
deleteRows with null start and end rows.
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/e840549a
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/e840549a
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/e840549a
Branch: refs/heads/master
Commit: e840549a9c6376c8c5dbe9663e14b4987c10dc5e
Parents: f1a47b3
Author: Christopher Tubbs <ct...@apache.org>
Authored: Sat Nov 1 18:53:14 2014 -0400
Committer: Christopher Tubbs <ct...@apache.org>
Committed: Thu Nov 6 18:39:55 2014 -0500
----------------------------------------------------------------------
.../client/impl/ReplicationOperationsImpl.java | 5 +-
.../accumulo/core/client/mock/MockAccumulo.java | 2 +
.../core/client/mock/MockTableOperations.java | 8 +-
.../replication/PrintReplicationRecords.java | 2 +-
.../core/replication/ReplicationTable.java | 51 ++++-
.../ReplicationTableOfflineException.java | 32 +++
.../ReplicationOperationsImplTest.java | 22 +--
.../apache/accumulo/server/ServerConstants.java | 8 +-
.../server/constraints/MetadataConstraints.java | 119 ++++++-----
.../apache/accumulo/server/init/Initialize.java | 174 ++++++++++------
.../server/replication/ReplicationUtil.java | 149 +-------------
.../server/security/SecurityOperation.java | 3 +-
.../accumulo/server/util/MetadataTableUtil.java | 24 ++-
.../replication/ReplicationTableTest.java | 133 -------------
.../gc/GarbageCollectWriteAheadLogs.java | 40 ++--
.../accumulo/gc/SimpleGarbageCollector.java | 3 +-
.../CloseWriteAheadLogReferences.java | 14 +-
.../gc/GarbageCollectWriteAheadLogsTest.java | 9 +-
.../CloseWriteAheadLogReferencesTest.java | 14 +-
.../accumulo/master/FateServiceHandler.java | 3 +-
.../java/org/apache/accumulo/master/Master.java | 21 +-
.../master/metrics/ReplicationMetrics.java | 5 +-
.../DistributedWorkQueueWorkAssigner.java | 8 +-
.../master/replication/FinishedWorkUpdater.java | 10 +-
.../RemoveCompleteReplicationRecords.java | 8 +-
.../master/replication/StatusMaker.java | 16 +-
.../accumulo/master/replication/WorkMaker.java | 10 +-
.../accumulo/master/util/TableValidators.java | 26 ++-
.../replication/FinishedWorkUpdaterTest.java | 7 -
.../RemoveCompleteReplicationRecordsTest.java | 8 +-
.../replication/SequentialWorkAssignerTest.java | 15 +-
.../replication/UnorderedWorkAssignerTest.java | 10 +-
.../master/replication/WorkMakerTest.java | 15 +-
.../monitor/servlets/ReplicationServlet.java | 5 -
.../replication/ReplicationProcessor.java | 6 +-
.../apache/accumulo/test/LargeSplitRowIT.java | 5 +-
.../test/MasterRepairsDualAssignmentIT.java | 8 +-
.../accumulo/test/MetaConstraintRetryIT.java | 7 +-
.../org/apache/accumulo/test/NamespacesIT.java | 6 +-
.../org/apache/accumulo/test/ShellServerIT.java | 17 +-
.../apache/accumulo/test/WaitForBalanceIT.java | 14 +-
.../replication/MultiInstanceReplicationIT.java | 19 +-
.../test/replication/ReplicationIT.java | 198 +++++++++++++------
.../test/replication/StatusCombinerMacTest.java | 19 +-
.../UnorderedWorkAssignerReplicationIT.java | 11 +-
45 files changed, 612 insertions(+), 677 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/core/src/main/java/org/apache/accumulo/core/client/impl/ReplicationOperationsImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/impl/ReplicationOperationsImpl.java b/core/src/main/java/org/apache/accumulo/core/client/impl/ReplicationOperationsImpl.java
index a8f3f21..cd57c2f 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/ReplicationOperationsImpl.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/ReplicationOperationsImpl.java
@@ -61,7 +61,7 @@ import org.slf4j.LoggerFactory;
import com.google.protobuf.InvalidProtocolBufferException;
/**
- *
+ *
*/
public class ReplicationOperationsImpl implements ReplicationOperations {
private static final Logger log = LoggerFactory.getLogger(ReplicationOperationsImpl.class);
@@ -216,9 +216,6 @@ public class ReplicationOperationsImpl implements ReplicationOperations {
protected Text getTableId(Connector conn, String tableName) throws AccumuloException, AccumuloSecurityException, TableNotFoundException {
TableOperations tops = conn.tableOperations();
- while (!tops.exists(ReplicationTable.NAME)) {
- UtilWaitThread.sleep(200);
- }
if (!conn.tableOperations().exists(tableName)) {
throw new TableNotFoundException(null, tableName, null);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/core/src/main/java/org/apache/accumulo/core/client/mock/MockAccumulo.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/mock/MockAccumulo.java b/core/src/main/java/org/apache/accumulo/core/client/mock/MockAccumulo.java
index 32dbb28..66035f7 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/mock/MockAccumulo.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/mock/MockAccumulo.java
@@ -31,6 +31,7 @@ import org.apache.accumulo.core.client.security.tokens.PasswordToken;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.metadata.MetadataTable;
import org.apache.accumulo.core.metadata.RootTable;
+import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.NamespacePermission;
import org.apache.accumulo.core.security.SystemPermission;
@@ -58,6 +59,7 @@ public class MockAccumulo {
namespaces.put(Namespaces.ACCUMULO_NAMESPACE, new MockNamespace());
createTable("root", RootTable.NAME, true, TimeType.LOGICAL);
createTable("root", MetadataTable.NAME, true, TimeType.LOGICAL);
+ createTable("root", ReplicationTable.NAME, true, TimeType.LOGICAL);
}
public FileSystem getFileSystem() {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/core/src/main/java/org/apache/accumulo/core/client/mock/MockTableOperations.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/mock/MockTableOperations.java b/core/src/main/java/org/apache/accumulo/core/client/mock/MockTableOperations.java
index 08750fe..59afc8b 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/mock/MockTableOperations.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/mock/MockTableOperations.java
@@ -38,8 +38,8 @@ import org.apache.accumulo.core.client.TableExistsException;
import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.client.admin.DiskUsage;
import org.apache.accumulo.core.client.admin.FindMax;
-import org.apache.accumulo.core.client.impl.TableOperationsHelper;
import org.apache.accumulo.core.client.admin.TimeType;
+import org.apache.accumulo.core.client.impl.TableOperationsHelper;
import org.apache.accumulo.core.client.impl.Tables;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.data.Key;
@@ -376,6 +376,10 @@ class MockTableOperations extends TableOperationsHelper {
throw new TableNotFoundException(tableName, tableName, "");
MockTable t = acu.tables.get(tableName);
Text startText = start != null ? new Text(start) : new Text();
+ if (startText.getLength() == 0 && end == null) {
+ t.table.clear();
+ return;
+ }
Text endText = end != null ? new Text(end) : new Text(t.table.lastKey().getRow().getBytes());
startText.append(ZERO, 0, 1);
endText.append(ZERO, 0, 1);
@@ -442,7 +446,7 @@ class MockTableOperations extends TableOperationsHelper {
try {
AccumuloVFSClassLoader.loadClass(className, Class.forName(asTypeName));
} catch (ClassNotFoundException e) {
- log.warn("Could not load class '"+className+"' with type name '"+asTypeName+"' in testClassLoad().", e);
+ log.warn("Could not load class '" + className + "' with type name '" + asTypeName + "' in testClassLoad().", e);
return false;
}
return true;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/core/src/main/java/org/apache/accumulo/core/replication/PrintReplicationRecords.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/replication/PrintReplicationRecords.java b/core/src/main/java/org/apache/accumulo/core/replication/PrintReplicationRecords.java
index ee606b5..35b6df6 100644
--- a/core/src/main/java/org/apache/accumulo/core/replication/PrintReplicationRecords.java
+++ b/core/src/main/java/org/apache/accumulo/core/replication/PrintReplicationRecords.java
@@ -37,7 +37,7 @@ import org.slf4j.LoggerFactory;
import com.google.protobuf.InvalidProtocolBufferException;
/**
- *
+ *
*/
public class PrintReplicationRecords implements Runnable {
private static final Logger log = LoggerFactory.getLogger(PrintReplicationRecords.class);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/core/src/main/java/org/apache/accumulo/core/replication/ReplicationTable.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/replication/ReplicationTable.java b/core/src/main/java/org/apache/accumulo/core/replication/ReplicationTable.java
index 2736762..b6d343f 100644
--- a/core/src/main/java/org/apache/accumulo/core/replication/ReplicationTable.java
+++ b/core/src/main/java/org/apache/accumulo/core/replication/ReplicationTable.java
@@ -20,12 +20,18 @@ import java.util.Collections;
import java.util.Map;
import java.util.Set;
+import org.apache.accumulo.core.client.AccumuloException;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.BatchScanner;
import org.apache.accumulo.core.client.BatchWriter;
import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.client.TableOfflineException;
+import org.apache.accumulo.core.client.impl.Namespaces;
+import org.apache.accumulo.core.client.impl.Tables;
+import org.apache.accumulo.core.master.state.tables.TableState;
import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
import org.apache.accumulo.core.security.Authorizations;
@@ -35,7 +41,8 @@ import com.google.common.collect.ImmutableMap;
public class ReplicationTable {
- public static final String NAME = "replication";
+ public static final String ID = "+rep";
+ public static final String NAME = Namespaces.ACCUMULO_NAMESPACE + ".replication";
public static final String COMBINER_NAME = "statuscombiner";
@@ -46,16 +53,46 @@ public class ReplicationTable {
public static final Map<String,Set<Text>> LOCALITY_GROUPS = ImmutableMap.of(STATUS_LG_NAME, STATUS_LG_COLFAMS, WORK_LG_NAME, WORK_LG_COLFAMS);
public static final String STATUS_FORMATTER_CLASS_NAME = StatusFormatter.class.getName();
- public static Scanner getScanner(Connector conn) throws TableNotFoundException {
- return conn.createScanner(NAME, Authorizations.EMPTY);
+ public static Scanner getScanner(Connector conn) throws ReplicationTableOfflineException {
+ try {
+ return conn.createScanner(NAME, Authorizations.EMPTY);
+ } catch (TableNotFoundException e) {
+ throw new AssertionError(NAME + " should exist, but doesn't.");
+ } catch (TableOfflineException e) {
+ throw new ReplicationTableOfflineException(e);
+ }
}
- public static BatchWriter getBatchWriter(Connector conn) throws TableNotFoundException {
- return conn.createBatchWriter(NAME, new BatchWriterConfig());
+ public static BatchWriter getBatchWriter(Connector conn) throws ReplicationTableOfflineException {
+ try {
+ return conn.createBatchWriter(NAME, new BatchWriterConfig());
+ } catch (TableNotFoundException e) {
+ throw new AssertionError(NAME + " should exist, but doesn't.");
+ } catch (TableOfflineException e) {
+ throw new ReplicationTableOfflineException(e);
+ }
}
- public static BatchScanner getBatchScanner(Connector conn, int queryThreads) throws TableNotFoundException {
- return conn.createBatchScanner(NAME, Authorizations.EMPTY, queryThreads);
+ public static BatchScanner getBatchScanner(Connector conn, int queryThreads) throws ReplicationTableOfflineException {
+ try {
+ return conn.createBatchScanner(NAME, Authorizations.EMPTY, queryThreads);
+ } catch (TableNotFoundException e) {
+ throw new AssertionError(NAME + " should exist, but doesn't.");
+ } catch (TableOfflineException e) {
+ throw new ReplicationTableOfflineException(e);
+ }
+ }
+
+ public static boolean isOnline(Connector conn) {
+ return TableState.ONLINE == Tables.getTableState(conn.getInstance(), ID);
+ }
+
+ public static void setOnline(Connector conn) throws AccumuloSecurityException, AccumuloException {
+ try {
+ conn.tableOperations().online(NAME, true);
+ } catch (TableNotFoundException e) {
+ throw new AssertionError(NAME + " should exist, but doesn't.");
+ }
}
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/core/src/main/java/org/apache/accumulo/core/replication/ReplicationTableOfflineException.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/replication/ReplicationTableOfflineException.java b/core/src/main/java/org/apache/accumulo/core/replication/ReplicationTableOfflineException.java
new file mode 100644
index 0000000..3cc56e0
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/replication/ReplicationTableOfflineException.java
@@ -0,0 +1,32 @@
+/*
+ * 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 org.apache.accumulo.core.client.TableOfflineException;
+
+/**
+ *
+ */
+public class ReplicationTableOfflineException extends Exception {
+
+ private static final long serialVersionUID = 1L;
+
+ public ReplicationTableOfflineException(TableOfflineException e) {
+ super(e);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/core/src/test/java/org/apache/accumulo/core/replication/ReplicationOperationsImplTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/accumulo/core/replication/ReplicationOperationsImplTest.java b/core/src/test/java/org/apache/accumulo/core/replication/ReplicationOperationsImplTest.java
index 54a9f8c..b404d3d 100644
--- a/core/src/test/java/org/apache/accumulo/core/replication/ReplicationOperationsImplTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/replication/ReplicationOperationsImplTest.java
@@ -49,7 +49,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- *
+ *
*/
public class ReplicationOperationsImplTest {
private static final Logger log = LoggerFactory.getLogger(ReplicationOperationsImplTest.class);
@@ -67,14 +67,13 @@ public class ReplicationOperationsImplTest {
@Test
public void waitsUntilEntriesAreReplicated() throws Exception {
Connector conn = inst.getConnector("root", new PasswordToken(""));
- conn.tableOperations().create(ReplicationTable.NAME);
conn.tableOperations().create("foo");
Text tableId = new Text(conn.tableOperations().tableIdMap().get("foo"));
String file1 = "/accumulo/wals/tserver+port/" + UUID.randomUUID(), file2 = "/accumulo/wals/tserver+port/" + UUID.randomUUID();
Status stat = Status.newBuilder().setBegin(0).setEnd(10000).setInfiniteEnd(false).setClosed(false).build();
- BatchWriter bw = conn.createBatchWriter(ReplicationTable.NAME, new BatchWriterConfig());
+ BatchWriter bw = ReplicationTable.getBatchWriter(conn);
Mutation m = new Mutation(file1);
StatusSection.add(m, tableId, ProtobufUtil.toValue(stat));
@@ -136,7 +135,7 @@ public class ReplicationOperationsImplTest {
Assert.assertFalse(done.get());
// Remove the replication entries too
- bw = conn.createBatchWriter(ReplicationTable.NAME, new BatchWriterConfig());
+ bw = ReplicationTable.getBatchWriter(conn);
m = new Mutation(file1);
m.putDelete(StatusSection.NAME, tableId);
bw.addMutation(m);
@@ -163,7 +162,6 @@ public class ReplicationOperationsImplTest {
@Test
public void unrelatedReplicationRecordsDontBlockDrain() throws Exception {
Connector conn = inst.getConnector("root", new PasswordToken(""));
- conn.tableOperations().create(ReplicationTable.NAME);
conn.tableOperations().create("foo");
conn.tableOperations().create("bar");
@@ -173,7 +171,7 @@ public class ReplicationOperationsImplTest {
String file1 = "/accumulo/wals/tserver+port/" + UUID.randomUUID(), file2 = "/accumulo/wals/tserver+port/" + UUID.randomUUID();
Status stat = Status.newBuilder().setBegin(0).setEnd(10000).setInfiniteEnd(false).setClosed(false).build();
- BatchWriter bw = conn.createBatchWriter(ReplicationTable.NAME, new BatchWriterConfig());
+ BatchWriter bw = ReplicationTable.getBatchWriter(conn);
Mutation m = new Mutation(file1);
StatusSection.add(m, tableId1, ProtobufUtil.toValue(stat));
@@ -227,7 +225,7 @@ public class ReplicationOperationsImplTest {
Assert.assertFalse(done.get());
// Remove the replication entries too
- bw = conn.createBatchWriter(ReplicationTable.NAME, new BatchWriterConfig());
+ bw = ReplicationTable.getBatchWriter(conn);
m = new Mutation(file1);
m.putDelete(StatusSection.NAME, tableId1);
bw.addMutation(m);
@@ -247,7 +245,6 @@ public class ReplicationOperationsImplTest {
@Test
public void inprogressReplicationRecordsBlockExecution() throws Exception {
Connector conn = inst.getConnector("root", new PasswordToken(""));
- conn.tableOperations().create(ReplicationTable.NAME);
conn.tableOperations().create("foo");
Text tableId1 = new Text(conn.tableOperations().tableIdMap().get("foo"));
@@ -255,7 +252,7 @@ public class ReplicationOperationsImplTest {
String file1 = "/accumulo/wals/tserver+port/" + UUID.randomUUID();
Status stat = Status.newBuilder().setBegin(0).setEnd(10000).setInfiniteEnd(false).setClosed(false).build();
- BatchWriter bw = conn.createBatchWriter(ReplicationTable.NAME, new BatchWriterConfig());
+ BatchWriter bw = ReplicationTable.getBatchWriter(conn);
Mutation m = new Mutation(file1);
StatusSection.add(m, tableId1, ProtobufUtil.toValue(stat));
@@ -313,7 +310,7 @@ public class ReplicationOperationsImplTest {
Assert.assertFalse(done.get());
// Remove the replication entries too
- bw = conn.createBatchWriter(ReplicationTable.NAME, new BatchWriterConfig());
+ bw = ReplicationTable.getBatchWriter(conn);
m = new Mutation(file1);
m.put(StatusSection.NAME, tableId1, ProtobufUtil.toValue(newStatus));
bw.addMutation(m);
@@ -333,7 +330,6 @@ public class ReplicationOperationsImplTest {
@Test
public void laterCreatedLogsDontBlockExecution() throws Exception {
Connector conn = inst.getConnector("root", new PasswordToken(""));
- conn.tableOperations().create(ReplicationTable.NAME);
conn.tableOperations().create("foo");
Text tableId1 = new Text(conn.tableOperations().tableIdMap().get("foo"));
@@ -341,7 +337,7 @@ public class ReplicationOperationsImplTest {
String file1 = "/accumulo/wals/tserver+port/" + UUID.randomUUID();
Status stat = Status.newBuilder().setBegin(0).setEnd(10000).setInfiniteEnd(false).setClosed(false).build();
- BatchWriter bw = conn.createBatchWriter(ReplicationTable.NAME, new BatchWriterConfig());
+ BatchWriter bw = ReplicationTable.getBatchWriter(conn);
Mutation m = new Mutation(file1);
StatusSection.add(m, tableId1, ProtobufUtil.toValue(stat));
bw.addMutation(m);
@@ -395,7 +391,7 @@ public class ReplicationOperationsImplTest {
System.out.println(e.getKey());
}
- bw = conn.createBatchWriter(ReplicationTable.NAME, new BatchWriterConfig());
+ bw = ReplicationTable.getBatchWriter(conn);
m = new Mutation(file1);
m.putDelete(StatusSection.NAME, tableId1);
bw.addMutation(m);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/base/src/main/java/org/apache/accumulo/server/ServerConstants.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/ServerConstants.java b/server/base/src/main/java/org/apache/accumulo/server/ServerConstants.java
index 6c331e0..9cc9bcb 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/ServerConstants.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/ServerConstants.java
@@ -47,9 +47,13 @@ public class ServerConstants {
public static final Integer WIRE_VERSION = 3;
/**
- * version (7) reflects the change in the representation of trace information in TraceRepo
+ * version (7) also reflects the addition of a replication table
*/
- public static final int DATA_VERSION = 7;
+ public static final int MOVE_TO_REPLICATION_TABLE = 7;
+ /**
+ * this is the current data version
+ */
+ public static final int DATA_VERSION = MOVE_TO_REPLICATION_TABLE;
/**
* version (6) reflects the addition of a separate root table (ACCUMULO-1481) in version 1.6.0
*/
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/base/src/main/java/org/apache/accumulo/server/constraints/MetadataConstraints.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/constraints/MetadataConstraints.java b/server/base/src/main/java/org/apache/accumulo/server/constraints/MetadataConstraints.java
index e4f61f9..487829e 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/constraints/MetadataConstraints.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/constraints/MetadataConstraints.java
@@ -49,48 +49,47 @@ import org.apache.hadoop.io.Text;
import org.apache.log4j.Logger;
public class MetadataConstraints implements Constraint {
-
+
private ZooCache zooCache = null;
private String zooRoot = null;
-
+
private static final Logger log = Logger.getLogger(MetadataConstraints.class);
-
+
private static boolean[] validTableNameChars = new boolean[256];
-
+
{
for (int i = 0; i < 256; i++) {
- validTableNameChars[i] = ((i >= 'a' && i <= 'z') || (i >= '0' && i <= '9')) || i == '!';
+ validTableNameChars[i] = ((i >= 'a' && i <= 'z') || (i >= '0' && i <= '9')) || i == '!' || i == '+';
}
}
-
- private static final HashSet<ColumnFQ> validColumnQuals = new HashSet<ColumnFQ>(Arrays.asList(
- TabletsSection.TabletColumnFamily.PREV_ROW_COLUMN, TabletsSection.TabletColumnFamily.OLD_PREV_ROW_COLUMN,
- TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN, TabletsSection.TabletColumnFamily.SPLIT_RATIO_COLUMN, TabletsSection.ServerColumnFamily.TIME_COLUMN,
- TabletsSection.ServerColumnFamily.LOCK_COLUMN, TabletsSection.ServerColumnFamily.FLUSH_COLUMN, TabletsSection.ServerColumnFamily.COMPACT_COLUMN));
-
- private static final HashSet<Text> validColumnFams = new HashSet<Text>(Arrays.asList(TabletsSection.BulkFileColumnFamily.NAME,
- LogColumnFamily.NAME, ScanFileColumnFamily.NAME, DataFileColumnFamily.NAME,
- TabletsSection.CurrentLocationColumnFamily.NAME, TabletsSection.LastLocationColumnFamily.NAME, TabletsSection.FutureLocationColumnFamily.NAME,
- ChoppedColumnFamily.NAME, ClonedColumnFamily.NAME));
-
+
+ private static final HashSet<ColumnFQ> validColumnQuals = new HashSet<ColumnFQ>(Arrays.asList(TabletsSection.TabletColumnFamily.PREV_ROW_COLUMN,
+ TabletsSection.TabletColumnFamily.OLD_PREV_ROW_COLUMN, TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN,
+ TabletsSection.TabletColumnFamily.SPLIT_RATIO_COLUMN, TabletsSection.ServerColumnFamily.TIME_COLUMN, TabletsSection.ServerColumnFamily.LOCK_COLUMN,
+ TabletsSection.ServerColumnFamily.FLUSH_COLUMN, TabletsSection.ServerColumnFamily.COMPACT_COLUMN));
+
+ private static final HashSet<Text> validColumnFams = new HashSet<Text>(Arrays.asList(TabletsSection.BulkFileColumnFamily.NAME, LogColumnFamily.NAME,
+ ScanFileColumnFamily.NAME, DataFileColumnFamily.NAME, TabletsSection.CurrentLocationColumnFamily.NAME, TabletsSection.LastLocationColumnFamily.NAME,
+ TabletsSection.FutureLocationColumnFamily.NAME, ChoppedColumnFamily.NAME, ClonedColumnFamily.NAME));
+
private static boolean isValidColumn(ColumnUpdate cu) {
-
+
if (validColumnFams.contains(new Text(cu.getColumnFamily())))
return true;
-
+
if (validColumnQuals.contains(new ColumnFQ(cu)))
return true;
-
+
return false;
}
-
+
static private ArrayList<Short> addViolation(ArrayList<Short> lst, int violation) {
if (lst == null)
lst = new ArrayList<Short>();
lst.add((short) violation);
return lst;
}
-
+
static private ArrayList<Short> addIfNotPresent(ArrayList<Short> lst, int intViolation) {
if (lst == null)
return addViolation(lst, intViolation);
@@ -99,38 +98,38 @@ public class MetadataConstraints implements Constraint {
return addViolation(lst, intViolation);
return lst;
}
-
+
@Override
public List<Short> check(Environment env, Mutation mutation) {
-
+
ArrayList<Short> violations = null;
-
+
Collection<ColumnUpdate> colUpdates = mutation.getUpdates();
-
+
// check the row, it should contains at least one ; or end with <
boolean containsSemiC = false;
-
+
byte[] row = mutation.getRow();
-
+
// always allow rows that fall within reserved areas
if (row.length > 0 && row[0] == '~')
return null;
if (row.length > 2 && row[0] == '!' && row[1] == '!' && row[2] == '~')
return null;
-
+
for (byte b : row) {
if (b == ';') {
containsSemiC = true;
}
-
+
if (b == ';' || b == '<')
break;
-
+
if (!validTableNameChars[0xff & b]) {
violations = addIfNotPresent(violations, 4);
}
}
-
+
if (!containsSemiC) {
// see if last row char is <
if (row.length == 0 || row[row.length - 1] != '<') {
@@ -141,38 +140,38 @@ public class MetadataConstraints implements Constraint {
violations = addIfNotPresent(violations, 4);
}
}
-
+
if (row.length > 0 && row[0] == '!') {
if (row.length < 3 || row[1] != '0' || (row[2] != '<' && row[2] != ';')) {
violations = addIfNotPresent(violations, 4);
}
}
-
+
// ensure row is not less than Constants.METADATA_TABLE_ID
if (new Text(row).compareTo(new Text(MetadataTable.ID)) < 0) {
violations = addViolation(violations, 5);
}
-
+
boolean checkedBulk = false;
-
+
for (ColumnUpdate columnUpdate : colUpdates) {
Text columnFamily = new Text(columnUpdate.getColumnFamily());
-
+
if (columnUpdate.isDeleted()) {
if (!isValidColumn(columnUpdate)) {
violations = addViolation(violations, 2);
}
continue;
}
-
+
if (columnUpdate.getValue().length == 0 && !columnFamily.equals(ScanFileColumnFamily.NAME)) {
violations = addViolation(violations, 6);
}
-
+
if (columnFamily.equals(DataFileColumnFamily.NAME)) {
try {
DataFileValue dfv = new DataFileValue(columnUpdate.getValue());
-
+
if (dfv.getSize() < 0 || dfv.getNumEntries() < 0) {
violations = addViolation(violations, 1);
}
@@ -182,7 +181,7 @@ public class MetadataConstraints implements Constraint {
violations = addViolation(violations, 1);
}
} else if (columnFamily.equals(ScanFileColumnFamily.NAME)) {
-
+
} else if (columnFamily.equals(TabletsSection.BulkFileColumnFamily.NAME)) {
if (!columnUpdate.isDeleted() && !checkedBulk) {
// splits, which also write the time reference, are allowed to write this reference even when
@@ -193,13 +192,13 @@ public class MetadataConstraints implements Constraint {
// but it writes everything. We allow it to re-write the bulk information if it is setting the location.
// See ACCUMULO-1230.
boolean isLocationMutation = false;
-
+
HashSet<Text> dataFiles = new HashSet<Text>();
HashSet<Text> loadedFiles = new HashSet<Text>();
String tidString = new String(columnUpdate.getValue(), UTF_8);
int otherTidCount = 0;
-
+
for (ColumnUpdate update : mutation.getUpdates()) {
if (new ColumnFQ(update).equals(TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN)) {
isSplitMutation = true;
@@ -209,16 +208,16 @@ public class MetadataConstraints implements Constraint {
dataFiles.add(new Text(update.getColumnQualifier()));
} else if (new Text(update.getColumnFamily()).equals(TabletsSection.BulkFileColumnFamily.NAME)) {
loadedFiles.add(new Text(update.getColumnQualifier()));
-
+
if (!new String(update.getValue(), UTF_8).equals(tidString)) {
otherTidCount++;
}
}
}
-
+
if (!isSplitMutation && !isLocationMutation) {
long tid = Long.parseLong(tidString);
-
+
try {
if (otherTidCount > 0 || !dataFiles.equals(loadedFiles) || !getArbitrator().transactionAlive(Constants.BULK_ARBITRATOR_TYPE, tid)) {
violations = addViolation(violations, 8);
@@ -227,7 +226,7 @@ public class MetadataConstraints implements Constraint {
violations = addViolation(violations, 8);
}
}
-
+
checkedBulk = true;
}
} else {
@@ -236,11 +235,11 @@ public class MetadataConstraints implements Constraint {
} else if (new ColumnFQ(columnUpdate).equals(TabletsSection.TabletColumnFamily.PREV_ROW_COLUMN) && columnUpdate.getValue().length > 0
&& (violations == null || !violations.contains((short) 4))) {
KeyExtent ke = new KeyExtent(new Text(mutation.getRow()), (Text) null);
-
+
Text per = KeyExtent.decodePrevEndRow(new Value(columnUpdate.getValue()));
-
+
boolean prevEndRowLessThanEndRow = per == null || ke.getEndRow() == null || per.compareTo(ke.getEndRow()) < 0;
-
+
if (!prevEndRowLessThanEndRow) {
violations = addViolation(violations, 3);
}
@@ -248,28 +247,28 @@ public class MetadataConstraints implements Constraint {
if (zooCache == null) {
zooCache = new ZooCache();
}
-
+
if (zooRoot == null) {
zooRoot = ZooUtil.getRoot(HdfsZooInstance.getInstance());
}
-
+
boolean lockHeld = false;
String lockId = new String(columnUpdate.getValue(), UTF_8);
-
+
try {
lockHeld = ZooLock.isLockHeld(zooCache, new ZooUtil.LockID(zooRoot, lockId));
} catch (Exception e) {
log.debug("Failed to verify lock was held " + lockId + " " + e.getMessage());
}
-
+
if (!lockHeld) {
violations = addViolation(violations, 7);
}
}
-
+
}
}
-
+
if (violations != null) {
log.debug("violating metadata mutation : " + new String(mutation.getRow(), UTF_8));
for (ColumnUpdate update : mutation.getUpdates()) {
@@ -277,14 +276,14 @@ public class MetadataConstraints implements Constraint {
+ (update.isDeleted() ? "[delete]" : new String(update.getValue(), UTF_8)));
}
}
-
+
return violations;
}
-
+
protected Arbitrator getArbitrator() {
return new ZooArbitrator();
}
-
+
@Override
public String getViolationDescription(short violationCode) {
switch (violationCode) {
@@ -307,11 +306,11 @@ public class MetadataConstraints implements Constraint {
}
return null;
}
-
+
@Override
protected void finalize() {
if (zooCache != null)
zooCache.clear();
}
-
+
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/base/src/main/java/org/apache/accumulo/server/init/Initialize.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/init/Initialize.java b/server/base/src/main/java/org/apache/accumulo/server/init/Initialize.java
index 45e883d..24ff637 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/init/Initialize.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/init/Initialize.java
@@ -25,10 +25,13 @@ import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
+import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Locale;
import java.util.Map.Entry;
+import java.util.Set;
+import java.util.TreeMap;
import java.util.UUID;
import jline.console.ConsoleReader;
@@ -57,10 +60,20 @@ import org.apache.accumulo.core.metadata.MetadataTable;
import org.apache.accumulo.core.metadata.RootTable;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.ReplicationSection;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.CurrentLocationColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.DataFileColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.FutureLocationColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.LogColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ServerColumnFamily;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.TabletColumnFamily;
import org.apache.accumulo.core.replication.ReplicationConstants;
+import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
+import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
+import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.security.SecurityUtil;
import org.apache.accumulo.core.util.CachedConfiguration;
import org.apache.accumulo.core.util.ColumnFQ;
+import org.apache.accumulo.core.util.LocalityGroupUtil;
import org.apache.accumulo.core.volume.VolumeConfiguration;
import org.apache.accumulo.core.zookeeper.ZooUtil;
import org.apache.accumulo.fate.zookeeper.IZooReaderWriter;
@@ -91,6 +104,7 @@ import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs.Ids;
import com.beust.jcommander.Parameter;
+import com.google.common.base.Joiner;
/**
* This class is used to setup the directory structure and the root tablet to get an instance started
@@ -99,7 +113,7 @@ import com.beust.jcommander.Parameter;
public class Initialize {
private static final Logger log = Logger.getLogger(Initialize.class);
private static final String DEFAULT_ROOT_USER = "root";
- public static final String TABLE_TABLETS_TABLET_DIR = "/table_info";
+ private static final String TABLE_TABLETS_TABLET_DIR = "/table_info";
private static ConsoleReader reader = null;
private static IZooReaderWriter zoo = ZooReaderWriter.getInstance();
@@ -130,6 +144,8 @@ public class Initialize {
}
private static HashMap<String,String> initialMetadataConf = new HashMap<String,String>();
+ private static HashMap<String,String> initialMetadataCombinerConf = new HashMap<String,String>();
+ private static HashMap<String,String> initialReplicationTableConf = new HashMap<String,String>();
static {
initialMetadataConf.put(Property.TABLE_FILE_COMPRESSED_BLOCK_SIZE.getKey(), "32K");
initialMetadataConf.put(Property.TABLE_FILE_REPLICATION.getKey(), "5");
@@ -146,13 +162,44 @@ public class Initialize {
initialMetadataConf.put(Property.TABLE_ITERATOR_PREFIX.getKey() + "majc.bulkLoadFilter", "20," + MetadataBulkLoadFilter.class.getName());
initialMetadataConf.put(Property.TABLE_FAILURES_IGNORE.getKey(), "false");
initialMetadataConf.put(Property.TABLE_LOCALITY_GROUP_PREFIX.getKey() + "tablet",
- String.format("%s,%s", TabletsSection.TabletColumnFamily.NAME, TabletsSection.CurrentLocationColumnFamily.NAME));
- initialMetadataConf.put(Property.TABLE_LOCALITY_GROUP_PREFIX.getKey() + "server", String.format("%s,%s,%s,%s", TabletsSection.DataFileColumnFamily.NAME,
- TabletsSection.LogColumnFamily.NAME, TabletsSection.ServerColumnFamily.NAME, TabletsSection.FutureLocationColumnFamily.NAME));
+ String.format("%s,%s", TabletColumnFamily.NAME, CurrentLocationColumnFamily.NAME));
+ initialMetadataConf.put(Property.TABLE_LOCALITY_GROUP_PREFIX.getKey() + "server",
+ String.format("%s,%s,%s,%s", DataFileColumnFamily.NAME, LogColumnFamily.NAME, ServerColumnFamily.NAME, FutureLocationColumnFamily.NAME));
initialMetadataConf.put(Property.TABLE_LOCALITY_GROUPS.getKey(), "tablet,server");
initialMetadataConf.put(Property.TABLE_DEFAULT_SCANTIME_VISIBILITY.getKey(), "");
initialMetadataConf.put(Property.TABLE_INDEXCACHE_ENABLED.getKey(), "true");
initialMetadataConf.put(Property.TABLE_BLOCKCACHE_ENABLED.getKey(), "true");
+
+ // ACCUMULO-3077 Set the combiner on accumulo.metadata during init to reduce the likelihood of a race
+ // condition where a tserver compacts away Status updates because it didn't see the Combiner configured
+ IteratorSetting setting = new IteratorSetting(9, ReplicationTableUtil.COMBINER_NAME, StatusCombiner.class);
+ Combiner.setColumns(setting, Collections.singletonList(new Column(ReplicationSection.COLF)));
+ for (IteratorScope scope : IteratorScope.values()) {
+ String root = String.format("%s%s.%s", Property.TABLE_ITERATOR_PREFIX, scope.name().toLowerCase(), setting.getName());
+ for (Entry<String,String> prop : setting.getOptions().entrySet()) {
+ initialMetadataCombinerConf.put(root + ".opt." + prop.getKey(), prop.getValue());
+ }
+ initialMetadataCombinerConf.put(root, setting.getPriority() + "," + setting.getIteratorClass());
+ }
+
+ // add combiners to replication table
+ setting = new IteratorSetting(30, ReplicationTable.COMBINER_NAME, StatusCombiner.class);
+ setting.setPriority(30);
+ Combiner.setColumns(setting, Arrays.asList(new Column(StatusSection.NAME), new Column(WorkSection.NAME)));
+ for (IteratorScope scope : EnumSet.allOf(IteratorScope.class)) {
+ String root = String.format("%s%s.%s", Property.TABLE_ITERATOR_PREFIX, scope.name().toLowerCase(), setting.getName());
+ for (Entry<String,String> prop : setting.getOptions().entrySet()) {
+ initialReplicationTableConf.put(root + ".opt." + prop.getKey(), prop.getValue());
+ }
+ initialReplicationTableConf.put(root, setting.getPriority() + "," + setting.getIteratorClass());
+ }
+ // add locality groups to replication table
+ for (Entry<String,Set<Text>> g : ReplicationTable.LOCALITY_GROUPS.entrySet()) {
+ initialReplicationTableConf.put(Property.TABLE_LOCALITY_GROUP_PREFIX + g.getKey(), LocalityGroupUtil.encodeColumnFamilies(g.getValue()));
+ }
+ initialReplicationTableConf.put(Property.TABLE_LOCALITY_GROUPS.getKey(), Joiner.on(",").join(ReplicationTable.LOCALITY_GROUPS.keySet()));
+ // add formatter to replication table
+ initialReplicationTableConf.put(Property.TABLE_FORMATTER_CLASS.getKey(), ReplicationTable.STATUS_FORMATTER_CLASS_NAME);
}
static boolean checkInit(Configuration conf, VolumeManager fs, SiteConfiguration sconf) throws IOException {
@@ -305,60 +352,79 @@ public class Initialize {
private static void initFileSystem(Opts opts, VolumeManager fs, UUID uuid, String rootTabletDir) throws IOException {
initDirs(fs, uuid, VolumeConfiguration.getVolumeUris(SiteConfiguration.getInstance()), false);
- // initialize initial metadata config in zookeeper
- initMetadataConfig();
+ // initialize initial system tables config in zookeeper
+ initSystemTablesConfig();
String tableMetadataTabletDir = fs.choose(ServerConstants.getBaseUris()) + Constants.HDFS_TABLES_DIR + Path.SEPARATOR + MetadataTable.ID
+ TABLE_TABLETS_TABLET_DIR;
+ String replicationTableDefaultTabletDir = fs.choose(ServerConstants.getBaseUris()) + Constants.HDFS_TABLES_DIR + Path.SEPARATOR + ReplicationTable.ID
+ + Constants.DEFAULT_TABLET_LOCATION;
+
String defaultMetadataTabletDir = fs.choose(ServerConstants.getBaseUris()) + Constants.HDFS_TABLES_DIR + Path.SEPARATOR + MetadataTable.ID
+ Constants.DEFAULT_TABLET_LOCATION;
// create table and default tablets directories
- createDirectories(fs, rootTabletDir, tableMetadataTabletDir, defaultMetadataTabletDir);
+ createDirectories(fs, rootTabletDir, tableMetadataTabletDir, defaultMetadataTabletDir, replicationTableDefaultTabletDir);
+
+ String ext = FileOperations.getNewFileExtension(AccumuloConfiguration.getDefaultConfiguration());
- // populate the root tablet with info about the metadata tablets
- String fileName = rootTabletDir + Path.SEPARATOR + "00000_00000." + FileOperations.getNewFileExtension(AccumuloConfiguration.getDefaultConfiguration());
- createMetadataFile(fs, fileName, MetadataTable.ID, tableMetadataTabletDir, defaultMetadataTabletDir);
+ // populate the metadata tables tablet with info about the replication table's one initial tablet
+ String metadataFileName = tableMetadataTabletDir + Path.SEPARATOR + "0_1." + ext;
+ Tablet replicationTablet = new Tablet(ReplicationTable.ID, replicationTableDefaultTabletDir, null, null);
+ createMetadataFile(fs, metadataFileName, replicationTablet);
+
+ // populate the root tablet with info about the metadata table's two initial tablets
+ String rootTabletFileName = rootTabletDir + Path.SEPARATOR + "00000_00000." + ext;
+ Text splitPoint = TabletsSection.getRange().getEndKey().getRow();
+ Tablet tablesTablet = new Tablet(MetadataTable.ID, tableMetadataTabletDir, null, splitPoint, metadataFileName);
+ Tablet defaultTablet = new Tablet(MetadataTable.ID, defaultMetadataTabletDir, splitPoint, null);
+ createMetadataFile(fs, rootTabletFileName, tablesTablet, defaultTablet);
}
- /**
- * Create an rfile in the default tablet's directory for a new table. This method is used to create the initial root tablet contents, with information about
- * the metadata table's tablets
- *
- * @param volmanager
- * The VolumeManager
- * @param fileName
- * The location to create the file
- * @param tableId
- * TableID that is being "created"
- * @param tableTabletDir
- * The table_info directory for the new table
- * @param defaultTabletDir
- * The default_tablet directory for the new table
- */
- private static void createMetadataFile(VolumeManager volmanager, String fileName, String tableId, String tableTabletDir, String defaultTabletDir)
- throws IOException {
+ private static class Tablet {
+ String tableId, dir;
+ Text prevEndRow, endRow;
+ String[] files;
+
+ Tablet(String tableId, String dir, Text prevEndRow, Text endRow, String... files) {
+ this.tableId = tableId;
+ this.dir = dir;
+ this.prevEndRow = prevEndRow;
+ this.endRow = endRow;
+ this.files = files;
+ }
+ }
+
+ private static void createMetadataFile(VolumeManager volmanager, String fileName, Tablet... tablets) throws IOException {
+ // sort file contents in memory, then play back to the file
+ TreeMap<Key,Value> sorted = new TreeMap<>();
+ for (Tablet tablet : tablets) {
+ createEntriesForTablet(sorted, tablet);
+ }
FileSystem fs = volmanager.getVolumeByPath(new Path(fileName)).getFileSystem();
FileSKVWriter tabletWriter = FileOperations.getInstance().openWriter(fileName, fs, fs.getConf(), AccumuloConfiguration.getDefaultConfiguration());
tabletWriter.startDefaultLocalityGroup();
- Text splitPoint = TabletsSection.getRange().getEndKey().getRow();
- createEntriesForTablet(tabletWriter, tableId, tableTabletDir, null, splitPoint);
- createEntriesForTablet(tabletWriter, tableId, defaultTabletDir, splitPoint, null);
+ for (Entry<Key,Value> entry : sorted.entrySet()) {
+ tabletWriter.append(entry.getKey(), entry.getValue());
+ }
tabletWriter.close();
}
- private static void createEntriesForTablet(FileSKVWriter writer, String tableId, String tabletDir, Text tabletPrevEndRow, Text tabletEndRow)
- throws IOException {
- Text extent = new Text(KeyExtent.getMetadataEntry(new Text(tableId), tabletEndRow));
- addEntry(writer, extent, DIRECTORY_COLUMN, new Value(tabletDir.getBytes(UTF_8)));
- addEntry(writer, extent, TIME_COLUMN, new Value((TabletTime.LOGICAL_TIME_ID + "0").getBytes(UTF_8)));
- addEntry(writer, extent, PREV_ROW_COLUMN, KeyExtent.encodePrevEndRow(tabletPrevEndRow));
+ private static void createEntriesForTablet(TreeMap<Key,Value> map, Tablet tablet) {
+ Value EMPTY_SIZE = new Value("0,0".getBytes(UTF_8));
+ Text extent = new Text(KeyExtent.getMetadataEntry(new Text(tablet.tableId), tablet.endRow));
+ addEntry(map, extent, DIRECTORY_COLUMN, new Value(tablet.dir.getBytes(UTF_8)));
+ addEntry(map, extent, TIME_COLUMN, new Value((TabletTime.LOGICAL_TIME_ID + "0").getBytes(UTF_8)));
+ addEntry(map, extent, PREV_ROW_COLUMN, KeyExtent.encodePrevEndRow(tablet.prevEndRow));
+ for (String file : tablet.files) {
+ addEntry(map, extent, new ColumnFQ(DataFileColumnFamily.NAME, new Text(file)), EMPTY_SIZE);
+ }
}
- private static void addEntry(FileSKVWriter writer, Text row, ColumnFQ col, Value value) throws IOException {
- writer.append(new Key(row, col.getColumnFamily(), col.getColumnQualifier(), 0), value);
+ private static void addEntry(TreeMap<Key,Value> map, Text row, ColumnFQ col, Value value) {
+ map.put(new Key(row, col.getColumnFamily(), col.getColumnQualifier(), 0), value);
}
private static void createDirectories(VolumeManager fs, String... dirs) throws IOException {
@@ -404,6 +470,8 @@ public class Initialize {
TableManager.prepareNewNamespaceState(uuid, Namespaces.ACCUMULO_NAMESPACE_ID, Namespaces.ACCUMULO_NAMESPACE, NodeExistsPolicy.FAIL);
TableManager.prepareNewTableState(uuid, RootTable.ID, Namespaces.ACCUMULO_NAMESPACE_ID, RootTable.NAME, TableState.ONLINE, NodeExistsPolicy.FAIL);
TableManager.prepareNewTableState(uuid, MetadataTable.ID, Namespaces.ACCUMULO_NAMESPACE_ID, MetadataTable.NAME, TableState.ONLINE, NodeExistsPolicy.FAIL);
+ TableManager.prepareNewTableState(uuid, ReplicationTable.ID, Namespaces.ACCUMULO_NAMESPACE_ID, ReplicationTable.NAME, TableState.OFFLINE,
+ NodeExistsPolicy.FAIL);
zoo.putPersistentData(zkInstanceRoot + Constants.ZTSERVERS, EMPTY_BYTE_ARRAY, NodeExistsPolicy.FAIL);
zoo.putPersistentData(zkInstanceRoot + Constants.ZPROBLEMS, EMPTY_BYTE_ARRAY, NodeExistsPolicy.FAIL);
zoo.putPersistentData(zkInstanceRoot + RootTable.ZROOT_TABLET, EMPTY_BYTE_ARRAY, NodeExistsPolicy.FAIL);
@@ -412,8 +480,7 @@ public class Initialize {
zoo.putPersistentData(zkInstanceRoot + Constants.ZTRACERS, EMPTY_BYTE_ARRAY, NodeExistsPolicy.FAIL);
zoo.putPersistentData(zkInstanceRoot + Constants.ZMASTERS, EMPTY_BYTE_ARRAY, NodeExistsPolicy.FAIL);
zoo.putPersistentData(zkInstanceRoot + Constants.ZMASTER_LOCK, EMPTY_BYTE_ARRAY, NodeExistsPolicy.FAIL);
- zoo.putPersistentData(zkInstanceRoot + Constants.ZMASTER_GOAL_STATE, MasterGoalState.NORMAL.toString().getBytes(UTF_8),
- NodeExistsPolicy.FAIL);
+ zoo.putPersistentData(zkInstanceRoot + Constants.ZMASTER_GOAL_STATE, MasterGoalState.NORMAL.toString().getBytes(UTF_8), NodeExistsPolicy.FAIL);
zoo.putPersistentData(zkInstanceRoot + Constants.ZGC, EMPTY_BYTE_ARRAY, NodeExistsPolicy.FAIL);
zoo.putPersistentData(zkInstanceRoot + Constants.ZGC_LOCK, EMPTY_BYTE_ARRAY, NodeExistsPolicy.FAIL);
zoo.putPersistentData(zkInstanceRoot + Constants.ZCONFIG, EMPTY_BYTE_ARRAY, NodeExistsPolicy.FAIL);
@@ -484,7 +551,7 @@ public class Initialize {
opts.rootpass);
}
- public static void initMetadataConfig() throws IOException {
+ public static void initSystemTablesConfig() throws IOException {
try {
Configuration conf = CachedConfiguration.getInstance();
int max = conf.getInt("dfs.replication.max", 512);
@@ -500,21 +567,16 @@ public class Initialize {
if (!TablePropUtil.setTableProperty(MetadataTable.ID, entry.getKey(), entry.getValue()))
throw new IOException("Cannot create per-table property " + entry.getKey());
}
- } catch (Exception e) {
- log.fatal("error talking to zookeeper", e);
- throw new IOException(e);
- }
- // ACCUMULO-3077 Set the combiner on accumulo.metadata during init to reduce the likelihood of a race
- // condition where a tserver compacts away Status updates because it didn't see the Combiner configured
- IteratorSetting setting = new IteratorSetting(9, ReplicationTableUtil.COMBINER_NAME, StatusCombiner.class);
- Combiner.setColumns(setting, Collections.singletonList(new Column(ReplicationSection.COLF)));
- try {
- for (IteratorScope scope : IteratorScope.values()) {
- String root = String.format("%s%s.%s", Property.TABLE_ITERATOR_PREFIX, scope.name().toLowerCase(), setting.getName());
- for (Entry<String,String> prop : setting.getOptions().entrySet()) {
- TablePropUtil.setTableProperty(MetadataTable.ID, root + ".opt." + prop.getKey(), prop.getValue());
- }
- TablePropUtil.setTableProperty(MetadataTable.ID, root, setting.getPriority() + "," + setting.getIteratorClass());
+ // Only add combiner config to accumulo.metadata table (ACCUMULO-3077)
+ for (Entry<String,String> entry : initialMetadataCombinerConf.entrySet()) {
+ if (!TablePropUtil.setTableProperty(MetadataTable.ID, entry.getKey(), entry.getValue()))
+ throw new IOException("Cannot create per-table property " + entry.getKey());
+ }
+
+ // add configuration to the replication table
+ for (Entry<String,String> entry : initialReplicationTableConf.entrySet()) {
+ if (!TablePropUtil.setTableProperty(ReplicationTable.ID, entry.getKey(), entry.getValue()))
+ throw new IOException("Cannot create per-table property " + entry.getKey());
}
} catch (Exception e) {
log.fatal("Error talking to ZooKeeper", e);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/base/src/main/java/org/apache/accumulo/server/replication/ReplicationUtil.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/replication/ReplicationUtil.java b/server/base/src/main/java/org/apache/accumulo/server/replication/ReplicationUtil.java
index e05a08c..50b15f5 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/replication/ReplicationUtil.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/replication/ReplicationUtil.java
@@ -18,9 +18,7 @@ package org.apache.accumulo.server.replication;
import static java.nio.charset.StandardCharsets.UTF_8;
-import java.util.Arrays;
import java.util.Collections;
-import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
@@ -32,10 +30,7 @@ import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.BatchScanner;
import org.apache.accumulo.core.client.Connector;
-import org.apache.accumulo.core.client.IteratorSetting;
-import org.apache.accumulo.core.client.IteratorSetting.Column;
import org.apache.accumulo.core.client.Scanner;
-import org.apache.accumulo.core.client.TableExistsException;
import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.client.admin.TableOperations;
import org.apache.accumulo.core.client.replication.ReplicaSystem;
@@ -44,20 +39,16 @@ import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
-import org.apache.accumulo.core.iterators.Combiner;
-import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
import org.apache.accumulo.core.master.thrift.MasterMonitorInfo;
import org.apache.accumulo.core.metadata.MetadataTable;
import org.apache.accumulo.core.metadata.RootTable;
-import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
import org.apache.accumulo.core.replication.ReplicationTable;
+import org.apache.accumulo.core.replication.ReplicationTableOfflineException;
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.Authorizations;
-import org.apache.accumulo.core.security.TablePermission;
-import org.apache.accumulo.fate.util.UtilWaitThread;
import org.apache.accumulo.server.zookeeper.ZooCache;
import org.apache.hadoop.io.Text;
import org.slf4j.Logger;
@@ -243,9 +234,9 @@ public class ReplicationUtil {
if (null != path) {
Scanner s;
try {
- s = conn.createScanner(ReplicationTable.NAME, Authorizations.EMPTY);
- } catch (TableNotFoundException e) {
- log.debug("Replication table no long exists", e);
+ s = ReplicationTable.getScanner(conn);
+ } catch (ReplicationTableOfflineException e) {
+ log.debug("Replication table no longer online", e);
return status;
}
@@ -294,136 +285,4 @@ public class ReplicationUtil {
return newMap;
}
- public static synchronized void createReplicationTable(Connector conn) {
- TableOperations tops = conn.tableOperations();
- if (tops.exists(ReplicationTable.NAME)) {
- if (configureReplicationTable(conn)) {
- return;
- }
- }
-
- for (int i = 0; i < 5; i++) {
- try {
- if (!tops.exists(ReplicationTable.NAME)) {
- tops.create(ReplicationTable.NAME, false);
- }
- break;
- } catch (AccumuloException | AccumuloSecurityException e) {
- log.error("Failed to create replication table", e);
- } catch (TableExistsException e) {
- // Shouldn't happen unless someone else made the table
- }
- log.error("Retrying table creation in 1 second...");
- UtilWaitThread.sleep(1000);
- }
-
- for (int i = 0; i < 5; i++) {
- if (configureReplicationTable(conn)) {
- return;
- }
-
- log.error("Failed to configure the replication table, retying...");
- UtilWaitThread.sleep(1000);
- }
-
- throw new RuntimeException("Could not configure replication table");
- }
-
- /**
- * Attempts to configure the replication table, will return false if it fails
- *
- * @param conn
- * Connector for the instance
- * @return True if the replication table is properly configured
- */
- protected static synchronized boolean configureReplicationTable(Connector conn) {
- try {
- conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.READ);
- } catch (AccumuloException | AccumuloSecurityException e) {
- log.warn("Could not grant root user read access to replication table", e);
- // Should this be fatal? It's only for convenience, all r/w is done by !SYSTEM
- }
-
- TableOperations tops = conn.tableOperations();
- Map<String,EnumSet<IteratorScope>> iterators = null;
- try {
- iterators = tops.listIterators(ReplicationTable.NAME);
- } catch (AccumuloSecurityException | AccumuloException | TableNotFoundException e) {
- log.error("Could not fetch iterators for " + ReplicationTable.NAME, e);
- return false;
- }
-
- if (!iterators.containsKey(ReplicationTable.COMBINER_NAME)) {
- // Set our combiner and combine all columns
- IteratorSetting setting = new IteratorSetting(30, ReplicationTable.COMBINER_NAME, StatusCombiner.class);
- Combiner.setColumns(setting, Arrays.asList(new Column(StatusSection.NAME), new Column(WorkSection.NAME)));
- try {
- tops.attachIterator(ReplicationTable.NAME, setting);
- } catch (AccumuloSecurityException | AccumuloException | TableNotFoundException e) {
- log.error("Could not set StatusCombiner on replication table", e);
- return false;
- }
- }
-
- Map<String,Set<Text>> localityGroups;
- try {
- localityGroups = tops.getLocalityGroups(ReplicationTable.NAME);
- } catch (TableNotFoundException | AccumuloException e) {
- log.error("Could not fetch locality groups", e);
- return false;
- }
-
- Set<Text> statusColfams = localityGroups.get(ReplicationTable.STATUS_LG_NAME), workColfams = localityGroups.get(ReplicationTable.WORK_LG_NAME);
- if (null == statusColfams || null == workColfams) {
- try {
- tops.setLocalityGroups(ReplicationTable.NAME, ReplicationTable.LOCALITY_GROUPS);
- } catch (AccumuloException | AccumuloSecurityException | TableNotFoundException e) {
- log.error("Could not set locality groups on replication table", e);
- return false;
- }
- }
-
- // Make sure the StatusFormatter is set on the metadata table
- Iterable<Entry<String,String>> properties;
- try {
- properties = tops.getProperties(ReplicationTable.NAME);
- } catch (AccumuloException | TableNotFoundException e) {
- log.error("Could not fetch table properties on replication table", e);
- return false;
- }
-
- boolean formatterConfigured = false;
- for (Entry<String,String> property : properties) {
- if (Property.TABLE_FORMATTER_CLASS.getKey().equals(property.getKey())) {
- if (!ReplicationTable.STATUS_FORMATTER_CLASS_NAME.equals(property.getValue())) {
- log.info("Changing formatter for {} table from {} to {}", ReplicationTable.NAME, property.getValue(), ReplicationTable.STATUS_FORMATTER_CLASS_NAME);
- try {
- tops.setProperty(ReplicationTable.NAME, Property.TABLE_FORMATTER_CLASS.getKey(), ReplicationTable.STATUS_FORMATTER_CLASS_NAME);
- } catch (AccumuloException | AccumuloSecurityException e) {
- log.error("Could not set formatter on replication table", e);
- return false;
- }
- }
-
- formatterConfigured = true;
-
- // Don't need to keep iterating over the properties after we found the one we were looking for
- break;
- }
- }
-
- if (!formatterConfigured) {
- try {
- tops.setProperty(ReplicationTable.NAME, Property.TABLE_FORMATTER_CLASS.getKey(), ReplicationTable.STATUS_FORMATTER_CLASS_NAME);
- } catch (AccumuloException | AccumuloSecurityException e) {
- log.error("Could not set formatter on replication table", e);
- return false;
- }
- }
-
- log.debug("Successfully configured replication table");
-
- return true;
- }
-
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/base/src/main/java/org/apache/accumulo/server/security/SecurityOperation.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/security/SecurityOperation.java b/server/base/src/main/java/org/apache/accumulo/server/security/SecurityOperation.java
index fa90b3e..71dcdcf 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/security/SecurityOperation.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/security/SecurityOperation.java
@@ -41,6 +41,7 @@ import org.apache.accumulo.core.data.thrift.TRange;
import org.apache.accumulo.core.master.thrift.FateOperation;
import org.apache.accumulo.core.metadata.MetadataTable;
import org.apache.accumulo.core.metadata.RootTable;
+import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.Credentials;
import org.apache.accumulo.core.security.NamespacePermission;
@@ -296,7 +297,7 @@ public class SecurityOperation {
protected boolean _hasTablePermission(String user, String table, TablePermission permission, boolean useCached) throws ThriftSecurityException {
targetUserExists(user);
- if ((table.equals(MetadataTable.ID) || table.equals(RootTable.ID)) && permission.equals(TablePermission.READ))
+ if ((table.equals(MetadataTable.ID) || table.equals(RootTable.ID) || table.equals(ReplicationTable.ID)) && permission.equals(TablePermission.READ))
return true;
try {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java
----------------------------------------------------------------------
diff --git 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
index 8f84169..903142b 100644
--- 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
@@ -17,6 +17,9 @@
package org.apache.accumulo.server.util;
import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ServerColumnFamily.DIRECTORY_COLUMN;
+import static org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ServerColumnFamily.TIME_COLUMN;
+import static org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.TabletColumnFamily.PREV_ROW_COLUMN;
import java.io.IOException;
import java.util.ArrayList;
@@ -64,6 +67,7 @@ import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.Da
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.LogColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ScanFileColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ServerColumnFamily;
+import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.Credentials;
import org.apache.accumulo.core.tabletserver.log.LogEntry;
@@ -82,6 +86,7 @@ import org.apache.accumulo.server.fs.FileRef;
import org.apache.accumulo.server.fs.VolumeManager;
import org.apache.accumulo.server.fs.VolumeManagerImpl;
import org.apache.accumulo.server.security.SystemCredentials;
+import org.apache.accumulo.server.tablets.TabletTime;
import org.apache.accumulo.server.zookeeper.ZooLock;
import org.apache.accumulo.server.zookeeper.ZooReaderWriter;
import org.apache.hadoop.fs.FileStatus;
@@ -132,10 +137,10 @@ public class MetadataTableUtil {
public static void update(Credentials credentials, ZooLock zooLock, Mutation m, KeyExtent extent) {
Writer t = extent.isMeta() ? getRootTable(credentials) : getMetadataTable(credentials);
- update(t, credentials, zooLock, m);
+ update(t, zooLock, m);
}
- public static void update(Writer t, Credentials credentials, ZooLock zooLock, Mutation m) {
+ public static void update(Writer t, ZooLock zooLock, Mutation m) {
if (zooLock != null)
putLockID(zooLock, m);
while (true) {
@@ -971,6 +976,20 @@ public class MetadataTableUtil {
}
/**
+ * During an upgrade from 1.6 to 1.7, we need to add the replication table
+ */
+ public static void createReplicationTable(Instance instance, SystemCredentials systemCredentials) throws IOException {
+ String dir = VolumeManagerImpl.get().choose(ServerConstants.getBaseUris()) + Constants.HDFS_TABLES_DIR + Path.SEPARATOR + ReplicationTable.ID
+ + Constants.DEFAULT_TABLET_LOCATION;
+
+ Mutation m = new Mutation(new Text(KeyExtent.getMetadataEntry(new Text(ReplicationTable.ID), null)));
+ m.put(DIRECTORY_COLUMN.getColumnFamily(), DIRECTORY_COLUMN.getColumnQualifier(), 0, new Value(dir.getBytes(UTF_8)));
+ m.put(TIME_COLUMN.getColumnFamily(), TIME_COLUMN.getColumnQualifier(), 0, new Value((TabletTime.LOGICAL_TIME_ID + "0").getBytes(UTF_8)));
+ m.put(PREV_ROW_COLUMN.getColumnFamily(), PREV_ROW_COLUMN.getColumnQualifier(), 0, KeyExtent.encodePrevEndRow(null));
+ update(getMetadataTable(systemCredentials), null, m);
+ }
+
+ /**
* During an upgrade we need to move deletion requests for files under the !METADATA table to the root tablet.
*/
public static void moveMetaDeleteMarkers(Instance instance, Credentials creds) {
@@ -988,7 +1007,6 @@ public class MetadataTableUtil {
break;
}
}
-
}
public static void moveMetaDeleteMarkersFrom14(Instance instance, Credentials creds) {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/base/src/test/java/org/apache/accumulo/server/replication/ReplicationTableTest.java
----------------------------------------------------------------------
diff --git a/server/base/src/test/java/org/apache/accumulo/server/replication/ReplicationTableTest.java b/server/base/src/test/java/org/apache/accumulo/server/replication/ReplicationTableTest.java
deleted file mode 100644
index 44ac3fe..0000000
--- a/server/base/src/test/java/org/apache/accumulo/server/replication/ReplicationTableTest.java
+++ /dev/null
@@ -1,133 +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.server.replication;
-
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.EnumSet;
-import java.util.Map;
-import java.util.Set;
-
-import org.apache.accumulo.core.client.Connector;
-import org.apache.accumulo.core.client.IteratorSetting;
-import org.apache.accumulo.core.client.IteratorSetting.Column;
-import org.apache.accumulo.core.client.admin.TableOperations;
-import org.apache.accumulo.core.client.mock.MockInstance;
-import org.apache.accumulo.core.client.security.tokens.PasswordToken;
-import org.apache.accumulo.core.iterators.Combiner;
-import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
-import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
-import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
-import org.apache.accumulo.core.replication.ReplicationTable;
-import org.apache.hadoop.io.Text;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TestName;
-
-import com.google.common.collect.ImmutableMap;
-
-/**
- *
- */
-public class ReplicationTableTest {
-
- @Rule
- public TestName testName = new TestName();
-
- private MockInstance instance;
- private Connector conn;
-
- @Before
- public void setupMockAccumulo() throws Exception {
- instance = new MockInstance(testName.getMethodName());
- conn = instance.getConnector("root", new PasswordToken(""));
- }
-
- @Test
- public void replicationTableCreated() {
- TableOperations tops = conn.tableOperations();
- Assert.assertFalse(tops.exists(ReplicationTable.NAME));
- ReplicationUtil.createReplicationTable(conn);
- Assert.assertTrue(tops.exists(ReplicationTable.NAME));
- }
-
- @Test
- public void replicationTableDoubleCreate() {
- TableOperations tops = conn.tableOperations();
- Assert.assertFalse(tops.exists(ReplicationTable.NAME));
-
- // Create the table
- ReplicationUtil.createReplicationTable(conn);
-
- // Make sure it exists and save off the id
- Assert.assertTrue(tops.exists(ReplicationTable.NAME));
- String tableId = tops.tableIdMap().get(ReplicationTable.NAME);
-
- // Try to make it again, should return quickly
- ReplicationUtil.createReplicationTable(conn);
- Assert.assertTrue(tops.exists(ReplicationTable.NAME));
-
- // Verify we have the same table as previously
- Assert.assertEquals(tableId, tops.tableIdMap().get(ReplicationTable.NAME));
- }
-
- @Test
- public void configureOnExistingTable() throws Exception {
- TableOperations tops = conn.tableOperations();
-
- // Create the table by hand
- tops.create(ReplicationTable.NAME);
- Map<String,EnumSet<IteratorScope>> iterators = tops.listIterators(ReplicationTable.NAME);
-
- Assert.assertFalse(iterators.containsKey(ReplicationTable.COMBINER_NAME));
-
- ReplicationUtil.configureReplicationTable(conn);
-
- // After configure the iterator should be set
- iterators = tops.listIterators(ReplicationTable.NAME);
- Assert.assertTrue(iterators.containsKey(ReplicationTable.COMBINER_NAME));
-
- // Needs to be set below versioning
- IteratorSetting expected = new IteratorSetting(30, ReplicationTable.COMBINER_NAME, StatusCombiner.class);
- Combiner.setColumns(expected, Arrays.asList(new Column(StatusSection.NAME), new Column(WorkSection.NAME)));
- IteratorSetting setting = tops.getIteratorSetting(ReplicationTable.NAME, ReplicationTable.COMBINER_NAME, IteratorScope.scan);
-
- Assert.assertEquals(expected, setting);
-
- // Check for locality groups too
- Map<String,Set<Text>> expectedLocalityGroups = ImmutableMap.of(ReplicationTable.STATUS_LG_NAME, ReplicationTable.STATUS_LG_COLFAMS,
- ReplicationTable.WORK_LG_NAME, ReplicationTable.WORK_LG_COLFAMS);
- Assert.assertEquals(expectedLocalityGroups, tops.getLocalityGroups(ReplicationTable.NAME));
- }
-
- @Test
- public void disablesVersioning() throws Exception {
- TableOperations tops = conn.tableOperations();
-
- if (tops.exists(ReplicationTable.NAME)) {
- tops.delete(ReplicationTable.NAME);
- }
-
- ReplicationUtil.createReplicationTable(conn);
-
- Set<String> iters = tops.listIterators(ReplicationTable.NAME).keySet();
-
- Assert.assertEquals(Collections.singleton(ReplicationTable.COMBINER_NAME), iters);
- }
-}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
----------------------------------------------------------------------
diff --git a/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java b/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
index 9928d3c..4c0a3a9 100644
--- a/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
+++ b/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
@@ -46,6 +46,7 @@ import org.apache.accumulo.core.metadata.schema.MetadataSchema.ReplicationSectio
import org.apache.accumulo.core.protobuf.ProtobufUtil;
import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
import org.apache.accumulo.core.replication.ReplicationTable;
+import org.apache.accumulo.core.replication.ReplicationTableOfflineException;
import org.apache.accumulo.core.replication.StatusUtil;
import org.apache.accumulo.core.replication.proto.Replication.Status;
import org.apache.accumulo.core.security.Authorizations;
@@ -86,7 +87,7 @@ public class GarbageCollectWriteAheadLogs {
/**
* Creates a new GC WAL object.
- *
+ *
* @param instance
* instance to use
* @param fs
@@ -102,7 +103,7 @@ public class GarbageCollectWriteAheadLogs {
/**
* Gets the instance used by this object.
- *
+ *
* @return instance
*/
Instance getInstance() {
@@ -111,7 +112,7 @@ public class GarbageCollectWriteAheadLogs {
/**
* Gets the volume manager used by this object.
- *
+ *
* @return volume manager
*/
VolumeManager getVolumeManager() {
@@ -120,7 +121,7 @@ public class GarbageCollectWriteAheadLogs {
/**
* Checks if the volume manager should move files to the trash rather than delete them.
- *
+ *
* @return true if trash is used
*/
boolean isUsingTrash() {
@@ -277,7 +278,7 @@ public class GarbageCollectWriteAheadLogs {
/**
* Converts a list of paths to their corresponding strings.
- *
+ *
* @param paths
* list of paths
* @return string forms of paths
@@ -292,7 +293,7 @@ public class GarbageCollectWriteAheadLogs {
/**
* Reverses the given mapping of file paths to servers. The returned map provides a list of file paths for each server. Any path whose name is not in the
* mapping of file names to paths is skipped.
- *
+ *
* @param fileToServerMap
* map of file paths to servers
* @param nameToFileMap
@@ -379,7 +380,7 @@ public class GarbageCollectWriteAheadLogs {
/**
* Determine if the given WAL is needed for replication
- *
+ *
* @param wal
* The full path (URI)
* @return True if the WAL is still needed by replication (not a candidate for deletion)
@@ -447,7 +448,7 @@ public class GarbageCollectWriteAheadLogs {
replScanner.setRange(Range.exact(wal));
return Iterables.concat(metaScanner, replScanner);
- } catch (TableNotFoundException e) {
+ } catch (ReplicationTableOfflineException e) {
// do nothing
}
@@ -458,15 +459,15 @@ public class GarbageCollectWriteAheadLogs {
return scanServers(ServerConstants.getWalDirs(), fileToServerMap, nameToFileMap);
}
- // TODO Remove deprecation warning suppression when Hadoop1 support is dropped
- @SuppressWarnings("deprecation")
/**
- * Scans write-ahead log directories for logs. The maps passed in are
- * populated with scan information.
+ * Scans write-ahead log directories for logs. The maps passed in are populated with scan information.
*
- * @param walDirs write-ahead log directories
- * @param fileToServerMap map of file paths to servers
- * @param nameToFileMap map of file names to paths
+ * @param walDirs
+ * write-ahead log directories
+ * @param fileToServerMap
+ * map of file paths to servers
+ * @param nameToFileMap
+ * map of file names to paths
* @return number of servers located (including those with no logs present)
*/
int scanServers(String[] walDirs, Map<Path,String> fileToServerMap, Map<String,Path> nameToFileMap) throws Exception {
@@ -484,7 +485,10 @@ public class GarbageCollectWriteAheadLogs {
continue;
for (FileStatus status : listing) {
String server = status.getPath().getName();
- if (status.isDir()) {
+ // TODO Remove deprecation warning suppression when Hadoop1 support is dropped
+ @SuppressWarnings("deprecation")
+ boolean isDirectory = status.isDir();
+ if (isDirectory) {
servers.add(server);
for (FileStatus file : fs.listStatus(new Path(walRoot, server))) {
if (isUUID(file.getPath().getName())) {
@@ -513,7 +517,7 @@ public class GarbageCollectWriteAheadLogs {
/**
* Looks for write-ahead logs in recovery directories.
- *
+ *
* @param recoveryDirs
* recovery directories
* @return map of log file names to paths
@@ -540,7 +544,7 @@ public class GarbageCollectWriteAheadLogs {
/**
* Checks if a string is a valid UUID.
- *
+ *
* @param name
* string to check
* @return true if string is a UUID
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
----------------------------------------------------------------------
diff --git a/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java b/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
index bbb2010..01fd2c8 100644
--- a/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
+++ b/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
@@ -63,6 +63,7 @@ import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.Da
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ScanFileColumnFamily;
import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
import org.apache.accumulo.core.replication.ReplicationTable;
+import org.apache.accumulo.core.replication.ReplicationTableOfflineException;
import org.apache.accumulo.core.replication.proto.Replication.Status;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.Credentials;
@@ -537,7 +538,7 @@ public class SimpleGarbageCollector implements Iface {
}
});
- } catch (TableNotFoundException e) {
+ } catch (ReplicationTableOfflineException e) {
// No elements that we need to preclude
return Iterators.emptyIterator();
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/gc/src/main/java/org/apache/accumulo/gc/replication/CloseWriteAheadLogReferences.java
----------------------------------------------------------------------
diff --git a/server/gc/src/main/java/org/apache/accumulo/gc/replication/CloseWriteAheadLogReferences.java b/server/gc/src/main/java/org/apache/accumulo/gc/replication/CloseWriteAheadLogReferences.java
index d13edf7..1c94253 100644
--- a/server/gc/src/main/java/org/apache/accumulo/gc/replication/CloseWriteAheadLogReferences.java
+++ b/server/gc/src/main/java/org/apache/accumulo/gc/replication/CloseWriteAheadLogReferences.java
@@ -91,8 +91,8 @@ public class CloseWriteAheadLogReferences implements Runnable {
throw new RuntimeException(e);
}
- if (!conn.tableOperations().exists(ReplicationTable.NAME)) {
- log.debug("Replication table doesn't exist, not attempting to clean up wals");
+ if (!ReplicationTable.isOnline(conn)) {
+ log.debug("Replication table isn't online, not attempting to clean up wals");
return;
}
@@ -124,7 +124,7 @@ public class CloseWriteAheadLogReferences implements Runnable {
/**
* Construct the set of referenced WALs from the metadata table
- *
+ *
* @param conn
* Connector
* @return The Set of WALs that are referenced in the metadata table
@@ -145,7 +145,7 @@ public class CloseWriteAheadLogReferences implements Runnable {
BatchScanner bs = null;
try {
// TODO Configurable number of threads
- bs = conn.createBatchScanner(MetadataTable.NAME, new Authorizations(), 4);
+ bs = conn.createBatchScanner(MetadataTable.NAME, Authorizations.EMPTY, 4);
bs.setRanges(Collections.singleton(TabletsSection.getRange()));
bs.fetchColumnFamily(LogColumnFamily.NAME);
@@ -176,7 +176,7 @@ public class CloseWriteAheadLogReferences implements Runnable {
/**
* Given the set of WALs which have references in the metadata table, close any status messages with reference that WAL.
- *
+ *
* @param conn
* Connector
* @param referencedWals
@@ -188,7 +188,7 @@ public class CloseWriteAheadLogReferences implements Runnable {
long recordsClosed = 0;
try {
bw = conn.createBatchWriter(MetadataTable.NAME, new BatchWriterConfig());
- bs = conn.createBatchScanner(MetadataTable.NAME, new Authorizations(), 4);
+ bs = conn.createBatchScanner(MetadataTable.NAME, Authorizations.EMPTY, 4);
bs.setRanges(Collections.singleton(Range.prefix(ReplicationSection.getRowPrefix())));
bs.fetchColumnFamily(ReplicationSection.COLF);
@@ -239,7 +239,7 @@ public class CloseWriteAheadLogReferences implements Runnable {
/**
* Write a closed {@link Status} mutation for the given {@link Key} using the provided {@link BatchWriter}
- *
+ *
* @param bw
* BatchWriter
* @param k
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java
----------------------------------------------------------------------
diff --git a/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java b/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java
index cd69efb..71e5f7d 100644
--- a/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java
+++ b/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java
@@ -55,7 +55,6 @@ import org.apache.accumulo.core.replication.StatusUtil;
import org.apache.accumulo.core.replication.proto.Replication.Status;
import org.apache.accumulo.core.security.Credentials;
import org.apache.accumulo.server.fs.VolumeManager;
-import org.apache.accumulo.server.replication.ReplicationUtil;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
@@ -384,11 +383,9 @@ public class GarbageCollectWriteAheadLogsTest {
GarbageCollectWriteAheadLogs gcWALs = new GarbageCollectWriteAheadLogs(inst, volMgr, false);
- ReplicationUtil.createReplicationTable(conn);
-
long file1CreateTime = System.currentTimeMillis();
long file2CreateTime = file1CreateTime + 50;
- BatchWriter bw = conn.createBatchWriter(ReplicationTable.NAME, new BatchWriterConfig());
+ BatchWriter bw = ReplicationTable.getBatchWriter(conn);
Mutation m = new Mutation("/wals/" + file1);
StatusSection.add(m, new Text("1"), StatusUtil.fileCreatedValue(file1CreateTime));
bw.addMutation(m);
@@ -428,8 +425,6 @@ public class GarbageCollectWriteAheadLogsTest {
GarbageCollectWriteAheadLogs gcWALs = new GarbageCollectWriteAheadLogs(inst, volMgr, false);
- ReplicationUtil.createReplicationTable(conn);
-
long file1CreateTime = System.currentTimeMillis();
long file2CreateTime = file1CreateTime + 50;
// Write some records to the metadata table, we haven't yet written status records to the replication table
@@ -488,8 +483,6 @@ public class GarbageCollectWriteAheadLogsTest {
public void fetchesReplicationEntriesFromMetadataAndReplicationTables() throws Exception {
Instance inst = new MockInstance(testName.getMethodName());
Connector conn = inst.getConnector("root", new PasswordToken(""));
- ReplicationUtil.createReplicationTable(conn);
-
long walCreateTime = System.currentTimeMillis();
String wal = "hdfs://localhost:8020/accumulo/wal/tserver+port/123456-1234-1234-12345678";
BatchWriter bw = conn.createBatchWriter(MetadataTable.NAME, new BatchWriterConfig());
[02/10] accumulo git commit: ACCUMULO-3147 Refactor ReplicationTable
constants
Posted by ct...@apache.org.
ACCUMULO-3147 Refactor ReplicationTable constants
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/f1a47b38
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/f1a47b38
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/f1a47b38
Branch: refs/heads/master
Commit: f1a47b38d3f5338a9ad1104cd3009b327e08c371
Parents: cc48e37
Author: Christopher Tubbs <ct...@apache.org>
Authored: Wed Oct 29 15:11:48 2014 -0400
Committer: Christopher Tubbs <ct...@apache.org>
Committed: Thu Nov 6 18:39:54 2014 -0500
----------------------------------------------------------------------
.../client/impl/ReplicationOperationsImpl.java | 8 +-
.../replication/PrintReplicationRecords.java | 2 +-
.../core/replication/ReplicaSystemHelper.java | 11 +-
.../core/replication/ReplicationConstants.java | 1 -
.../core/replication/ReplicationTable.java | 61 +++++
.../ReplicationOperationsImplTest.java | 24 +-
.../server/replication/ReplicationTable.java | 227 -------------------
.../server/replication/ReplicationUtil.java | 184 +++++++++++++--
.../replication/ReplicationTableTest.java | 11 +-
.../gc/GarbageCollectWriteAheadLogs.java | 2 +-
.../accumulo/gc/SimpleGarbageCollector.java | 4 +-
.../CloseWriteAheadLogReferences.java | 2 +-
.../gc/GarbageCollectWriteAheadLogsTest.java | 36 ++-
.../CloseWriteAheadLogReferencesTest.java | 9 +-
.../master/metrics/ReplicationMetrics.java | 4 +-
.../DistributedWorkQueueWorkAssigner.java | 12 +-
.../master/replication/FinishedWorkUpdater.java | 5 +-
.../RemoveCompleteReplicationRecords.java | 4 +-
.../master/replication/StatusMaker.java | 45 ++--
.../accumulo/master/replication/WorkMaker.java | 2 +-
.../replication/FinishedWorkUpdaterTest.java | 15 +-
.../RemoveCompleteReplicationRecordsTest.java | 13 +-
.../replication/SequentialWorkAssignerTest.java | 11 +-
.../master/replication/StatusMakerTest.java | 2 +-
.../replication/UnorderedWorkAssignerTest.java | 16 +-
.../master/replication/WorkMakerTest.java | 2 +-
.../monitor/servlets/ReplicationServlet.java | 3 +-
.../replication/ReplicationProcessor.java | 2 +-
.../replication/MultiInstanceReplicationIT.java | 6 +-
.../test/replication/ReplicationIT.java | 18 +-
.../test/replication/StatusCombinerMacTest.java | 5 +-
.../UnorderedWorkAssignerReplicationIT.java | 7 +-
32 files changed, 393 insertions(+), 361 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/core/src/main/java/org/apache/accumulo/core/client/impl/ReplicationOperationsImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/impl/ReplicationOperationsImpl.java b/core/src/main/java/org/apache/accumulo/core/client/impl/ReplicationOperationsImpl.java
index f820aa4..a8f3f21 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/ReplicationOperationsImpl.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/ReplicationOperationsImpl.java
@@ -44,8 +44,8 @@ import org.apache.accumulo.core.metadata.schema.MetadataSchema;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.ReplicationSection;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.LogColumnFamily;
import org.apache.accumulo.core.protobuf.ProtobufUtil;
-import org.apache.accumulo.core.replication.ReplicationConstants;
import org.apache.accumulo.core.replication.ReplicationSchema.OrderSection;
+import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.StatusUtil;
import org.apache.accumulo.core.replication.proto.Replication.Status;
import org.apache.accumulo.core.security.Authorizations;
@@ -152,7 +152,7 @@ public class ReplicationOperationsImpl implements ReplicationOperations {
log.info("reading from replication table");
boolean allReplicationRefsReplicated = false;
while (!allReplicationRefsReplicated) {
- BatchScanner bs = conn.createBatchScanner(ReplicationConstants.TABLE_NAME, Authorizations.EMPTY, 4);
+ BatchScanner bs = conn.createBatchScanner(ReplicationTable.NAME, Authorizations.EMPTY, 4);
bs.setRanges(Collections.singleton(new Range()));
try {
allReplicationRefsReplicated = allReferencesReplicated(bs, tableId, wals);
@@ -216,7 +216,7 @@ public class ReplicationOperationsImpl implements ReplicationOperations {
protected Text getTableId(Connector conn, String tableName) throws AccumuloException, AccumuloSecurityException, TableNotFoundException {
TableOperations tops = conn.tableOperations();
- while (!tops.exists(ReplicationConstants.TABLE_NAME)) {
+ while (!tops.exists(ReplicationTable.NAME)) {
UtilWaitThread.sleep(200);
}
@@ -232,7 +232,7 @@ public class ReplicationOperationsImpl implements ReplicationOperations {
}
}
- return new Text(strTableId);
+ return new Text(strTableId);
}
@Override
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/core/src/main/java/org/apache/accumulo/core/replication/PrintReplicationRecords.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/replication/PrintReplicationRecords.java b/core/src/main/java/org/apache/accumulo/core/replication/PrintReplicationRecords.java
index 2aef652..ee606b5 100644
--- a/core/src/main/java/org/apache/accumulo/core/replication/PrintReplicationRecords.java
+++ b/core/src/main/java/org/apache/accumulo/core/replication/PrintReplicationRecords.java
@@ -80,7 +80,7 @@ public class PrintReplicationRecords implements Runnable {
out.println("--------------------------------------------------------------------");
try {
- s = conn.createScanner(ReplicationConstants.TABLE_NAME, Authorizations.EMPTY);
+ s = conn.createScanner(ReplicationTable.NAME, Authorizations.EMPTY);
} catch (TableNotFoundException e) {
log.error("Replication table does not exist");
return;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/core/src/main/java/org/apache/accumulo/core/replication/ReplicaSystemHelper.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/replication/ReplicaSystemHelper.java b/core/src/main/java/org/apache/accumulo/core/replication/ReplicaSystemHelper.java
index a022362..b27aa43 100644
--- a/core/src/main/java/org/apache/accumulo/core/replication/ReplicaSystemHelper.java
+++ b/core/src/main/java/org/apache/accumulo/core/replication/ReplicaSystemHelper.java
@@ -33,7 +33,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- *
+ *
*/
public class ReplicaSystemHelper {
private static final Logger log = LoggerFactory.getLogger(ReplicaSystemHelper.class);
@@ -48,7 +48,7 @@ public class ReplicaSystemHelper {
/**
* Record the updated Status for this file and target
- *
+ *
* @param filePath
* Path to file being replicated
* @param status
@@ -56,11 +56,12 @@ public class ReplicaSystemHelper {
* @param target
* Peer that was replicated to
*/
- public void recordNewStatus(Path filePath, Status status, ReplicationTarget target) throws AccumuloException, AccumuloSecurityException, TableNotFoundException {
+ public void recordNewStatus(Path filePath, Status status, ReplicationTarget target) throws AccumuloException, AccumuloSecurityException,
+ TableNotFoundException {
Connector conn = inst.getConnector(creds.getPrincipal(), creds.getToken());
- BatchWriter bw = conn.createBatchWriter(ReplicationConstants.TABLE_NAME, new BatchWriterConfig());
+ BatchWriter bw = conn.createBatchWriter(ReplicationTable.NAME, new BatchWriterConfig());
try {
- log.debug("Recording new status for {}, {}", filePath.toString(), ProtobufUtil.toString (status));
+ log.debug("Recording new status for {}, {}", filePath.toString(), ProtobufUtil.toString(status));
Mutation m = new Mutation(filePath.toString());
WorkSection.add(m, target.toText(), ProtobufUtil.toValue(status));
bw.addMutation(m);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/core/src/main/java/org/apache/accumulo/core/replication/ReplicationConstants.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/replication/ReplicationConstants.java b/core/src/main/java/org/apache/accumulo/core/replication/ReplicationConstants.java
index 9815634..3d71681 100644
--- a/core/src/main/java/org/apache/accumulo/core/replication/ReplicationConstants.java
+++ b/core/src/main/java/org/apache/accumulo/core/replication/ReplicationConstants.java
@@ -20,7 +20,6 @@ package org.apache.accumulo.core.replication;
*
*/
public class ReplicationConstants {
- public static final String TABLE_NAME = "replication";
// Constants for replication information in zookeeper
public static final String ZOO_BASE = "/replication";
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/core/src/main/java/org/apache/accumulo/core/replication/ReplicationTable.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/replication/ReplicationTable.java b/core/src/main/java/org/apache/accumulo/core/replication/ReplicationTable.java
new file mode 100644
index 0000000..2736762
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/replication/ReplicationTable.java
@@ -0,0 +1,61 @@
+/*
+ * 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.util.Collections;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.accumulo.core.client.BatchScanner;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
+import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.hadoop.io.Text;
+
+import com.google.common.collect.ImmutableMap;
+
+public class ReplicationTable {
+
+ public static final String NAME = "replication";
+
+ public static final String COMBINER_NAME = "statuscombiner";
+
+ public static final String STATUS_LG_NAME = StatusSection.NAME.toString();
+ public static final Set<Text> STATUS_LG_COLFAMS = Collections.singleton(StatusSection.NAME);
+ public static final String WORK_LG_NAME = WorkSection.NAME.toString();
+ public static final Set<Text> WORK_LG_COLFAMS = Collections.singleton(WorkSection.NAME);
+ public static final Map<String,Set<Text>> LOCALITY_GROUPS = ImmutableMap.of(STATUS_LG_NAME, STATUS_LG_COLFAMS, WORK_LG_NAME, WORK_LG_COLFAMS);
+ public static final String STATUS_FORMATTER_CLASS_NAME = StatusFormatter.class.getName();
+
+ public static Scanner getScanner(Connector conn) throws TableNotFoundException {
+ return conn.createScanner(NAME, Authorizations.EMPTY);
+ }
+
+ public static BatchWriter getBatchWriter(Connector conn) throws TableNotFoundException {
+ return conn.createBatchWriter(NAME, new BatchWriterConfig());
+ }
+
+ public static BatchScanner getBatchScanner(Connector conn, int queryThreads) throws TableNotFoundException {
+ return conn.createBatchScanner(NAME, Authorizations.EMPTY, queryThreads);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/core/src/test/java/org/apache/accumulo/core/replication/ReplicationOperationsImplTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/accumulo/core/replication/ReplicationOperationsImplTest.java b/core/src/test/java/org/apache/accumulo/core/replication/ReplicationOperationsImplTest.java
index 7cc839f..54a9f8c 100644
--- a/core/src/test/java/org/apache/accumulo/core/replication/ReplicationOperationsImplTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/replication/ReplicationOperationsImplTest.java
@@ -67,14 +67,14 @@ public class ReplicationOperationsImplTest {
@Test
public void waitsUntilEntriesAreReplicated() throws Exception {
Connector conn = inst.getConnector("root", new PasswordToken(""));
- conn.tableOperations().create(ReplicationConstants.TABLE_NAME);
+ conn.tableOperations().create(ReplicationTable.NAME);
conn.tableOperations().create("foo");
Text tableId = new Text(conn.tableOperations().tableIdMap().get("foo"));
String file1 = "/accumulo/wals/tserver+port/" + UUID.randomUUID(), file2 = "/accumulo/wals/tserver+port/" + UUID.randomUUID();
Status stat = Status.newBuilder().setBegin(0).setEnd(10000).setInfiniteEnd(false).setClosed(false).build();
- BatchWriter bw = conn.createBatchWriter(ReplicationConstants.TABLE_NAME, new BatchWriterConfig());
+ BatchWriter bw = conn.createBatchWriter(ReplicationTable.NAME, new BatchWriterConfig());
Mutation m = new Mutation(file1);
StatusSection.add(m, tableId, ProtobufUtil.toValue(stat));
@@ -136,7 +136,7 @@ public class ReplicationOperationsImplTest {
Assert.assertFalse(done.get());
// Remove the replication entries too
- bw = conn.createBatchWriter(ReplicationConstants.TABLE_NAME, new BatchWriterConfig());
+ bw = conn.createBatchWriter(ReplicationTable.NAME, new BatchWriterConfig());
m = new Mutation(file1);
m.putDelete(StatusSection.NAME, tableId);
bw.addMutation(m);
@@ -163,7 +163,7 @@ public class ReplicationOperationsImplTest {
@Test
public void unrelatedReplicationRecordsDontBlockDrain() throws Exception {
Connector conn = inst.getConnector("root", new PasswordToken(""));
- conn.tableOperations().create(ReplicationConstants.TABLE_NAME);
+ conn.tableOperations().create(ReplicationTable.NAME);
conn.tableOperations().create("foo");
conn.tableOperations().create("bar");
@@ -173,7 +173,7 @@ public class ReplicationOperationsImplTest {
String file1 = "/accumulo/wals/tserver+port/" + UUID.randomUUID(), file2 = "/accumulo/wals/tserver+port/" + UUID.randomUUID();
Status stat = Status.newBuilder().setBegin(0).setEnd(10000).setInfiniteEnd(false).setClosed(false).build();
- BatchWriter bw = conn.createBatchWriter(ReplicationConstants.TABLE_NAME, new BatchWriterConfig());
+ BatchWriter bw = conn.createBatchWriter(ReplicationTable.NAME, new BatchWriterConfig());
Mutation m = new Mutation(file1);
StatusSection.add(m, tableId1, ProtobufUtil.toValue(stat));
@@ -227,7 +227,7 @@ public class ReplicationOperationsImplTest {
Assert.assertFalse(done.get());
// Remove the replication entries too
- bw = conn.createBatchWriter(ReplicationConstants.TABLE_NAME, new BatchWriterConfig());
+ bw = conn.createBatchWriter(ReplicationTable.NAME, new BatchWriterConfig());
m = new Mutation(file1);
m.putDelete(StatusSection.NAME, tableId1);
bw.addMutation(m);
@@ -247,7 +247,7 @@ public class ReplicationOperationsImplTest {
@Test
public void inprogressReplicationRecordsBlockExecution() throws Exception {
Connector conn = inst.getConnector("root", new PasswordToken(""));
- conn.tableOperations().create(ReplicationConstants.TABLE_NAME);
+ conn.tableOperations().create(ReplicationTable.NAME);
conn.tableOperations().create("foo");
Text tableId1 = new Text(conn.tableOperations().tableIdMap().get("foo"));
@@ -255,7 +255,7 @@ public class ReplicationOperationsImplTest {
String file1 = "/accumulo/wals/tserver+port/" + UUID.randomUUID();
Status stat = Status.newBuilder().setBegin(0).setEnd(10000).setInfiniteEnd(false).setClosed(false).build();
- BatchWriter bw = conn.createBatchWriter(ReplicationConstants.TABLE_NAME, new BatchWriterConfig());
+ BatchWriter bw = conn.createBatchWriter(ReplicationTable.NAME, new BatchWriterConfig());
Mutation m = new Mutation(file1);
StatusSection.add(m, tableId1, ProtobufUtil.toValue(stat));
@@ -313,7 +313,7 @@ public class ReplicationOperationsImplTest {
Assert.assertFalse(done.get());
// Remove the replication entries too
- bw = conn.createBatchWriter(ReplicationConstants.TABLE_NAME, new BatchWriterConfig());
+ bw = conn.createBatchWriter(ReplicationTable.NAME, new BatchWriterConfig());
m = new Mutation(file1);
m.put(StatusSection.NAME, tableId1, ProtobufUtil.toValue(newStatus));
bw.addMutation(m);
@@ -333,7 +333,7 @@ public class ReplicationOperationsImplTest {
@Test
public void laterCreatedLogsDontBlockExecution() throws Exception {
Connector conn = inst.getConnector("root", new PasswordToken(""));
- conn.tableOperations().create(ReplicationConstants.TABLE_NAME);
+ conn.tableOperations().create(ReplicationTable.NAME);
conn.tableOperations().create("foo");
Text tableId1 = new Text(conn.tableOperations().tableIdMap().get("foo"));
@@ -341,7 +341,7 @@ public class ReplicationOperationsImplTest {
String file1 = "/accumulo/wals/tserver+port/" + UUID.randomUUID();
Status stat = Status.newBuilder().setBegin(0).setEnd(10000).setInfiniteEnd(false).setClosed(false).build();
- BatchWriter bw = conn.createBatchWriter(ReplicationConstants.TABLE_NAME, new BatchWriterConfig());
+ BatchWriter bw = conn.createBatchWriter(ReplicationTable.NAME, new BatchWriterConfig());
Mutation m = new Mutation(file1);
StatusSection.add(m, tableId1, ProtobufUtil.toValue(stat));
bw.addMutation(m);
@@ -395,7 +395,7 @@ public class ReplicationOperationsImplTest {
System.out.println(e.getKey());
}
- bw = conn.createBatchWriter(ReplicationConstants.TABLE_NAME, new BatchWriterConfig());
+ bw = conn.createBatchWriter(ReplicationTable.NAME, new BatchWriterConfig());
m = new Mutation(file1);
m.putDelete(StatusSection.NAME, tableId1);
bw.addMutation(m);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/server/base/src/main/java/org/apache/accumulo/server/replication/ReplicationTable.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/replication/ReplicationTable.java b/server/base/src/main/java/org/apache/accumulo/server/replication/ReplicationTable.java
deleted file mode 100644
index e7f10dc..0000000
--- a/server/base/src/main/java/org/apache/accumulo/server/replication/ReplicationTable.java
+++ /dev/null
@@ -1,227 +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.server.replication;
-
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.EnumSet;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
-
-import org.apache.accumulo.core.client.AccumuloException;
-import org.apache.accumulo.core.client.AccumuloSecurityException;
-import org.apache.accumulo.core.client.BatchScanner;
-import org.apache.accumulo.core.client.BatchWriter;
-import org.apache.accumulo.core.client.BatchWriterConfig;
-import org.apache.accumulo.core.client.Connector;
-import org.apache.accumulo.core.client.IteratorSetting;
-import org.apache.accumulo.core.client.Scanner;
-import org.apache.accumulo.core.client.IteratorSetting.Column;
-import org.apache.accumulo.core.client.TableExistsException;
-import org.apache.accumulo.core.client.TableNotFoundException;
-import org.apache.accumulo.core.client.admin.TableOperations;
-import org.apache.accumulo.core.conf.Property;
-import org.apache.accumulo.core.iterators.Combiner;
-import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
-import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
-import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
-import org.apache.accumulo.core.replication.ReplicationConstants;
-import org.apache.accumulo.core.replication.StatusFormatter;
-import org.apache.accumulo.core.security.Authorizations;
-import org.apache.accumulo.core.security.TablePermission;
-import org.apache.accumulo.fate.util.UtilWaitThread;
-import org.apache.hadoop.io.Text;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.collect.ImmutableMap;
-
-public class ReplicationTable {
- private static final Logger log = LoggerFactory.getLogger(ReplicationTable.class);
-
- public static final String NAME = ReplicationConstants.TABLE_NAME;
-
- public static final String COMBINER_NAME = "statuscombiner";
-
- public static final String STATUS_LG_NAME = StatusSection.NAME.toString();
- public static final Set<Text> STATUS_LG_COLFAMS = Collections.singleton(StatusSection.NAME);
- public static final String WORK_LG_NAME = WorkSection.NAME.toString();
- public static final Set<Text> WORK_LG_COLFAMS = Collections.singleton(WorkSection.NAME);
- public static final Map<String,Set<Text>> LOCALITY_GROUPS = ImmutableMap.of(STATUS_LG_NAME, STATUS_LG_COLFAMS, WORK_LG_NAME, WORK_LG_COLFAMS);
- public static final String STATUS_FORMATTER_CLASS_NAME = StatusFormatter.class.getName();
-
- public static synchronized void create(Connector conn) {
- TableOperations tops = conn.tableOperations();
- if (tops.exists(NAME)) {
- if (configureReplicationTable(conn)) {
- return;
- }
- }
-
- for (int i = 0; i < 5; i++) {
- try {
- if (!tops.exists(NAME)) {
- tops.create(NAME, false);
- }
- break;
- } catch (AccumuloException | AccumuloSecurityException e) {
- log.error("Failed to create replication table", e);
- } catch (TableExistsException e) {
- // Shouldn't happen unless someone else made the table
- }
- log.error("Retrying table creation in 1 second...");
- UtilWaitThread.sleep(1000);
- }
-
- for (int i = 0; i < 5; i++) {
- if (configureReplicationTable(conn)) {
- return;
- }
-
- log.error("Failed to configure the replication table, retying...");
- UtilWaitThread.sleep(1000);
- }
-
- throw new RuntimeException("Could not configure replication table");
- }
-
- /**
- * Attempts to configure the replication table, will return false if it fails
- *
- * @param conn
- * Connector for the instance
- * @return True if the replication table is properly configured
- */
- protected static synchronized boolean configureReplicationTable(Connector conn) {
- try {
- conn.securityOperations().grantTablePermission("root", NAME, TablePermission.READ);
- } catch (AccumuloException | AccumuloSecurityException e) {
- log.warn("Could not grant root user read access to replication table", e);
- // Should this be fatal? It's only for convenience, all r/w is done by !SYSTEM
- }
-
- TableOperations tops = conn.tableOperations();
- Map<String,EnumSet<IteratorScope>> iterators = null;
- try {
- iterators = tops.listIterators(NAME);
- } catch (AccumuloSecurityException | AccumuloException | TableNotFoundException e) {
- log.error("Could not fetch iterators for " + NAME, e);
- return false;
- }
-
- if (!iterators.containsKey(COMBINER_NAME)) {
- // Set our combiner and combine all columns
- IteratorSetting setting = new IteratorSetting(30, COMBINER_NAME, StatusCombiner.class);
- Combiner.setColumns(setting, Arrays.asList(new Column(StatusSection.NAME), new Column(WorkSection.NAME)));
- try {
- tops.attachIterator(NAME, setting);
- } catch (AccumuloSecurityException | AccumuloException | TableNotFoundException e) {
- log.error("Could not set StatusCombiner on replication table", e);
- return false;
- }
- }
-
- Map<String,Set<Text>> localityGroups;
- try {
- localityGroups = tops.getLocalityGroups(NAME);
- } catch (TableNotFoundException | AccumuloException e) {
- log.error("Could not fetch locality groups", e);
- return false;
- }
-
- Set<Text> statusColfams = localityGroups.get(STATUS_LG_NAME), workColfams = localityGroups.get(WORK_LG_NAME);
- if (null == statusColfams || null == workColfams) {
- try {
- tops.setLocalityGroups(NAME, LOCALITY_GROUPS);
- } catch (AccumuloException | AccumuloSecurityException | TableNotFoundException e) {
- log.error("Could not set locality groups on replication table", e);
- return false;
- }
- }
-
- // Make sure the StatusFormatter is set on the metadata table
- Iterable<Entry<String,String>> properties;
- try {
- properties = tops.getProperties(NAME);
- } catch (AccumuloException | TableNotFoundException e) {
- log.error("Could not fetch table properties on replication table", e);
- return false;
- }
-
- boolean formatterConfigured = false;
- for (Entry<String,String> property : properties) {
- if (Property.TABLE_FORMATTER_CLASS.getKey().equals(property.getKey())) {
- if (!STATUS_FORMATTER_CLASS_NAME.equals(property.getValue())) {
- log.info("Changing formatter for {} table from {} to {}", NAME, property.getValue(), STATUS_FORMATTER_CLASS_NAME);
- try {
- tops.setProperty(NAME, Property.TABLE_FORMATTER_CLASS.getKey(), STATUS_FORMATTER_CLASS_NAME);
- } catch (AccumuloException | AccumuloSecurityException e) {
- log.error("Could not set formatter on replication table", e);
- return false;
- }
- }
-
- formatterConfigured = true;
-
- // Don't need to keep iterating over the properties after we found the one we were looking for
- break;
- }
- }
-
- if (!formatterConfigured) {
- try {
- tops.setProperty(NAME, Property.TABLE_FORMATTER_CLASS.getKey(), STATUS_FORMATTER_CLASS_NAME);
- } catch (AccumuloException | AccumuloSecurityException e) {
- log.error("Could not set formatter on replication table", e);
- return false;
- }
- }
-
- log.debug("Successfully configured replication table");
-
- return true;
- }
-
- public static Scanner getScanner(Connector conn, Authorizations auths) throws TableNotFoundException {
- return conn.createScanner(NAME, auths);
- }
-
- public static Scanner getScanner(Connector conn) throws TableNotFoundException {
- return getScanner(conn, new Authorizations());
- }
-
- public static BatchWriter getBatchWriter(Connector conn) throws TableNotFoundException {
- return getBatchWriter(conn, new BatchWriterConfig());
- }
-
- public static BatchWriter getBatchWriter(Connector conn, BatchWriterConfig config) throws TableNotFoundException {
- return conn.createBatchWriter(NAME, config);
- }
-
- public static BatchScanner getBatchScanner(Connector conn, int queryThreads) throws TableNotFoundException {
- return conn.createBatchScanner(NAME, new Authorizations(), queryThreads);
- }
-
- public static boolean exists(Connector conn) {
- return exists(conn.tableOperations());
- }
-
- public static boolean exists(TableOperations tops) {
- return tops.exists(NAME);
- }
-}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/server/base/src/main/java/org/apache/accumulo/server/replication/ReplicationUtil.java
----------------------------------------------------------------------
diff --git a/server/base/src/main/java/org/apache/accumulo/server/replication/ReplicationUtil.java b/server/base/src/main/java/org/apache/accumulo/server/replication/ReplicationUtil.java
index 8e9612a..e05a08c 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/replication/ReplicationUtil.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/replication/ReplicationUtil.java
@@ -18,7 +18,9 @@ package org.apache.accumulo.server.replication;
import static java.nio.charset.StandardCharsets.UTF_8;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
@@ -30,7 +32,10 @@ import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.BatchScanner;
import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.IteratorSetting;
+import org.apache.accumulo.core.client.IteratorSetting.Column;
import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.TableExistsException;
import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.client.admin.TableOperations;
import org.apache.accumulo.core.client.replication.ReplicaSystem;
@@ -39,15 +44,20 @@ import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.iterators.Combiner;
+import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
import org.apache.accumulo.core.master.thrift.MasterMonitorInfo;
import org.apache.accumulo.core.metadata.MetadataTable;
import org.apache.accumulo.core.metadata.RootTable;
-import org.apache.accumulo.core.replication.ReplicationConstants;
+import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
+import org.apache.accumulo.core.replication.ReplicationTable;
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.Authorizations;
+import org.apache.accumulo.core.security.TablePermission;
+import org.apache.accumulo.fate.util.UtilWaitThread;
import org.apache.accumulo.server.zookeeper.ZooCache;
import org.apache.hadoop.io.Text;
import org.slf4j.Logger;
@@ -88,7 +98,9 @@ public class ReplicationUtil {
/**
* Extract replication peers from system configuration
- * @param systemProperties System properties, typically from Connector.instanceOperations().getSystemConfiguration()
+ *
+ * @param systemProperties
+ * System properties, typically from Connector.instanceOperations().getSystemConfiguration()
* @return Configured replication peers
*/
public Map<String,String> getPeers(Map<String,String> systemProperties) {
@@ -99,11 +111,12 @@ public class ReplicationUtil {
for (Entry<String,String> property : systemProperties.entrySet()) {
String key = property.getKey();
// Filter out cruft that we don't want
- if (key.startsWith(definedPeersPrefix) && !key.startsWith(Property.REPLICATION_PEER_USER.getKey()) && !key.startsWith(Property.REPLICATION_PEER_PASSWORD.getKey())) {
+ if (key.startsWith(definedPeersPrefix) && !key.startsWith(Property.REPLICATION_PEER_USER.getKey())
+ && !key.startsWith(Property.REPLICATION_PEER_PASSWORD.getKey())) {
String peerName = property.getKey().substring(definedPeersPrefix.length());
ReplicaSystem replica;
try {
- replica = ReplicaSystemFactory.get(property.getValue());
+ replica = ReplicaSystemFactory.get(property.getValue());
} catch (Exception e) {
log.warn("Could not instantiate ReplicaSystem for {} with configuration {}", property.getKey(), property.getValue(), e);
continue;
@@ -162,7 +175,7 @@ public class ReplicationUtil {
// Read over the queued work
BatchScanner bs;
try {
- bs = conn.createBatchScanner(ReplicationConstants.TABLE_NAME, Authorizations.EMPTY, 4);
+ bs = conn.createBatchScanner(ReplicationTable.NAME, Authorizations.EMPTY, 4);
} catch (TableNotFoundException e) {
log.debug("No replication table exists", e);
return counts;
@@ -194,9 +207,13 @@ public class ReplicationUtil {
/**
* Fetches the absolute path of the file to be replicated.
- * @param conn Accumulo Connector
- * @param workQueuePath Root path for the Replication WorkQueue
- * @param queueKey The Replication work queue key
+ *
+ * @param conn
+ * Accumulo Connector
+ * @param workQueuePath
+ * Root path for the Replication WorkQueue
+ * @param queueKey
+ * The Replication work queue key
* @return The absolute path for the file, or null if the key is no longer in ZooKeeper
*/
public String getAbsolutePath(Connector conn, String workQueuePath, String queueKey) {
@@ -210,9 +227,13 @@ public class ReplicationUtil {
/**
* Compute a progress string for the replication of the given WAL
- * @param conn Accumulo Connector
- * @param path Absolute path to a WAL, or null
- * @param target ReplicationTarget the WAL is being replicated to
+ *
+ * @param conn
+ * Accumulo Connector
+ * @param path
+ * Absolute path to a WAL, or null
+ * @param target
+ * ReplicationTarget the WAL is being replicated to
* @return A status message for a file being replicated
*/
public String getProgress(Connector conn, String path, ReplicationTarget target) {
@@ -222,7 +243,7 @@ public class ReplicationUtil {
if (null != path) {
Scanner s;
try {
- s = conn.createScanner(ReplicationConstants.TABLE_NAME, Authorizations.EMPTY);
+ s = conn.createScanner(ReplicationTable.NAME, Authorizations.EMPTY);
} catch (TableNotFoundException e) {
log.debug("Replication table no long exists", e);
return status;
@@ -236,8 +257,8 @@ public class ReplicationUtil {
try {
kv = Iterables.getOnlyElement(s);
} catch (NoSuchElementException e) {
- log.trace("Could not find status of {} replicating to {}", path, target);
- status = "Unknown";
+ log.trace("Could not find status of {} replicating to {}", path, target);
+ status = "Unknown";
} finally {
s.close();
}
@@ -267,9 +288,142 @@ public class ReplicationUtil {
public Map<String,String> invert(Map<String,String> map) {
Map<String,String> newMap = Maps.newHashMapWithExpectedSize(map.size());
- for(Entry<String,String> entry : map.entrySet()) {
+ for (Entry<String,String> entry : map.entrySet()) {
newMap.put(entry.getValue(), entry.getKey());
}
return newMap;
}
+
+ public static synchronized void createReplicationTable(Connector conn) {
+ TableOperations tops = conn.tableOperations();
+ if (tops.exists(ReplicationTable.NAME)) {
+ if (configureReplicationTable(conn)) {
+ return;
+ }
+ }
+
+ for (int i = 0; i < 5; i++) {
+ try {
+ if (!tops.exists(ReplicationTable.NAME)) {
+ tops.create(ReplicationTable.NAME, false);
+ }
+ break;
+ } catch (AccumuloException | AccumuloSecurityException e) {
+ log.error("Failed to create replication table", e);
+ } catch (TableExistsException e) {
+ // Shouldn't happen unless someone else made the table
+ }
+ log.error("Retrying table creation in 1 second...");
+ UtilWaitThread.sleep(1000);
+ }
+
+ for (int i = 0; i < 5; i++) {
+ if (configureReplicationTable(conn)) {
+ return;
+ }
+
+ log.error("Failed to configure the replication table, retying...");
+ UtilWaitThread.sleep(1000);
+ }
+
+ throw new RuntimeException("Could not configure replication table");
+ }
+
+ /**
+ * Attempts to configure the replication table, will return false if it fails
+ *
+ * @param conn
+ * Connector for the instance
+ * @return True if the replication table is properly configured
+ */
+ protected static synchronized boolean configureReplicationTable(Connector conn) {
+ try {
+ conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.READ);
+ } catch (AccumuloException | AccumuloSecurityException e) {
+ log.warn("Could not grant root user read access to replication table", e);
+ // Should this be fatal? It's only for convenience, all r/w is done by !SYSTEM
+ }
+
+ TableOperations tops = conn.tableOperations();
+ Map<String,EnumSet<IteratorScope>> iterators = null;
+ try {
+ iterators = tops.listIterators(ReplicationTable.NAME);
+ } catch (AccumuloSecurityException | AccumuloException | TableNotFoundException e) {
+ log.error("Could not fetch iterators for " + ReplicationTable.NAME, e);
+ return false;
+ }
+
+ if (!iterators.containsKey(ReplicationTable.COMBINER_NAME)) {
+ // Set our combiner and combine all columns
+ IteratorSetting setting = new IteratorSetting(30, ReplicationTable.COMBINER_NAME, StatusCombiner.class);
+ Combiner.setColumns(setting, Arrays.asList(new Column(StatusSection.NAME), new Column(WorkSection.NAME)));
+ try {
+ tops.attachIterator(ReplicationTable.NAME, setting);
+ } catch (AccumuloSecurityException | AccumuloException | TableNotFoundException e) {
+ log.error("Could not set StatusCombiner on replication table", e);
+ return false;
+ }
+ }
+
+ Map<String,Set<Text>> localityGroups;
+ try {
+ localityGroups = tops.getLocalityGroups(ReplicationTable.NAME);
+ } catch (TableNotFoundException | AccumuloException e) {
+ log.error("Could not fetch locality groups", e);
+ return false;
+ }
+
+ Set<Text> statusColfams = localityGroups.get(ReplicationTable.STATUS_LG_NAME), workColfams = localityGroups.get(ReplicationTable.WORK_LG_NAME);
+ if (null == statusColfams || null == workColfams) {
+ try {
+ tops.setLocalityGroups(ReplicationTable.NAME, ReplicationTable.LOCALITY_GROUPS);
+ } catch (AccumuloException | AccumuloSecurityException | TableNotFoundException e) {
+ log.error("Could not set locality groups on replication table", e);
+ return false;
+ }
+ }
+
+ // Make sure the StatusFormatter is set on the metadata table
+ Iterable<Entry<String,String>> properties;
+ try {
+ properties = tops.getProperties(ReplicationTable.NAME);
+ } catch (AccumuloException | TableNotFoundException e) {
+ log.error("Could not fetch table properties on replication table", e);
+ return false;
+ }
+
+ boolean formatterConfigured = false;
+ for (Entry<String,String> property : properties) {
+ if (Property.TABLE_FORMATTER_CLASS.getKey().equals(property.getKey())) {
+ if (!ReplicationTable.STATUS_FORMATTER_CLASS_NAME.equals(property.getValue())) {
+ log.info("Changing formatter for {} table from {} to {}", ReplicationTable.NAME, property.getValue(), ReplicationTable.STATUS_FORMATTER_CLASS_NAME);
+ try {
+ tops.setProperty(ReplicationTable.NAME, Property.TABLE_FORMATTER_CLASS.getKey(), ReplicationTable.STATUS_FORMATTER_CLASS_NAME);
+ } catch (AccumuloException | AccumuloSecurityException e) {
+ log.error("Could not set formatter on replication table", e);
+ return false;
+ }
+ }
+
+ formatterConfigured = true;
+
+ // Don't need to keep iterating over the properties after we found the one we were looking for
+ break;
+ }
+ }
+
+ if (!formatterConfigured) {
+ try {
+ tops.setProperty(ReplicationTable.NAME, Property.TABLE_FORMATTER_CLASS.getKey(), ReplicationTable.STATUS_FORMATTER_CLASS_NAME);
+ } catch (AccumuloException | AccumuloSecurityException e) {
+ log.error("Could not set formatter on replication table", e);
+ return false;
+ }
+ }
+
+ log.debug("Successfully configured replication table");
+
+ return true;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/server/base/src/test/java/org/apache/accumulo/server/replication/ReplicationTableTest.java
----------------------------------------------------------------------
diff --git a/server/base/src/test/java/org/apache/accumulo/server/replication/ReplicationTableTest.java b/server/base/src/test/java/org/apache/accumulo/server/replication/ReplicationTableTest.java
index bb80c81..44ac3fe 100644
--- a/server/base/src/test/java/org/apache/accumulo/server/replication/ReplicationTableTest.java
+++ b/server/base/src/test/java/org/apache/accumulo/server/replication/ReplicationTableTest.java
@@ -32,6 +32,7 @@ import org.apache.accumulo.core.iterators.Combiner;
import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
+import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.hadoop.io.Text;
import org.junit.Assert;
import org.junit.Before;
@@ -62,7 +63,7 @@ public class ReplicationTableTest {
public void replicationTableCreated() {
TableOperations tops = conn.tableOperations();
Assert.assertFalse(tops.exists(ReplicationTable.NAME));
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
Assert.assertTrue(tops.exists(ReplicationTable.NAME));
}
@@ -72,14 +73,14 @@ public class ReplicationTableTest {
Assert.assertFalse(tops.exists(ReplicationTable.NAME));
// Create the table
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
// Make sure it exists and save off the id
Assert.assertTrue(tops.exists(ReplicationTable.NAME));
String tableId = tops.tableIdMap().get(ReplicationTable.NAME);
// Try to make it again, should return quickly
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
Assert.assertTrue(tops.exists(ReplicationTable.NAME));
// Verify we have the same table as previously
@@ -96,7 +97,7 @@ public class ReplicationTableTest {
Assert.assertFalse(iterators.containsKey(ReplicationTable.COMBINER_NAME));
- ReplicationTable.configureReplicationTable(conn);
+ ReplicationUtil.configureReplicationTable(conn);
// After configure the iterator should be set
iterators = tops.listIterators(ReplicationTable.NAME);
@@ -123,7 +124,7 @@ public class ReplicationTableTest {
tops.delete(ReplicationTable.NAME);
}
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
Set<String> iters = tops.listIterators(ReplicationTable.NAME).keySet();
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
----------------------------------------------------------------------
diff --git a/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java b/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
index 9646be9..9928d3c 100644
--- a/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
+++ b/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogs.java
@@ -45,6 +45,7 @@ import org.apache.accumulo.core.metadata.MetadataTable;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.ReplicationSection;
import org.apache.accumulo.core.protobuf.ProtobufUtil;
import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
+import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.StatusUtil;
import org.apache.accumulo.core.replication.proto.Replication.Status;
import org.apache.accumulo.core.security.Authorizations;
@@ -58,7 +59,6 @@ import org.apache.accumulo.core.zookeeper.ZooUtil;
import org.apache.accumulo.server.ServerConstants;
import org.apache.accumulo.server.conf.ServerConfigurationFactory;
import org.apache.accumulo.server.fs.VolumeManager;
-import org.apache.accumulo.server.replication.ReplicationTable;
import org.apache.accumulo.server.security.SystemCredentials;
import org.apache.accumulo.server.util.MetadataTableUtil;
import org.apache.accumulo.server.zookeeper.ZooReaderWriter;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
----------------------------------------------------------------------
diff --git a/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java b/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
index f943ac1..bbb2010 100644
--- a/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
+++ b/server/gc/src/main/java/org/apache/accumulo/gc/SimpleGarbageCollector.java
@@ -62,6 +62,7 @@ import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.DataFileColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ScanFileColumnFamily;
import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
+import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.proto.Replication.Status;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.Credentials;
@@ -87,7 +88,6 @@ import org.apache.accumulo.server.fs.VolumeManager;
import org.apache.accumulo.server.fs.VolumeManager.FileType;
import org.apache.accumulo.server.fs.VolumeManagerImpl;
import org.apache.accumulo.server.fs.VolumeUtil;
-import org.apache.accumulo.server.replication.ReplicationTable;
import org.apache.accumulo.server.security.SystemCredentials;
import org.apache.accumulo.server.tables.TableManager;
import org.apache.accumulo.server.util.Halt;
@@ -756,7 +756,7 @@ public class SimpleGarbageCollector implements Iface {
HostAndPort result = HostAndPort.fromParts(opts.getAddress(), port);
log.debug("Starting garbage collector listening on " + result);
try {
- return TServerUtils.startTServer(result, processor, this.getClass().getSimpleName(), "GC Monitor Service", 2,
+ return TServerUtils.startTServer(result, processor, this.getClass().getSimpleName(), "GC Monitor Service", 2,
config.getCount(Property.GENERAL_SIMPLETIMER_THREADPOOL_SIZE), 1000, maxMessageSize, SslConnectionParams.forServer(config), 0).address;
} catch (Exception ex) {
log.fatal(ex, ex);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/server/gc/src/main/java/org/apache/accumulo/gc/replication/CloseWriteAheadLogReferences.java
----------------------------------------------------------------------
diff --git a/server/gc/src/main/java/org/apache/accumulo/gc/replication/CloseWriteAheadLogReferences.java b/server/gc/src/main/java/org/apache/accumulo/gc/replication/CloseWriteAheadLogReferences.java
index 74da72d..d13edf7 100644
--- a/server/gc/src/main/java/org/apache/accumulo/gc/replication/CloseWriteAheadLogReferences.java
+++ b/server/gc/src/main/java/org/apache/accumulo/gc/replication/CloseWriteAheadLogReferences.java
@@ -38,12 +38,12 @@ import org.apache.accumulo.core.metadata.MetadataTable;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.ReplicationSection;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.LogColumnFamily;
+import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.StatusUtil;
import org.apache.accumulo.core.replication.proto.Replication.Status;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.Credentials;
import org.apache.accumulo.core.tabletserver.log.LogEntry;
-import org.apache.accumulo.server.replication.ReplicationTable;
import org.apache.accumulo.trace.instrument.Span;
import org.apache.accumulo.trace.instrument.Trace;
import org.apache.hadoop.fs.Path;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java
----------------------------------------------------------------------
diff --git a/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java b/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java
index 74b0919..cd69efb 100644
--- a/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java
+++ b/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectWriteAheadLogsTest.java
@@ -50,11 +50,12 @@ import org.apache.accumulo.core.metadata.MetadataTable;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.ReplicationSection;
import org.apache.accumulo.core.protobuf.ProtobufUtil;
import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
+import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.StatusUtil;
import org.apache.accumulo.core.replication.proto.Replication.Status;
import org.apache.accumulo.core.security.Credentials;
import org.apache.accumulo.server.fs.VolumeManager;
-import org.apache.accumulo.server.replication.ReplicationTable;
+import org.apache.accumulo.server.replication.ReplicationUtil;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
@@ -116,12 +117,14 @@ public class GarbageCollectWriteAheadLogsTest {
@Test
public void testMapServersToFiles() {
+ // @formatter:off
/*
* Test fileToServerMap:
* /dir1/server1/uuid1 -> server1 (new-style)
* /dir1/uuid2 -> "" (old-style)
* /dir3/server3/uuid3 -> server3 (new-style)
*/
+ // @formatter:on
Map<Path,String> fileToServerMap = new java.util.HashMap<Path,String>();
Path path1 = new Path(new Path(DIR_1_PATH, "server1"), UUID1);
fileToServerMap.put(path1, "server1"); // new-style
@@ -129,20 +132,24 @@ public class GarbageCollectWriteAheadLogsTest {
fileToServerMap.put(path2, ""); // old-style
Path path3 = new Path(new Path(DIR_3_PATH, "server3"), UUID3);
fileToServerMap.put(path3, "server3"); // old-style
+ // @formatter:off
/*
* Test nameToFileMap:
* uuid1 -> /dir1/server1/uuid1
* uuid3 -> /dir3/server3/uuid3
*/
+ // @formatter:on
Map<String,Path> nameToFileMap = new java.util.HashMap<String,Path>();
nameToFileMap.put(UUID1, path1);
nameToFileMap.put(UUID3, path3);
- /**
+ // @formatter:off
+ /*
* Expected map:
* server1 -> [ /dir1/server1/uuid1 ]
* server3 -> [ /dir3/server3/uuid3 ]
*/
+ // @formatter:on
Map<String,ArrayList<Path>> result = GarbageCollectWriteAheadLogs.mapServersToFiles(fileToServerMap, nameToFileMap);
assertEquals(2, result.size());
ArrayList<Path> list1 = result.get("server1");
@@ -157,6 +164,7 @@ public class GarbageCollectWriteAheadLogsTest {
boolean isDir = (size == 0);
return new FileStatus(size, isDir, 3, BLOCK_SIZE, modTime, path);
}
+
private void mockListStatus(Path dir, FileStatus... fileStatuses) throws Exception {
expect(volMgr.listStatus(dir)).andReturn(fileStatuses);
}
@@ -164,6 +172,7 @@ public class GarbageCollectWriteAheadLogsTest {
@Test
public void testScanServers_NewStyle() throws Exception {
String[] walDirs = new String[] {"/dir1", "/dir2", "/dir3"};
+ // @formatter:off
/*
* Test directory layout:
* /dir1/
@@ -176,6 +185,7 @@ public class GarbageCollectWriteAheadLogsTest {
* server3/
* uuid3
*/
+ // @formatter:on
Path serverDir1Path = new Path(DIR_1_PATH, "server1");
FileStatus serverDir1 = makeFileStatus(0, serverDir1Path);
Path subDir2Path = new Path(DIR_1_PATH, "subdir2");
@@ -199,19 +209,23 @@ public class GarbageCollectWriteAheadLogsTest {
Map<String,Path> nameToFileMap = new java.util.HashMap<String,Path>();
int count = gcwal.scanServers(walDirs, fileToServerMap, nameToFileMap);
assertEquals(3, count);
+ // @formatter:off
/*
* Expected fileToServerMap:
* /dir1/server1/uuid1 -> server1
* /dir3/server3/uuid3 -> server3
*/
+ // @formatter:on
assertEquals(2, fileToServerMap.size());
assertEquals("server1", fileToServerMap.get(path1));
assertEquals("server3", fileToServerMap.get(path3));
+ // @formatter:off
/*
* Expected nameToFileMap:
* uuid1 -> /dir1/server1/uuid1
* uuid3 -> /dir3/server3/uuid3
*/
+ // @formatter:on
assertEquals(2, nameToFileMap.size());
assertEquals(path1, nameToFileMap.get(UUID1));
assertEquals(path3, nameToFileMap.get(UUID3));
@@ -219,6 +233,7 @@ public class GarbageCollectWriteAheadLogsTest {
@Test
public void testScanServers_OldStyle() throws Exception {
+ // @formatter:off
/*
* Test directory layout:
* /dir1/
@@ -226,6 +241,7 @@ public class GarbageCollectWriteAheadLogsTest {
* /dir3/
* uuid3
*/
+ // @formatter:on
String[] walDirs = new String[] {"/dir1", "/dir3"};
Path serverFile1Path = new Path(DIR_1_PATH, UUID1);
FileStatus serverFile1 = makeFileStatus(100, serverFile1Path);
@@ -242,19 +258,23 @@ public class GarbageCollectWriteAheadLogsTest {
* Expect only a single server, the non-server entry for upgrade WALs
*/
assertEquals(1, count);
+ // @formatter:off
/*
* Expected fileToServerMap:
* /dir1/uuid1 -> ""
* /dir3/uuid3 -> ""
*/
+ // @formatter:on
assertEquals(2, fileToServerMap.size());
assertEquals("", fileToServerMap.get(serverFile1Path));
assertEquals("", fileToServerMap.get(serverFile3Path));
+ // @formatter:off
/*
* Expected nameToFileMap:
* uuid1 -> /dir1/uuid1
* uuid3 -> /dir3/uuid3
*/
+ // @formatter:on
assertEquals(2, nameToFileMap.size());
assertEquals(serverFile1Path, nameToFileMap.get(UUID1));
assertEquals(serverFile3Path, nameToFileMap.get(UUID3));
@@ -263,6 +283,7 @@ public class GarbageCollectWriteAheadLogsTest {
@Test
public void testGetSortedWALogs() throws Exception {
String[] recoveryDirs = new String[] {"/dir1", "/dir2", "/dir3"};
+ // @formatter:off
/*
* Test directory layout:
* /dir1/
@@ -272,6 +293,7 @@ public class GarbageCollectWriteAheadLogsTest {
* /dir3/
* uuid3
*/
+ // @formatter:on
expect(volMgr.exists(DIR_1_PATH)).andReturn(true);
expect(volMgr.exists(DIR_2_PATH)).andReturn(false);
expect(volMgr.exists(DIR_3_PATH)).andReturn(true);
@@ -285,11 +307,13 @@ public class GarbageCollectWriteAheadLogsTest {
replay(volMgr);
Map<String,Path> sortedWalogs = gcwal.getSortedWALogs(recoveryDirs);
- /**
+ // @formatter:off
+ /*
* Expected map:
* uuid1 -> /dir1/uuid1
* uuid3 -> /dir3/uuid3
*/
+ // @formatter:on
assertEquals(2, sortedWalogs.size());
assertEquals(path1, sortedWalogs.get(UUID1));
assertEquals(path3, sortedWalogs.get(UUID3));
@@ -360,7 +384,7 @@ public class GarbageCollectWriteAheadLogsTest {
GarbageCollectWriteAheadLogs gcWALs = new GarbageCollectWriteAheadLogs(inst, volMgr, false);
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
long file1CreateTime = System.currentTimeMillis();
long file2CreateTime = file1CreateTime + 50;
@@ -404,7 +428,7 @@ public class GarbageCollectWriteAheadLogsTest {
GarbageCollectWriteAheadLogs gcWALs = new GarbageCollectWriteAheadLogs(inst, volMgr, false);
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
long file1CreateTime = System.currentTimeMillis();
long file2CreateTime = file1CreateTime + 50;
@@ -464,7 +488,7 @@ public class GarbageCollectWriteAheadLogsTest {
public void fetchesReplicationEntriesFromMetadataAndReplicationTables() throws Exception {
Instance inst = new MockInstance(testName.getMethodName());
Connector conn = inst.getConnector("root", new PasswordToken(""));
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
long walCreateTime = System.currentTimeMillis();
String wal = "hdfs://localhost:8020/accumulo/wal/tserver+port/123456-1234-1234-12345678";
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/server/gc/src/test/java/org/apache/accumulo/gc/replication/CloseWriteAheadLogReferencesTest.java
----------------------------------------------------------------------
diff --git a/server/gc/src/test/java/org/apache/accumulo/gc/replication/CloseWriteAheadLogReferencesTest.java b/server/gc/src/test/java/org/apache/accumulo/gc/replication/CloseWriteAheadLogReferencesTest.java
index 13d2263..4d99495 100644
--- a/server/gc/src/test/java/org/apache/accumulo/gc/replication/CloseWriteAheadLogReferencesTest.java
+++ b/server/gc/src/test/java/org/apache/accumulo/gc/replication/CloseWriteAheadLogReferencesTest.java
@@ -47,12 +47,13 @@ import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.LogColumnFamily;
import org.apache.accumulo.core.protobuf.ProtobufUtil;
import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
+import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.StatusUtil;
import org.apache.accumulo.core.replication.proto.Replication.Status;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.Credentials;
import org.apache.accumulo.core.tabletserver.log.LogEntry;
-import org.apache.accumulo.server.replication.ReplicationTable;
+import org.apache.accumulo.server.replication.ReplicationUtil;
import org.apache.hadoop.io.Text;
import org.easymock.IAnswer;
import org.junit.Assert;
@@ -308,7 +309,7 @@ public class CloseWriteAheadLogReferencesTest {
Instance inst = new MockInstance(testName.getMethodName());
Connector conn = inst.getConnector("root", new PasswordToken(""));
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
BatchWriter bw = conn.createBatchWriter(MetadataTable.NAME, new BatchWriterConfig());
Mutation m = new Mutation(ReplicationSection.getRowPrefix() + "file:/accumulo/wal/tserver+port/12345");
m.put(ReplicationSection.COLF, new Text("1"), StatusUtil.fileCreatedValue(System.currentTimeMillis()));
@@ -330,7 +331,7 @@ public class CloseWriteAheadLogReferencesTest {
Instance inst = new MockInstance(testName.getMethodName());
Connector conn = inst.getConnector("root", new PasswordToken(""));
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
BatchWriter bw = conn.createBatchWriter(MetadataTable.NAME, new BatchWriterConfig());
Mutation m = new Mutation(ReplicationSection.getRowPrefix() + file);
m.put(ReplicationSection.COLF, new Text("1"), StatusUtil.fileCreatedValue(System.currentTimeMillis()));
@@ -352,7 +353,7 @@ public class CloseWriteAheadLogReferencesTest {
Instance inst = new MockInstance(testName.getMethodName());
Connector conn = inst.getConnector("root", new PasswordToken(""));
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
BatchWriter bw = conn.createBatchWriter(ReplicationTable.NAME, new BatchWriterConfig());
Mutation m = new Mutation(file);
StatusSection.add(m, new Text("1"), ProtobufUtil.toValue(StatusUtil.ingestedUntil(1000)));
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/server/master/src/main/java/org/apache/accumulo/master/metrics/ReplicationMetrics.java
----------------------------------------------------------------------
diff --git a/server/master/src/main/java/org/apache/accumulo/master/metrics/ReplicationMetrics.java b/server/master/src/main/java/org/apache/accumulo/master/metrics/ReplicationMetrics.java
index 06a653d..d4a9af2 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/metrics/ReplicationMetrics.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/metrics/ReplicationMetrics.java
@@ -29,7 +29,7 @@ import org.apache.accumulo.core.client.admin.TableOperations;
import org.apache.accumulo.core.client.impl.MasterClient;
import org.apache.accumulo.core.master.thrift.MasterClientService;
import org.apache.accumulo.core.master.thrift.MasterMonitorInfo;
-import org.apache.accumulo.core.replication.ReplicationConstants;
+import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.ReplicationTarget;
import org.apache.accumulo.server.client.HdfsZooInstance;
import org.apache.accumulo.server.metrics.AbstractMetricsImpl;
@@ -61,7 +61,7 @@ public class ReplicationMetrics extends AbstractMetricsImpl implements Replicati
@Override
public int getNumFilesPendingReplication() {
- if (!tops.exists(ReplicationConstants.TABLE_NAME)) {
+ if (!tops.exists(ReplicationTable.NAME)) {
return 0;
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/server/master/src/main/java/org/apache/accumulo/master/replication/DistributedWorkQueueWorkAssigner.java
----------------------------------------------------------------------
diff --git a/server/master/src/main/java/org/apache/accumulo/master/replication/DistributedWorkQueueWorkAssigner.java b/server/master/src/main/java/org/apache/accumulo/master/replication/DistributedWorkQueueWorkAssigner.java
index 3810372..8e1d036 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/replication/DistributedWorkQueueWorkAssigner.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/replication/DistributedWorkQueueWorkAssigner.java
@@ -32,13 +32,13 @@ import org.apache.accumulo.core.protobuf.ProtobufUtil;
import org.apache.accumulo.core.replication.ReplicationConstants;
import org.apache.accumulo.core.replication.ReplicationSchema.OrderSection;
import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
+import org.apache.accumulo.core.replication.ReplicationTable;
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.util.UtilWaitThread;
import org.apache.accumulo.core.zookeeper.ZooUtil;
import org.apache.accumulo.server.replication.DistributedWorkQueueWorkAssignerHelper;
-import org.apache.accumulo.server.replication.ReplicationTable;
import org.apache.accumulo.server.replication.WorkAssigner;
import org.apache.accumulo.server.zookeeper.DistributedWorkQueue;
import org.apache.accumulo.server.zookeeper.ZooCache;
@@ -252,14 +252,18 @@ public abstract class DistributedWorkQueueWorkAssigner implements WorkAssigner {
/**
* Queue the given work for the target
- * @param path File to replicate
- * @param target Target for the work
+ *
+ * @param path
+ * File to replicate
+ * @param target
+ * Target for the work
* @return True if the work was queued, false otherwise
*/
protected abstract boolean queueWork(Path path, ReplicationTarget target);
/**
- * @param target Target for the work
+ * @param target
+ * Target for the work
* @return Queued work for the given target
*/
protected abstract Set<String> getQueuedWork(ReplicationTarget target);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/server/master/src/main/java/org/apache/accumulo/master/replication/FinishedWorkUpdater.java
----------------------------------------------------------------------
diff --git a/server/master/src/main/java/org/apache/accumulo/master/replication/FinishedWorkUpdater.java b/server/master/src/main/java/org/apache/accumulo/master/replication/FinishedWorkUpdater.java
index b816eab..8bc18eb 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/replication/FinishedWorkUpdater.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/replication/FinishedWorkUpdater.java
@@ -37,10 +37,10 @@ import org.apache.accumulo.core.iterators.user.WholeRowIterator;
import org.apache.accumulo.core.protobuf.ProtobufUtil;
import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
+import org.apache.accumulo.core.replication.ReplicationTable;
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.server.replication.ReplicationTable;
import org.apache.hadoop.io.Text;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -48,8 +48,7 @@ import org.slf4j.LoggerFactory;
import com.google.protobuf.InvalidProtocolBufferException;
/**
- * Update the status record in the replication table with work
- * that has been replicated to each configured peer.
+ * Update the status record in the replication table with work that has been replicated to each configured peer.
*/
public class FinishedWorkUpdater implements Runnable {
private static final Logger log = LoggerFactory.getLogger(FinishedWorkUpdater.class);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/server/master/src/main/java/org/apache/accumulo/master/replication/RemoveCompleteReplicationRecords.java
----------------------------------------------------------------------
diff --git a/server/master/src/main/java/org/apache/accumulo/master/replication/RemoveCompleteReplicationRecords.java b/server/master/src/main/java/org/apache/accumulo/master/replication/RemoveCompleteReplicationRecords.java
index c670d51..ef76916 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/replication/RemoveCompleteReplicationRecords.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/replication/RemoveCompleteReplicationRecords.java
@@ -39,10 +39,10 @@ import org.apache.accumulo.core.iterators.user.WholeRowIterator;
import org.apache.accumulo.core.replication.ReplicationSchema.OrderSection;
import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
+import org.apache.accumulo.core.replication.ReplicationTable;
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.server.replication.ReplicationTable;
import org.apache.hadoop.io.Text;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -182,7 +182,7 @@ public class RemoveCompleteReplicationRecords implements Runnable {
Long timeClosed = tableToTimeCreated.get(tableId);
if (null == timeClosed) {
tableToTimeCreated.put(tableId, status.getCreatedTime());
- } else if (timeClosed != status.getCreatedTime()){
+ } else if (timeClosed != status.getCreatedTime()) {
log.warn("Found multiple values for timeClosed for {}: {} and {}", row, timeClosed, status.getCreatedTime());
}
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/server/master/src/main/java/org/apache/accumulo/master/replication/StatusMaker.java
----------------------------------------------------------------------
diff --git a/server/master/src/main/java/org/apache/accumulo/master/replication/StatusMaker.java b/server/master/src/main/java/org/apache/accumulo/master/replication/StatusMaker.java
index 68dd005..c0f2a6c 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/replication/StatusMaker.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/replication/StatusMaker.java
@@ -33,9 +33,10 @@ import org.apache.accumulo.core.metadata.schema.MetadataSchema.ReplicationSectio
import org.apache.accumulo.core.protobuf.ProtobufUtil;
import org.apache.accumulo.core.replication.ReplicationSchema.OrderSection;
import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
+import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.proto.Replication.Status;
import org.apache.accumulo.core.security.Authorizations;
-import org.apache.accumulo.server.replication.ReplicationTable;
+import org.apache.accumulo.server.replication.ReplicationUtil;
import org.apache.accumulo.trace.instrument.Span;
import org.apache.accumulo.trace.instrument.Trace;
import org.apache.hadoop.io.Text;
@@ -45,8 +46,8 @@ import org.slf4j.LoggerFactory;
import com.google.protobuf.InvalidProtocolBufferException;
/**
- * Reads replication records from the metadata table and creates status records in the replication table. Deletes the record
- * from the metadata table when it's closed.
+ * Reads replication records from the metadata table and creates status records in the replication table. Deletes the record from the metadata table when it's
+ * closed.
*/
public class StatusMaker {
private static final Logger log = LoggerFactory.getLogger(StatusMaker.class);
@@ -64,7 +65,7 @@ public class StatusMaker {
* Not for public use -- visible only for testing
* <p>
* Used to read records from a table other than 'metadata'
- *
+ *
* @param table
* The table to read from
*/
@@ -92,7 +93,7 @@ public class StatusMaker {
// Get a writer to the replication table
if (null == replicationWriter) {
// Ensures table exists and is properly configured
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
try {
setBatchWriter(ReplicationTable.getBatchWriter(conn));
} catch (TableNotFoundException e) {
@@ -105,7 +106,6 @@ public class StatusMaker {
MetadataSchema.ReplicationSection.getFile(entry.getKey(), file);
MetadataSchema.ReplicationSection.getTableId(entry.getKey(), tableId);
-
Status status;
try {
status = Status.parseFrom(entry.getValue().get());
@@ -180,13 +180,17 @@ public class StatusMaker {
}
/**
- * Create a record to track when the file was closed to ensure that replication preference
- * is given to files that have been closed the longest and allow the work assigner to try to
- * replicate in order that data was ingested (avoid replay in different order)
- * @param file File being replicated
- * @param tableId Table ID the file was used by
- * @param stat Status msg
- * @param value Serialized version of the Status msg
+ * Create a record to track when the file was closed to ensure that replication preference is given to files that have been closed the longest and allow the
+ * work assigner to try to replicate in order that data was ingested (avoid replay in different order)
+ *
+ * @param file
+ * File being replicated
+ * @param tableId
+ * Table ID the file was used by
+ * @param stat
+ * Status msg
+ * @param value
+ * Serialized version of the Status msg
*/
protected boolean addOrderRecord(Text file, Text tableId, Status stat, Value value) {
try {
@@ -219,15 +223,14 @@ public class StatusMaker {
}
/**
- * Because there is only one active Master, and thus one active StatusMaker, the only
- * safe time that we can issue the delete for a Status which is closed is immediately
- * after writing it to the replication table.
+ * Because there is only one active Master, and thus one active StatusMaker, the only safe time that we can issue the delete for a Status which is closed is
+ * immediately after writing it to the replication table.
* <p>
- * If we try to defer and delete these entries in another thread/process, we will have
- * no assurance that the Status message was propagated to the replication table. It is
- * easiest, in terms of concurrency, to do this all in one step.
- *
- * @param k The Key to delete
+ * If we try to defer and delete these entries in another thread/process, we will have no assurance that the Status message was propagated to the replication
+ * table. It is easiest, in terms of concurrency, to do this all in one step.
+ *
+ * @param k
+ * The Key to delete
*/
protected void deleteStatusRecord(Key k) {
log.debug("Deleting {} from metadata table as it's no longer needed", k.toStringNoTruncate());
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/server/master/src/main/java/org/apache/accumulo/master/replication/WorkMaker.java
----------------------------------------------------------------------
diff --git a/server/master/src/main/java/org/apache/accumulo/master/replication/WorkMaker.java b/server/master/src/main/java/org/apache/accumulo/master/replication/WorkMaker.java
index eabcc84..95f607f 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/replication/WorkMaker.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/replication/WorkMaker.java
@@ -33,12 +33,12 @@ import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.replication.ReplicationSchema;
import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
+import org.apache.accumulo.core.replication.ReplicationTable;
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.server.conf.ServerConfigurationFactory;
import org.apache.accumulo.server.conf.TableConfiguration;
-import org.apache.accumulo.server.replication.ReplicationTable;
import org.apache.accumulo.trace.instrument.Span;
import org.apache.accumulo.trace.instrument.Trace;
import org.apache.hadoop.io.DataOutputBuffer;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/server/master/src/test/java/org/apache/accumulo/master/replication/FinishedWorkUpdaterTest.java
----------------------------------------------------------------------
diff --git a/server/master/src/test/java/org/apache/accumulo/master/replication/FinishedWorkUpdaterTest.java b/server/master/src/test/java/org/apache/accumulo/master/replication/FinishedWorkUpdaterTest.java
index 7633823..f8cc775 100644
--- a/server/master/src/test/java/org/apache/accumulo/master/replication/FinishedWorkUpdaterTest.java
+++ b/server/master/src/test/java/org/apache/accumulo/master/replication/FinishedWorkUpdaterTest.java
@@ -31,9 +31,10 @@ import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.protobuf.ProtobufUtil;
import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
+import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.ReplicationTarget;
import org.apache.accumulo.core.replication.proto.Replication.Status;
-import org.apache.accumulo.server.replication.ReplicationTable;
+import org.apache.accumulo.server.replication.ReplicationUtil;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Rule;
@@ -43,7 +44,7 @@ import org.junit.rules.TestName;
import com.google.common.collect.Iterables;
/**
- *
+ *
*/
public class FinishedWorkUpdaterTest {
@@ -67,7 +68,7 @@ public class FinishedWorkUpdaterTest {
@Test
public void recordsWithProgressUpdateBothTables() throws Exception {
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
String file = "/accumulo/wals/tserver+port/" + UUID.randomUUID();
Status stat = Status.newBuilder().setBegin(100).setEnd(200).setClosed(true).setInfiniteEnd(false).build();
@@ -97,15 +98,17 @@ public class FinishedWorkUpdaterTest {
@Test
public void chooseMinimumBeginOffset() throws Exception {
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
String file = "/accumulo/wals/tserver+port/" + UUID.randomUUID();
+ // @formatter:off
Status stat1 = Status.newBuilder().setBegin(100).setEnd(1000).setClosed(true).setInfiniteEnd(false).build(),
stat2 = Status.newBuilder().setBegin(500).setEnd(1000).setClosed(true).setInfiniteEnd(false).build(),
stat3 = Status.newBuilder().setBegin(1).setEnd(1000).setClosed(true).setInfiniteEnd(false).build();
ReplicationTarget target1 = new ReplicationTarget("peer1", "table1", "1"),
target2 = new ReplicationTarget("peer2", "table2", "1"),
target3 = new ReplicationTarget("peer3", "table3", "1");
+ // @formatter:on
// Create a single work record for a file to some peer
BatchWriter bw = ReplicationTable.getBatchWriter(conn);
@@ -133,15 +136,17 @@ public class FinishedWorkUpdaterTest {
@Test
public void chooseMinimumBeginOffsetInfiniteEnd() throws Exception {
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
String file = "/accumulo/wals/tserver+port/" + UUID.randomUUID();
+ // @formatter:off
Status stat1 = Status.newBuilder().setBegin(100).setEnd(1000).setClosed(true).setInfiniteEnd(true).build(),
stat2 = Status.newBuilder().setBegin(1).setEnd(1000).setClosed(true).setInfiniteEnd(true).build(),
stat3 = Status.newBuilder().setBegin(500).setEnd(1000).setClosed(true).setInfiniteEnd(true).build();
ReplicationTarget target1 = new ReplicationTarget("peer1", "table1", "1"),
target2 = new ReplicationTarget("peer2", "table2", "1"),
target3 = new ReplicationTarget("peer3", "table3", "1");
+ // @formatter:on
// Create a single work record for a file to some peer
BatchWriter bw = ReplicationTable.getBatchWriter(conn);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/server/master/src/test/java/org/apache/accumulo/master/replication/RemoveCompleteReplicationRecordsTest.java
----------------------------------------------------------------------
diff --git a/server/master/src/test/java/org/apache/accumulo/master/replication/RemoveCompleteReplicationRecordsTest.java b/server/master/src/test/java/org/apache/accumulo/master/replication/RemoveCompleteReplicationRecordsTest.java
index ec19a94..2255e21 100644
--- a/server/master/src/test/java/org/apache/accumulo/master/replication/RemoveCompleteReplicationRecordsTest.java
+++ b/server/master/src/test/java/org/apache/accumulo/master/replication/RemoveCompleteReplicationRecordsTest.java
@@ -37,10 +37,11 @@ import org.apache.accumulo.core.protobuf.ProtobufUtil;
import org.apache.accumulo.core.replication.ReplicationSchema.OrderSection;
import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
+import org.apache.accumulo.core.replication.ReplicationTable;
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.server.replication.ReplicationTable;
+import org.apache.accumulo.server.replication.ReplicationUtil;
import org.apache.hadoop.io.Text;
import org.easymock.EasyMock;
import org.junit.Assert;
@@ -72,7 +73,7 @@ public class RemoveCompleteReplicationRecordsTest {
@Test
public void notYetReplicationRecordsIgnored() throws Exception {
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
BatchWriter bw = ReplicationTable.getBatchWriter(conn);
int numRecords = 3;
for (int i = 0; i < numRecords; i++) {
@@ -102,7 +103,7 @@ public class RemoveCompleteReplicationRecordsTest {
@Test
public void partiallyReplicatedRecordsIgnored() throws Exception {
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
BatchWriter bw = ReplicationTable.getBatchWriter(conn);
int numRecords = 3;
Status.Builder builder = Status.newBuilder();
@@ -137,7 +138,7 @@ public class RemoveCompleteReplicationRecordsTest {
@Test
public void replicatedClosedWorkRecordsAreNotRemovedWithoutClosedStatusRecords() throws Exception {
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
BatchWriter replBw = ReplicationTable.getBatchWriter(conn);
int numRecords = 3;
@@ -190,7 +191,7 @@ public class RemoveCompleteReplicationRecordsTest {
@Test
public void replicatedClosedRowsAreRemoved() throws Exception {
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
BatchWriter replBw = ReplicationTable.getBatchWriter(conn);
int numRecords = 3;
@@ -282,7 +283,7 @@ public class RemoveCompleteReplicationRecordsTest {
@Test
public void partiallyReplicatedEntriesPrecludeRowDeletion() throws Exception {
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
BatchWriter replBw = ReplicationTable.getBatchWriter(conn);
int numRecords = 3;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/server/master/src/test/java/org/apache/accumulo/master/replication/SequentialWorkAssignerTest.java
----------------------------------------------------------------------
diff --git a/server/master/src/test/java/org/apache/accumulo/master/replication/SequentialWorkAssignerTest.java b/server/master/src/test/java/org/apache/accumulo/master/replication/SequentialWorkAssignerTest.java
index 3d96ea1..d5bcfbd 100644
--- a/server/master/src/test/java/org/apache/accumulo/master/replication/SequentialWorkAssignerTest.java
+++ b/server/master/src/test/java/org/apache/accumulo/master/replication/SequentialWorkAssignerTest.java
@@ -37,13 +37,14 @@ import org.apache.accumulo.core.protobuf.ProtobufUtil;
import org.apache.accumulo.core.replication.ReplicationConstants;
import org.apache.accumulo.core.replication.ReplicationSchema.OrderSection;
import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
+import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.ReplicationTarget;
import org.apache.accumulo.core.replication.proto.Replication.Status;
import org.apache.accumulo.core.security.Credentials;
import org.apache.accumulo.core.security.TablePermission;
import org.apache.accumulo.core.zookeeper.ZooUtil;
import org.apache.accumulo.server.replication.DistributedWorkQueueWorkAssignerHelper;
-import org.apache.accumulo.server.replication.ReplicationTable;
+import org.apache.accumulo.server.replication.ReplicationUtil;
import org.apache.accumulo.server.zookeeper.DistributedWorkQueue;
import org.apache.accumulo.server.zookeeper.ZooCache;
import org.apache.hadoop.io.Text;
@@ -85,7 +86,7 @@ public class SequentialWorkAssignerTest {
assigner.setConnector(conn);
// Create and grant ourselves write to the replication table
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.WRITE);
// Create two mutations, both of which need replication work done
@@ -158,7 +159,7 @@ public class SequentialWorkAssignerTest {
assigner.setConnector(conn);
// Create and grant ourselves write to the replication table
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.WRITE);
// Create two mutations, both of which need replication work done
@@ -238,7 +239,7 @@ public class SequentialWorkAssignerTest {
assigner.setConnector(conn);
// Create and grant ourselves write to the replication table
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.WRITE);
// Create two mutations, both of which need replication work done
@@ -359,7 +360,7 @@ public class SequentialWorkAssignerTest {
assigner.setConnector(conn);
// Create and grant ourselves write to the replication table
- ReplicationTable.create(conn);
+ ReplicationUtil.createReplicationTable(conn);
conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.WRITE);
// Create two mutations, both of which need replication work done
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f1a47b38/server/master/src/test/java/org/apache/accumulo/master/replication/StatusMakerTest.java
----------------------------------------------------------------------
diff --git a/server/master/src/test/java/org/apache/accumulo/master/replication/StatusMakerTest.java b/server/master/src/test/java/org/apache/accumulo/master/replication/StatusMakerTest.java
index 966e2bd..6a8fa4d 100644
--- a/server/master/src/test/java/org/apache/accumulo/master/replication/StatusMakerTest.java
+++ b/server/master/src/test/java/org/apache/accumulo/master/replication/StatusMakerTest.java
@@ -38,11 +38,11 @@ import org.apache.accumulo.core.metadata.schema.MetadataSchema.ReplicationSectio
import org.apache.accumulo.core.protobuf.ProtobufUtil;
import org.apache.accumulo.core.replication.ReplicationSchema.OrderSection;
import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
+import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.StatusUtil;
import org.apache.accumulo.core.replication.proto.Replication.Status;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.Credentials;
-import org.apache.accumulo.server.replication.ReplicationTable;
import org.apache.accumulo.server.util.ReplicationTableUtil;
import org.apache.hadoop.io.Text;
import org.junit.Assert;
[08/10] accumulo git commit: ACCUMULO-3147 Fix ReplicationServlet
Posted by ct...@apache.org.
ACCUMULO-3147 Fix ReplicationServlet
Properly show an informative banner message when the replication table is
offline.
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/5cf1888e
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/5cf1888e
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/5cf1888e
Branch: refs/heads/master
Commit: 5cf1888eb402dd3d22e94decedb75fb86a2264b1
Parents: 5753e16
Author: Christopher Tubbs <ct...@apache.org>
Authored: Fri Nov 7 15:12:39 2014 -0500
Committer: Christopher Tubbs <ct...@apache.org>
Committed: Fri Nov 7 15:12:39 2014 -0500
----------------------------------------------------------------------
.../apache/accumulo/monitor/servlets/ReplicationServlet.java | 5 +++++
1 file changed, 5 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/5cf1888e/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/ReplicationServlet.java
----------------------------------------------------------------------
diff --git a/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/ReplicationServlet.java b/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/ReplicationServlet.java
index 7d3cb14..94765f8 100644
--- a/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/ReplicationServlet.java
+++ b/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/ReplicationServlet.java
@@ -28,6 +28,7 @@ import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.admin.TableOperations;
import org.apache.accumulo.core.master.thrift.MasterMonitorInfo;
import org.apache.accumulo.core.replication.ReplicationConstants;
+import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.ReplicationTarget;
import org.apache.accumulo.core.security.Credentials;
import org.apache.accumulo.core.zookeeper.ZooUtil;
@@ -75,6 +76,10 @@ public class ReplicationServlet extends BasicServlet {
int totalWorkQueueSize = replicationUtil.getMaxReplicationThreads(systemProps, mmi);
TableOperations tops = conn.tableOperations();
+ if (!ReplicationTable.isOnline(conn)) {
+ banner(sb, "", "Replication table is offline");
+ return;
+ }
Table replicationStats = new Table("replicationStats", "Replication Status");
replicationStats.addSortableColumn("Table");
[09/10] accumulo git commit: Merge remote-tracking branch
'origin/master' into ACCUMULO-3147
Posted by ct...@apache.org.
Merge remote-tracking branch 'origin/master' into ACCUMULO-3147
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/bc624743
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/bc624743
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/bc624743
Branch: refs/heads/master
Commit: bc624743f5c703bb2b8f5a3a2d8918de72d478f5
Parents: 5cf1888 5fdbb81
Author: Christopher Tubbs <ct...@apache.org>
Authored: Fri Nov 7 15:13:29 2014 -0500
Committer: Christopher Tubbs <ct...@apache.org>
Committed: Fri Nov 7 15:13:29 2014 -0500
----------------------------------------------------------------------
.../accumulo/server/util/RandomizeVolumes.java | 15 ++++--
.../apache/accumulo/test/FairVolumeChooser.java | 48 ++++++++++++++++++++
.../test/RewriteTabletDirectoriesIT.java | 19 ++++++--
3 files changed, 75 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
[04/10] accumulo git commit: ACCUMULO-3147 Create replication table
at initialize, upgrade
Posted by ct...@apache.org.
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/gc/src/test/java/org/apache/accumulo/gc/replication/CloseWriteAheadLogReferencesTest.java
----------------------------------------------------------------------
diff --git a/server/gc/src/test/java/org/apache/accumulo/gc/replication/CloseWriteAheadLogReferencesTest.java b/server/gc/src/test/java/org/apache/accumulo/gc/replication/CloseWriteAheadLogReferencesTest.java
index 4d99495..86ac87e 100644
--- a/server/gc/src/test/java/org/apache/accumulo/gc/replication/CloseWriteAheadLogReferencesTest.java
+++ b/server/gc/src/test/java/org/apache/accumulo/gc/replication/CloseWriteAheadLogReferencesTest.java
@@ -53,7 +53,6 @@ import org.apache.accumulo.core.replication.proto.Replication.Status;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.Credentials;
import org.apache.accumulo.core.tabletserver.log.LogEntry;
-import org.apache.accumulo.server.replication.ReplicationUtil;
import org.apache.hadoop.io.Text;
import org.easymock.IAnswer;
import org.junit.Assert;
@@ -98,7 +97,7 @@ public class CloseWriteAheadLogReferencesTest {
data.add(Maps.immutableEntry(new Key(logEntry.getRow(), logEntry.getColumnFamily(), logEntry.getColumnQualifier()), new Value(logEntry.getValue())));
// Get a batchscanner, scan the tablets section, fetch only the logs
- expect(conn.createBatchScanner(MetadataTable.NAME, new Authorizations(), 4)).andReturn(bs);
+ expect(conn.createBatchScanner(MetadataTable.NAME, Authorizations.EMPTY, 4)).andReturn(bs);
bs.setRanges(Collections.singleton(TabletsSection.getRange()));
expectLastCall().once();
bs.fetchColumnFamily(LogColumnFamily.NAME);
@@ -141,7 +140,7 @@ public class CloseWriteAheadLogReferencesTest {
data.add(Maps.immutableEntry(new Key(logEntry.getRow(), logEntry.getColumnFamily(), logEntry.getColumnQualifier()), new Value(logEntry.getValue())));
// Get a batchscanner, scan the tablets section, fetch only the logs
- expect(conn.createBatchScanner(MetadataTable.NAME, new Authorizations(), 4)).andReturn(bs);
+ expect(conn.createBatchScanner(MetadataTable.NAME, Authorizations.EMPTY, 4)).andReturn(bs);
bs.setRanges(Collections.singleton(TabletsSection.getRange()));
expectLastCall().once();
bs.fetchColumnFamily(LogColumnFamily.NAME);
@@ -197,7 +196,7 @@ public class CloseWriteAheadLogReferencesTest {
data.add(Maps.immutableEntry(new Key(logEntry.getRow(), logEntry.getColumnFamily(), logEntry.getColumnQualifier()), new Value(logEntry.getValue())));
// Get a batchscanner, scan the tablets section, fetch only the logs
- expect(conn.createBatchScanner(MetadataTable.NAME, new Authorizations(), 4)).andReturn(bs);
+ expect(conn.createBatchScanner(MetadataTable.NAME, Authorizations.EMPTY, 4)).andReturn(bs);
bs.setRanges(Collections.singleton(TabletsSection.getRange()));
expectLastCall().once();
bs.fetchColumnFamily(LogColumnFamily.NAME);
@@ -277,7 +276,7 @@ public class CloseWriteAheadLogReferencesTest {
data.add(Maps.immutableEntry(new Key(logEntry.getRow(), logEntry.getColumnFamily(), logEntry.getColumnQualifier()), new Value(logEntry.getValue())));
// Get a batchscanner, scan the tablets section, fetch only the logs
- expect(conn.createBatchScanner(MetadataTable.NAME, new Authorizations(), 4)).andReturn(bs);
+ expect(conn.createBatchScanner(MetadataTable.NAME, Authorizations.EMPTY, 4)).andReturn(bs);
bs.setRanges(Collections.singleton(TabletsSection.getRange()));
expectLastCall().once();
bs.fetchColumnFamily(LogColumnFamily.NAME);
@@ -309,7 +308,6 @@ public class CloseWriteAheadLogReferencesTest {
Instance inst = new MockInstance(testName.getMethodName());
Connector conn = inst.getConnector("root", new PasswordToken(""));
- ReplicationUtil.createReplicationTable(conn);
BatchWriter bw = conn.createBatchWriter(MetadataTable.NAME, new BatchWriterConfig());
Mutation m = new Mutation(ReplicationSection.getRowPrefix() + "file:/accumulo/wal/tserver+port/12345");
m.put(ReplicationSection.COLF, new Text("1"), StatusUtil.fileCreatedValue(System.currentTimeMillis()));
@@ -331,7 +329,6 @@ public class CloseWriteAheadLogReferencesTest {
Instance inst = new MockInstance(testName.getMethodName());
Connector conn = inst.getConnector("root", new PasswordToken(""));
- ReplicationUtil.createReplicationTable(conn);
BatchWriter bw = conn.createBatchWriter(MetadataTable.NAME, new BatchWriterConfig());
Mutation m = new Mutation(ReplicationSection.getRowPrefix() + file);
m.put(ReplicationSection.COLF, new Text("1"), StatusUtil.fileCreatedValue(System.currentTimeMillis()));
@@ -353,8 +350,7 @@ public class CloseWriteAheadLogReferencesTest {
Instance inst = new MockInstance(testName.getMethodName());
Connector conn = inst.getConnector("root", new PasswordToken(""));
- ReplicationUtil.createReplicationTable(conn);
- BatchWriter bw = conn.createBatchWriter(ReplicationTable.NAME, new BatchWriterConfig());
+ BatchWriter bw = ReplicationTable.getBatchWriter(conn);
Mutation m = new Mutation(file);
StatusSection.add(m, new Text("1"), ProtobufUtil.toValue(StatusUtil.ingestedUntil(1000)));
bw.addMutation(m);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/master/src/main/java/org/apache/accumulo/master/FateServiceHandler.java
----------------------------------------------------------------------
diff --git a/server/master/src/main/java/org/apache/accumulo/master/FateServiceHandler.java b/server/master/src/main/java/org/apache/accumulo/master/FateServiceHandler.java
index 41ca911..891f267 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/FateServiceHandler.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/FateServiceHandler.java
@@ -16,6 +16,7 @@
*/
package org.apache.accumulo.master;
+import static org.apache.accumulo.master.util.TableValidators.NOT_METADATA;
import static org.apache.accumulo.master.util.TableValidators.NOT_ROOT_ID;
import static org.apache.accumulo.master.util.TableValidators.NOT_SYSTEM;
import static org.apache.accumulo.master.util.TableValidators.VALID_ID;
@@ -318,7 +319,7 @@ class FateServiceHandler implements FateService.Iface {
}
case TABLE_DELETE_RANGE: {
TableOperation tableOp = TableOperation.DELETE_RANGE;
- String tableName = validateTableNameArgument(arguments.get(0), tableOp, NOT_SYSTEM);
+ String tableName = validateTableNameArgument(arguments.get(0), tableOp, NOT_METADATA);
Text startRow = ByteBufferUtil.toText(arguments.get(1));
Text endRow = ByteBufferUtil.toText(arguments.get(2));
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/master/src/main/java/org/apache/accumulo/master/Master.java
----------------------------------------------------------------------
diff --git a/server/master/src/main/java/org/apache/accumulo/master/Master.java b/server/master/src/main/java/org/apache/accumulo/master/Master.java
index e3fc69d..b6b96a0 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/Master.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/Master.java
@@ -62,6 +62,7 @@ import org.apache.accumulo.core.master.thrift.TabletServerStatus;
import org.apache.accumulo.core.metadata.MetadataTable;
import org.apache.accumulo.core.metadata.RootTable;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection;
+import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.thrift.ReplicationCoordinator;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.Credentials;
@@ -365,11 +366,16 @@ public class Master implements LiveTServerSet.Listener, TableObserver, CurrentSt
TableManager.prepareNewNamespaceState(instance.getInstanceID(), id, ns, NodeExistsPolicy.SKIP);
}
+ // create replication table in zk
+ log.debug("Upgrade creating table " + ReplicationTable.NAME + " (ID: " + ReplicationTable.ID + ")");
+ TableManager.prepareNewTableState(instance.getInstanceID(), ReplicationTable.ID, Namespaces.ACCUMULO_NAMESPACE_ID, ReplicationTable.NAME,
+ TableState.OFFLINE, NodeExistsPolicy.SKIP);
+
// create root table
log.debug("Upgrade creating table " + RootTable.NAME + " (ID: " + RootTable.ID + ")");
TableManager.prepareNewTableState(instance.getInstanceID(), RootTable.ID, Namespaces.ACCUMULO_NAMESPACE_ID, RootTable.NAME, TableState.ONLINE,
NodeExistsPolicy.SKIP);
- Initialize.initMetadataConfig();
+ Initialize.initSystemTablesConfig();
// ensure root user can flush root table
security.grantTablePermission(SystemCredentials.get().toThrift(instance), security.getRootUsername(), RootTable.ID, TablePermission.ALTER_TABLE,
Namespaces.ACCUMULO_NAMESPACE_ID);
@@ -379,15 +385,15 @@ public class Master implements LiveTServerSet.Listener, TableObserver, CurrentSt
for (String tableId : zoo.getChildren(tables)) {
String targetNamespace = (MetadataTable.ID.equals(tableId) || RootTable.ID.equals(tableId)) ? Namespaces.ACCUMULO_NAMESPACE_ID
: Namespaces.DEFAULT_NAMESPACE_ID;
- log.debug("Upgrade moving table " + new String(zoo.getData(tables + "/" + tableId + Constants.ZTABLE_NAME, null), UTF_8) + " (ID: "
- + tableId + ") into namespace with ID " + targetNamespace);
+ log.debug("Upgrade moving table " + new String(zoo.getData(tables + "/" + tableId + Constants.ZTABLE_NAME, null), UTF_8) + " (ID: " + tableId
+ + ") into namespace with ID " + targetNamespace);
zoo.putPersistentData(tables + "/" + tableId + Constants.ZTABLE_NAMESPACE, targetNamespace.getBytes(UTF_8), NodeExistsPolicy.SKIP);
}
// 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(UTF_8),
+ NodeExistsPolicy.OVERWRITE);
moveRootTabletToRootTable(zoo);
@@ -445,6 +451,11 @@ public class Master implements LiveTServerSet.Listener, TableObserver, CurrentSt
MetadataTableUtil.moveMetaDeleteMarkers(instance, SystemCredentials.get());
version++;
}
+ if (version == ServerConstants.MOVE_TO_REPLICATION_TABLE - 1) {
+ log.info("Updating metadata table with entries for the replication table");
+ MetadataTableUtil.createReplicationTable(instance, SystemCredentials.get());
+ version++;
+ }
log.info("Updating persistent data version.");
Accumulo.updateAccumuloVersion(fs, accumuloPersistentVersion);
log.info("Upgrade complete");
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/master/src/main/java/org/apache/accumulo/master/metrics/ReplicationMetrics.java
----------------------------------------------------------------------
diff --git a/server/master/src/main/java/org/apache/accumulo/master/metrics/ReplicationMetrics.java b/server/master/src/main/java/org/apache/accumulo/master/metrics/ReplicationMetrics.java
index d4a9af2..1554f52 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/metrics/ReplicationMetrics.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/metrics/ReplicationMetrics.java
@@ -25,7 +25,6 @@ import javax.management.ObjectName;
import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.Connector;
-import org.apache.accumulo.core.client.admin.TableOperations;
import org.apache.accumulo.core.client.impl.MasterClient;
import org.apache.accumulo.core.master.thrift.MasterClientService;
import org.apache.accumulo.core.master.thrift.MasterMonitorInfo;
@@ -47,21 +46,19 @@ public class ReplicationMetrics extends AbstractMetricsImpl implements Replicati
private static final String METRICS_PREFIX = "replication";
private Connector conn;
- private TableOperations tops;
private ObjectName objectName = null;
private ReplicationUtil replicationUtil;
public ReplicationMetrics(Connector conn) throws MalformedObjectNameException {
super();
this.conn = conn;
- this.tops = conn.tableOperations();
objectName = new ObjectName("accumulo.server.metrics:service=Replication Metrics,name=ReplicationMBean,instance=" + Thread.currentThread().getName());
replicationUtil = new ReplicationUtil();
}
@Override
public int getNumFilesPendingReplication() {
- if (!tops.exists(ReplicationTable.NAME)) {
+ if (!ReplicationTable.isOnline(conn)) {
return 0;
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/master/src/main/java/org/apache/accumulo/master/replication/DistributedWorkQueueWorkAssigner.java
----------------------------------------------------------------------
diff --git a/server/master/src/main/java/org/apache/accumulo/master/replication/DistributedWorkQueueWorkAssigner.java b/server/master/src/main/java/org/apache/accumulo/master/replication/DistributedWorkQueueWorkAssigner.java
index 8e1d036..3e966c4 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/replication/DistributedWorkQueueWorkAssigner.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/replication/DistributedWorkQueueWorkAssigner.java
@@ -22,7 +22,6 @@ import java.util.Set;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Scanner;
-import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.Key;
@@ -33,6 +32,7 @@ import org.apache.accumulo.core.replication.ReplicationConstants;
import org.apache.accumulo.core.replication.ReplicationSchema.OrderSection;
import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
import org.apache.accumulo.core.replication.ReplicationTable;
+import org.apache.accumulo.core.replication.ReplicationTableOfflineException;
import org.apache.accumulo.core.replication.ReplicationTarget;
import org.apache.accumulo.core.replication.StatusUtil;
import org.apache.accumulo.core.replication.proto.Replication.Status;
@@ -151,7 +151,7 @@ public abstract class DistributedWorkQueueWorkAssigner implements WorkAssigner {
Scanner s;
try {
s = ReplicationTable.getScanner(conn);
- } catch (TableNotFoundException e) {
+ } catch (ReplicationTableOfflineException e) {
return;
}
@@ -175,8 +175,8 @@ public abstract class DistributedWorkQueueWorkAssigner implements WorkAssigner {
Scanner workScanner;
try {
workScanner = ReplicationTable.getScanner(conn);
- } catch (TableNotFoundException e) {
- log.warn("Replication table was deleted. Will retry...");
+ } catch (ReplicationTableOfflineException e) {
+ log.warn("Replication table is offline. Will retry...");
UtilWaitThread.sleep(5000);
return;
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/master/src/main/java/org/apache/accumulo/master/replication/FinishedWorkUpdater.java
----------------------------------------------------------------------
diff --git a/server/master/src/main/java/org/apache/accumulo/master/replication/FinishedWorkUpdater.java b/server/master/src/main/java/org/apache/accumulo/master/replication/FinishedWorkUpdater.java
index 8bc18eb..51c8c1f 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/replication/FinishedWorkUpdater.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/replication/FinishedWorkUpdater.java
@@ -28,7 +28,6 @@ import org.apache.accumulo.core.client.BatchWriter;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.client.MutationsRejectedException;
-import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Range;
@@ -38,6 +37,7 @@ import org.apache.accumulo.core.protobuf.ProtobufUtil;
import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
import org.apache.accumulo.core.replication.ReplicationTable;
+import org.apache.accumulo.core.replication.ReplicationTableOfflineException;
import org.apache.accumulo.core.replication.ReplicationTarget;
import org.apache.accumulo.core.replication.StatusUtil;
import org.apache.accumulo.core.replication.proto.Replication.Status;
@@ -63,8 +63,8 @@ public class FinishedWorkUpdater implements Runnable {
public void run() {
log.debug("Looking for finished replication work");
- if (!conn.tableOperations().exists(ReplicationTable.NAME)) {
- log.debug("Replication table doesn't yet exist, will retry");
+ if (!ReplicationTable.isOnline(conn)) {
+ log.debug("Replication table is not yet online, will retry");
return;
}
@@ -73,8 +73,8 @@ public class FinishedWorkUpdater implements Runnable {
try {
bs = ReplicationTable.getBatchScanner(conn, 4);
replBw = ReplicationTable.getBatchWriter(conn);
- } catch (TableNotFoundException e) {
- log.debug("Table did exist, but was deleted, will retry");
+ } catch (ReplicationTableOfflineException e) {
+ log.debug("Table is no longer online, will retry");
return;
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/master/src/main/java/org/apache/accumulo/master/replication/RemoveCompleteReplicationRecords.java
----------------------------------------------------------------------
diff --git a/server/master/src/main/java/org/apache/accumulo/master/replication/RemoveCompleteReplicationRecords.java b/server/master/src/main/java/org/apache/accumulo/master/replication/RemoveCompleteReplicationRecords.java
index ef76916..7aaa465 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/replication/RemoveCompleteReplicationRecords.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/replication/RemoveCompleteReplicationRecords.java
@@ -30,7 +30,6 @@ import org.apache.accumulo.core.client.BatchWriter;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.client.MutationsRejectedException;
-import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Range;
@@ -40,6 +39,7 @@ import org.apache.accumulo.core.replication.ReplicationSchema.OrderSection;
import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
import org.apache.accumulo.core.replication.ReplicationTable;
+import org.apache.accumulo.core.replication.ReplicationTableOfflineException;
import org.apache.accumulo.core.replication.ReplicationTarget;
import org.apache.accumulo.core.replication.StatusUtil;
import org.apache.accumulo.core.replication.proto.Replication.Status;
@@ -51,7 +51,7 @@ import com.google.common.base.Stopwatch;
import com.google.protobuf.InvalidProtocolBufferException;
/**
- * Delete replication entries from {@link ReplicationTable#NAME} that are full replicated and closed
+ * Delete replication entries from the replication table that are fully replicated and closed
*/
public class RemoveCompleteReplicationRecords implements Runnable {
private static final Logger log = LoggerFactory.getLogger(RemoveCompleteReplicationRecords.class);
@@ -69,8 +69,8 @@ public class RemoveCompleteReplicationRecords implements Runnable {
try {
bs = ReplicationTable.getBatchScanner(conn, 4);
bw = ReplicationTable.getBatchWriter(conn);
- } catch (TableNotFoundException e) {
- log.debug("Not attempting to remove complete replication records as the table ({}) doesn't yet exist", ReplicationTable.NAME);
+ } catch (ReplicationTableOfflineException e) {
+ log.debug("Not attempting to remove complete replication records as the table ({}) isn't yet online", ReplicationTable.NAME);
return;
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/master/src/main/java/org/apache/accumulo/master/replication/StatusMaker.java
----------------------------------------------------------------------
diff --git a/server/master/src/main/java/org/apache/accumulo/master/replication/StatusMaker.java b/server/master/src/main/java/org/apache/accumulo/master/replication/StatusMaker.java
index c0f2a6c..3f831a8 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/replication/StatusMaker.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/replication/StatusMaker.java
@@ -18,6 +18,8 @@ package org.apache.accumulo.master.replication;
import java.util.Map.Entry;
+import org.apache.accumulo.core.client.AccumuloException;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.BatchWriter;
import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
@@ -34,9 +36,9 @@ import org.apache.accumulo.core.protobuf.ProtobufUtil;
import org.apache.accumulo.core.replication.ReplicationSchema.OrderSection;
import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
import org.apache.accumulo.core.replication.ReplicationTable;
+import org.apache.accumulo.core.replication.ReplicationTableOfflineException;
import org.apache.accumulo.core.replication.proto.Replication.Status;
import org.apache.accumulo.core.security.Authorizations;
-import org.apache.accumulo.server.replication.ReplicationUtil;
import org.apache.accumulo.trace.instrument.Span;
import org.apache.accumulo.trace.instrument.Trace;
import org.apache.hadoop.io.Text;
@@ -79,7 +81,7 @@ public class StatusMaker {
// Read from a source table (typically accumulo.metadata)
final Scanner s;
try {
- s = conn.createScanner(sourceTableName, new Authorizations());
+ s = conn.createScanner(sourceTableName, Authorizations.EMPTY);
} catch (TableNotFoundException e) {
throw new RuntimeException(e);
}
@@ -92,12 +94,12 @@ public class StatusMaker {
for (Entry<Key,Value> entry : s) {
// Get a writer to the replication table
if (null == replicationWriter) {
- // Ensures table exists and is properly configured
- ReplicationUtil.createReplicationTable(conn);
+ // Ensures table is online
try {
- setBatchWriter(ReplicationTable.getBatchWriter(conn));
- } catch (TableNotFoundException e) {
- log.warn("Replication table did exist, but does not anymore");
+ ReplicationTable.setOnline(conn);
+ replicationWriter = ReplicationTable.getBatchWriter(conn);
+ } catch (ReplicationTableOfflineException | AccumuloSecurityException | AccumuloException e) {
+ log.warn("Replication table did not come online");
replicationWriter = null;
return;
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/master/src/main/java/org/apache/accumulo/master/replication/WorkMaker.java
----------------------------------------------------------------------
diff --git a/server/master/src/main/java/org/apache/accumulo/master/replication/WorkMaker.java b/server/master/src/main/java/org/apache/accumulo/master/replication/WorkMaker.java
index 95f607f..f4fd676 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/replication/WorkMaker.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/replication/WorkMaker.java
@@ -25,7 +25,6 @@ import org.apache.accumulo.core.client.BatchWriter;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.MutationsRejectedException;
import org.apache.accumulo.core.client.Scanner;
-import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Mutation;
@@ -34,6 +33,7 @@ import org.apache.accumulo.core.replication.ReplicationSchema;
import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
import org.apache.accumulo.core.replication.ReplicationTable;
+import org.apache.accumulo.core.replication.ReplicationTableOfflineException;
import org.apache.accumulo.core.replication.ReplicationTarget;
import org.apache.accumulo.core.replication.StatusUtil;
import org.apache.accumulo.core.replication.proto.Replication.Status;
@@ -64,8 +64,8 @@ public class WorkMaker {
public void run() {
ServerConfigurationFactory serverConf = new ServerConfigurationFactory(conn.getInstance());
- if (!conn.tableOperations().exists(ReplicationTable.NAME)) {
- log.info("Replication table does not yet exist");
+ if (!ReplicationTable.isOnline(conn)) {
+ log.info("Replication table is not yet online");
return;
}
@@ -77,8 +77,8 @@ public class WorkMaker {
if (null == writer) {
setBatchWriter(ReplicationTable.getBatchWriter(conn));
}
- } catch (TableNotFoundException e) {
- log.warn("Replication table did exist, but does not anymore");
+ } catch (ReplicationTableOfflineException e) {
+ log.warn("Replication table was online, but not anymore");
writer = null;
return;
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/master/src/main/java/org/apache/accumulo/master/util/TableValidators.java
----------------------------------------------------------------------
diff --git a/server/master/src/main/java/org/apache/accumulo/master/util/TableValidators.java b/server/master/src/main/java/org/apache/accumulo/master/util/TableValidators.java
index 162b813..a770b47 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/util/TableValidators.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/util/TableValidators.java
@@ -19,14 +19,20 @@ package org.apache.accumulo.master.util;
import static org.apache.accumulo.core.client.impl.Tables.VALID_NAME_REGEX;
import static org.apache.accumulo.core.client.impl.Tables.qualify;
+import java.util.Arrays;
+import java.util.List;
+
import org.apache.accumulo.core.client.impl.Namespaces;
import org.apache.accumulo.core.metadata.MetadataTable;
import org.apache.accumulo.core.metadata.RootTable;
+import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.util.Validator;
+import com.google.common.base.Joiner;
+
public class TableValidators {
public static final String VALID_ID_REGEX = "^([a-z0-9]+)$"; // BigDecimal base36
-
+
public static final Validator<String> VALID_NAME = new Validator<String>() {
@Override
public boolean isValid(String tableName) {
@@ -44,7 +50,8 @@ public class TableValidators {
public static final Validator<String> VALID_ID = new Validator<String>() {
@Override
public boolean isValid(String tableId) {
- return tableId != null && (RootTable.ID.equals(tableId) || MetadataTable.ID.equals(tableId) || tableId.matches(VALID_ID_REGEX));
+ return tableId != null
+ && (RootTable.ID.equals(tableId) || MetadataTable.ID.equals(tableId) || ReplicationTable.ID.equals(tableId) || tableId.matches(VALID_ID_REGEX));
}
@Override
@@ -55,6 +62,21 @@ public class TableValidators {
}
};
+ public static final Validator<String> NOT_METADATA = new Validator<String>() {
+
+ private List<String> metadataTables = Arrays.asList(RootTable.NAME, MetadataTable.NAME);
+
+ @Override
+ public boolean isValid(String tableName) {
+ return !metadataTables.contains(tableName);
+ }
+
+ @Override
+ public String invalidMessage(String tableName) {
+ return "Table cannot be one of {" + Joiner.on(",").join(metadataTables) + "}";
+ }
+ };
+
public static final Validator<String> NOT_SYSTEM = new Validator<String>() {
@Override
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/master/src/test/java/org/apache/accumulo/master/replication/FinishedWorkUpdaterTest.java
----------------------------------------------------------------------
diff --git a/server/master/src/test/java/org/apache/accumulo/master/replication/FinishedWorkUpdaterTest.java b/server/master/src/test/java/org/apache/accumulo/master/replication/FinishedWorkUpdaterTest.java
index f8cc775..7e8f762 100644
--- a/server/master/src/test/java/org/apache/accumulo/master/replication/FinishedWorkUpdaterTest.java
+++ b/server/master/src/test/java/org/apache/accumulo/master/replication/FinishedWorkUpdaterTest.java
@@ -34,7 +34,6 @@ import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.ReplicationTarget;
import org.apache.accumulo.core.replication.proto.Replication.Status;
-import org.apache.accumulo.server.replication.ReplicationUtil;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Rule;
@@ -68,8 +67,6 @@ public class FinishedWorkUpdaterTest {
@Test
public void recordsWithProgressUpdateBothTables() throws Exception {
- ReplicationUtil.createReplicationTable(conn);
-
String file = "/accumulo/wals/tserver+port/" + UUID.randomUUID();
Status stat = Status.newBuilder().setBegin(100).setEnd(200).setClosed(true).setInfiniteEnd(false).build();
ReplicationTarget target = new ReplicationTarget("peer", "table1", "1");
@@ -98,8 +95,6 @@ public class FinishedWorkUpdaterTest {
@Test
public void chooseMinimumBeginOffset() throws Exception {
- ReplicationUtil.createReplicationTable(conn);
-
String file = "/accumulo/wals/tserver+port/" + UUID.randomUUID();
// @formatter:off
Status stat1 = Status.newBuilder().setBegin(100).setEnd(1000).setClosed(true).setInfiniteEnd(false).build(),
@@ -136,8 +131,6 @@ public class FinishedWorkUpdaterTest {
@Test
public void chooseMinimumBeginOffsetInfiniteEnd() throws Exception {
- ReplicationUtil.createReplicationTable(conn);
-
String file = "/accumulo/wals/tserver+port/" + UUID.randomUUID();
// @formatter:off
Status stat1 = Status.newBuilder().setBegin(100).setEnd(1000).setClosed(true).setInfiniteEnd(true).build(),
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/master/src/test/java/org/apache/accumulo/master/replication/RemoveCompleteReplicationRecordsTest.java
----------------------------------------------------------------------
diff --git a/server/master/src/test/java/org/apache/accumulo/master/replication/RemoveCompleteReplicationRecordsTest.java b/server/master/src/test/java/org/apache/accumulo/master/replication/RemoveCompleteReplicationRecordsTest.java
index 2255e21..4bd4b36 100644
--- a/server/master/src/test/java/org/apache/accumulo/master/replication/RemoveCompleteReplicationRecordsTest.java
+++ b/server/master/src/test/java/org/apache/accumulo/master/replication/RemoveCompleteReplicationRecordsTest.java
@@ -41,7 +41,6 @@ import org.apache.accumulo.core.replication.ReplicationTable;
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.server.replication.ReplicationUtil;
import org.apache.hadoop.io.Text;
import org.easymock.EasyMock;
import org.junit.Assert;
@@ -53,7 +52,7 @@ import org.junit.rules.TestName;
import com.google.common.collect.Iterables;
/**
- *
+ *
*/
public class RemoveCompleteReplicationRecordsTest {
@@ -73,7 +72,6 @@ public class RemoveCompleteReplicationRecordsTest {
@Test
public void notYetReplicationRecordsIgnored() throws Exception {
- ReplicationUtil.createReplicationTable(conn);
BatchWriter bw = ReplicationTable.getBatchWriter(conn);
int numRecords = 3;
for (int i = 0; i < numRecords; i++) {
@@ -103,7 +101,6 @@ public class RemoveCompleteReplicationRecordsTest {
@Test
public void partiallyReplicatedRecordsIgnored() throws Exception {
- ReplicationUtil.createReplicationTable(conn);
BatchWriter bw = ReplicationTable.getBatchWriter(conn);
int numRecords = 3;
Status.Builder builder = Status.newBuilder();
@@ -138,7 +135,6 @@ public class RemoveCompleteReplicationRecordsTest {
@Test
public void replicatedClosedWorkRecordsAreNotRemovedWithoutClosedStatusRecords() throws Exception {
- ReplicationUtil.createReplicationTable(conn);
BatchWriter replBw = ReplicationTable.getBatchWriter(conn);
int numRecords = 3;
@@ -191,7 +187,6 @@ public class RemoveCompleteReplicationRecordsTest {
@Test
public void replicatedClosedRowsAreRemoved() throws Exception {
- ReplicationUtil.createReplicationTable(conn);
BatchWriter replBw = ReplicationTable.getBatchWriter(conn);
int numRecords = 3;
@@ -283,7 +278,6 @@ public class RemoveCompleteReplicationRecordsTest {
@Test
public void partiallyReplicatedEntriesPrecludeRowDeletion() throws Exception {
- ReplicationUtil.createReplicationTable(conn);
BatchWriter replBw = ReplicationTable.getBatchWriter(conn);
int numRecords = 3;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/master/src/test/java/org/apache/accumulo/master/replication/SequentialWorkAssignerTest.java
----------------------------------------------------------------------
diff --git a/server/master/src/test/java/org/apache/accumulo/master/replication/SequentialWorkAssignerTest.java b/server/master/src/test/java/org/apache/accumulo/master/replication/SequentialWorkAssignerTest.java
index d5bcfbd..0f1a7a1 100644
--- a/server/master/src/test/java/org/apache/accumulo/master/replication/SequentialWorkAssignerTest.java
+++ b/server/master/src/test/java/org/apache/accumulo/master/replication/SequentialWorkAssignerTest.java
@@ -44,7 +44,6 @@ import org.apache.accumulo.core.security.Credentials;
import org.apache.accumulo.core.security.TablePermission;
import org.apache.accumulo.core.zookeeper.ZooUtil;
import org.apache.accumulo.server.replication.DistributedWorkQueueWorkAssignerHelper;
-import org.apache.accumulo.server.replication.ReplicationUtil;
import org.apache.accumulo.server.zookeeper.DistributedWorkQueue;
import org.apache.accumulo.server.zookeeper.ZooCache;
import org.apache.hadoop.io.Text;
@@ -55,7 +54,7 @@ import org.junit.Test;
import org.junit.rules.TestName;
/**
- *
+ *
*/
public class SequentialWorkAssignerTest {
@@ -85,8 +84,7 @@ public class SequentialWorkAssignerTest {
// Set the connector
assigner.setConnector(conn);
- // Create and grant ourselves write to the replication table
- ReplicationUtil.createReplicationTable(conn);
+ // grant ourselves write to the replication table
conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.WRITE);
// Create two mutations, both of which need replication work done
@@ -158,8 +156,7 @@ public class SequentialWorkAssignerTest {
// Set the connector
assigner.setConnector(conn);
- // Create and grant ourselves write to the replication table
- ReplicationUtil.createReplicationTable(conn);
+ // grant ourselves write to the replication table
conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.WRITE);
// Create two mutations, both of which need replication work done
@@ -238,8 +235,7 @@ public class SequentialWorkAssignerTest {
// Set the connector
assigner.setConnector(conn);
- // Create and grant ourselves write to the replication table
- ReplicationUtil.createReplicationTable(conn);
+ // grant ourselves write to the replication table
conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.WRITE);
// Create two mutations, both of which need replication work done
@@ -359,8 +355,7 @@ public class SequentialWorkAssignerTest {
// Set the connector
assigner.setConnector(conn);
- // Create and grant ourselves write to the replication table
- ReplicationUtil.createReplicationTable(conn);
+ // grant ourselves write to the replication table
conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.WRITE);
// Create two mutations, both of which need replication work done
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/master/src/test/java/org/apache/accumulo/master/replication/UnorderedWorkAssignerTest.java
----------------------------------------------------------------------
diff --git a/server/master/src/test/java/org/apache/accumulo/master/replication/UnorderedWorkAssignerTest.java b/server/master/src/test/java/org/apache/accumulo/master/replication/UnorderedWorkAssignerTest.java
index 5d69216..4282e4e 100644
--- a/server/master/src/test/java/org/apache/accumulo/master/replication/UnorderedWorkAssignerTest.java
+++ b/server/master/src/test/java/org/apache/accumulo/master/replication/UnorderedWorkAssignerTest.java
@@ -48,7 +48,6 @@ import org.apache.accumulo.core.replication.proto.Replication.Status;
import org.apache.accumulo.core.security.Credentials;
import org.apache.accumulo.core.security.TablePermission;
import org.apache.accumulo.server.replication.DistributedWorkQueueWorkAssignerHelper;
-import org.apache.accumulo.server.replication.ReplicationUtil;
import org.apache.accumulo.server.zookeeper.DistributedWorkQueue;
import org.apache.accumulo.server.zookeeper.ZooCache;
import org.apache.hadoop.fs.Path;
@@ -136,8 +135,7 @@ public class UnorderedWorkAssignerTest {
// Set the connector
assigner.setConnector(conn);
- // Create and grant ourselves write to the replication table
- ReplicationUtil.createReplicationTable(conn);
+ // grant ourselves write to the replication table
conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.WRITE);
Status.Builder builder = Status.newBuilder().setBegin(0).setEnd(0).setInfiniteEnd(true).setClosed(false).setCreatedTime(5l);
@@ -199,8 +197,7 @@ public class UnorderedWorkAssignerTest {
// Set the connector
assigner.setConnector(conn);
- // Create and grant ourselves write to the replication table
- ReplicationUtil.createReplicationTable(conn);
+ // grant ourselves write to the replication table
conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.WRITE);
// Create two mutations, both of which need replication work done
@@ -271,8 +268,7 @@ public class UnorderedWorkAssignerTest {
// Set the connector
assigner.setConnector(conn);
- // Create and grant ourselves write to the replication table
- ReplicationUtil.createReplicationTable(conn);
+ // grant ourselves write to the replication table
conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.WRITE);
// Create two mutations, both of which need replication work done
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/master/src/test/java/org/apache/accumulo/master/replication/WorkMakerTest.java
----------------------------------------------------------------------
diff --git a/server/master/src/test/java/org/apache/accumulo/master/replication/WorkMakerTest.java b/server/master/src/test/java/org/apache/accumulo/master/replication/WorkMakerTest.java
index 280ca82..0455c44 100644
--- a/server/master/src/test/java/org/apache/accumulo/master/replication/WorkMakerTest.java
+++ b/server/master/src/test/java/org/apache/accumulo/master/replication/WorkMakerTest.java
@@ -16,7 +16,6 @@
*/
package org.apache.accumulo.master.replication;
-import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Map.Entry;
@@ -37,7 +36,7 @@ import org.apache.accumulo.core.replication.ReplicationTable;
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.server.conf.TableConfiguration;
+import org.apache.accumulo.core.security.TablePermission;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.junit.Assert;
@@ -54,8 +53,6 @@ import com.google.common.collect.Iterables;
*/
public class WorkMakerTest {
- private final Map<String,TableConfiguration> tableConfs = new HashMap<>();
-
private MockInstance instance;
private Connector conn;
@@ -66,14 +63,8 @@ public class WorkMakerTest {
public void createMockAccumulo() throws Exception {
instance = new MockInstance();
conn = instance.getConnector("root", new PasswordToken(""));
-
- if (conn.tableOperations().exists(ReplicationTable.NAME)) {
- conn.tableOperations().delete(ReplicationTable.NAME);
- }
-
- conn.tableOperations().create(ReplicationTable.NAME);
-
- tableConfs.clear();
+ conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.WRITE);
+ conn.tableOperations().deleteRows(ReplicationTable.NAME, null, null);
}
@Test
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/ReplicationServlet.java
----------------------------------------------------------------------
diff --git a/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/ReplicationServlet.java b/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/ReplicationServlet.java
index 4c21c39..7d3cb14 100644
--- a/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/ReplicationServlet.java
+++ b/server/monitor/src/main/java/org/apache/accumulo/monitor/servlets/ReplicationServlet.java
@@ -28,7 +28,6 @@ import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.admin.TableOperations;
import org.apache.accumulo.core.master.thrift.MasterMonitorInfo;
import org.apache.accumulo.core.replication.ReplicationConstants;
-import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.ReplicationTarget;
import org.apache.accumulo.core.security.Credentials;
import org.apache.accumulo.core.zookeeper.ZooUtil;
@@ -76,10 +75,6 @@ public class ReplicationServlet extends BasicServlet {
int totalWorkQueueSize = replicationUtil.getMaxReplicationThreads(systemProps, mmi);
TableOperations tops = conn.tableOperations();
- if (!tops.exists(ReplicationTable.NAME)) {
- banner(sb, "", "Replication table does not yet exist");
- return;
- }
Table replicationStats = new Table("replicationStats", "Replication Status");
replicationStats.addSortableColumn("Table");
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/server/tserver/src/main/java/org/apache/accumulo/tserver/replication/ReplicationProcessor.java
----------------------------------------------------------------------
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/replication/ReplicationProcessor.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/replication/ReplicationProcessor.java
index 1843acb..c23cd94 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
@@ -26,7 +26,6 @@ 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;
@@ -36,6 +35,7 @@ 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.ReplicationTable;
+import org.apache.accumulo.core.replication.ReplicationTableOfflineException;
import org.apache.accumulo.core.replication.ReplicationTarget;
import org.apache.accumulo.core.replication.StatusUtil;
import org.apache.accumulo.core.replication.proto.Replication.Status;
@@ -101,7 +101,7 @@ public class ReplicationProcessor implements Processor {
Status status;
try {
status = getStatus(file, target);
- } catch (TableNotFoundException | AccumuloException | AccumuloSecurityException e) {
+ } catch (ReplicationTableOfflineException | 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) {
@@ -171,7 +171,7 @@ public class ReplicationProcessor implements Processor {
return true;
}
- protected Status getStatus(String file, ReplicationTarget target) throws TableNotFoundException, AccumuloException, AccumuloSecurityException,
+ protected Status getStatus(String file, ReplicationTarget target) throws ReplicationTableOfflineException, AccumuloException, AccumuloSecurityException,
InvalidProtocolBufferException {
Scanner s = ReplicationTable.getScanner(inst.getConnector(creds.getPrincipal(), creds.getToken()));
s.setRange(Range.exact(file));
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/test/src/test/java/org/apache/accumulo/test/LargeSplitRowIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/LargeSplitRowIT.java b/test/src/test/java/org/apache/accumulo/test/LargeSplitRowIT.java
index 7ebcdd7..8ceabcb 100644
--- a/test/src/test/java/org/apache/accumulo/test/LargeSplitRowIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/LargeSplitRowIT.java
@@ -19,8 +19,8 @@ package org.apache.accumulo.test;
import static java.nio.charset.StandardCharsets.UTF_8;
import java.util.HashMap;
-import java.util.Map.Entry;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.SortedSet;
import java.util.TreeSet;
@@ -29,6 +29,7 @@ import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.impl.AccumuloServerException;
+import org.apache.accumulo.core.client.impl.Namespaces;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Mutation;
@@ -185,7 +186,7 @@ public class LargeSplitRowIT extends ConfigurableMacIT {
while (iterator.hasNext()) {
String curr = iterator.next();
- if (!curr.equals("accumulo.metadata") && !curr.equals("accumulo.root")) {
+ if (!curr.startsWith(Namespaces.ACCUMULO_NAMESPACE + ".")) {
tableName = curr;
}
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/test/src/test/java/org/apache/accumulo/test/MasterRepairsDualAssignmentIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/MasterRepairsDualAssignmentIT.java b/test/src/test/java/org/apache/accumulo/test/MasterRepairsDualAssignmentIT.java
index c69de2d..da2d20f 100644
--- a/test/src/test/java/org/apache/accumulo/test/MasterRepairsDualAssignmentIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/MasterRepairsDualAssignmentIT.java
@@ -33,6 +33,7 @@ import org.apache.accumulo.core.data.KeyExtent;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.metadata.MetadataTable;
import org.apache.accumulo.core.metadata.RootTable;
+import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.security.Credentials;
import org.apache.accumulo.core.security.TablePermission;
import org.apache.accumulo.fate.util.UtilWaitThread;
@@ -50,7 +51,7 @@ import org.apache.hadoop.io.Text;
import org.junit.Test;
public class MasterRepairsDualAssignmentIT extends ConfigurableMacIT {
-
+
@Override
public int defaultTimeoutSeconds() {
return 5 * 60;
@@ -95,14 +96,18 @@ public class MasterRepairsDualAssignmentIT extends ConfigurableMacIT {
assertEquals(2, states.size());
// Kill a tablet server... we don't care which one... wait for everything to be reassigned
cluster.killProcess(ServerType.TABLET_SERVER, cluster.getProcesses().get(ServerType.TABLET_SERVER).iterator().next());
+ Set<TServerInstance> replStates = new HashSet<>();
// Find out which tablet server remains
while (true) {
UtilWaitThread.sleep(1000);
states.clear();
+ replStates.clear();
boolean allAssigned = true;
for (TabletLocationState tls : store) {
if (tls != null && tls.current != null) {
states.add(tls.current);
+ } else if (tls.extent.equals(new KeyExtent(new Text(ReplicationTable.ID), null, null))) {
+ replStates.add(tls.current);
} else {
allAssigned = false;
}
@@ -111,6 +116,7 @@ public class MasterRepairsDualAssignmentIT extends ConfigurableMacIT {
if (states.size() != 2 && allAssigned == true)
break;
}
+ assertEquals(1, replStates.size());
assertEquals(1, states.size());
// pick an assigned tablet and assign it to the old tablet
TabletLocationState moved = null;
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/test/src/test/java/org/apache/accumulo/test/MetaConstraintRetryIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/MetaConstraintRetryIT.java b/test/src/test/java/org/apache/accumulo/test/MetaConstraintRetryIT.java
index b3c3640..5dc75d0 100644
--- a/test/src/test/java/org/apache/accumulo/test/MetaConstraintRetryIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/MetaConstraintRetryIT.java
@@ -31,13 +31,13 @@ import org.apache.hadoop.io.Text;
import org.junit.Test;
public class MetaConstraintRetryIT extends SimpleMacIT {
-
+
@Override
public int defaultTimeoutSeconds() {
return 30;
}
- //a test for ACCUMULO-3096
+ // a test for ACCUMULO-3096
@Test(expected = ConstraintViolationException.class)
public void test() throws Exception {
@@ -47,13 +47,12 @@ public class MetaConstraintRetryIT extends SimpleMacIT {
Writer w = new Writer(super.getConnector().getInstance(), credentials, MetadataTable.ID);
KeyExtent extent = new KeyExtent(new Text("5"), null, null);
-
Mutation m = new Mutation(extent.getMetadataEntry());
// unknown columns should cause contraint violation
m.put("badcolfam", "badcolqual", "3");
try {
- MetadataTableUtil.update(w, credentials, null, m);
+ MetadataTableUtil.update(w, null, m);
} catch (RuntimeException e) {
if (e.getCause().getClass().equals(ConstraintViolationException.class)) {
throw (ConstraintViolationException) e.getCause();
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/test/src/test/java/org/apache/accumulo/test/NamespacesIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/NamespacesIT.java b/test/src/test/java/org/apache/accumulo/test/NamespacesIT.java
index 5e885b5..0527a9a 100644
--- a/test/src/test/java/org/apache/accumulo/test/NamespacesIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/NamespacesIT.java
@@ -102,7 +102,7 @@ public class NamespacesIT extends SimpleMacIT {
for (String t : c.tableOperations().list())
if (!Tables.qualify(t).getFirst().equals(Namespaces.ACCUMULO_NAMESPACE))
c.tableOperations().delete(t);
- assertEquals(2, c.tableOperations().list().size());
+ assertEquals(3, c.tableOperations().list().size());
for (String n : c.namespaceOperations().list())
if (!n.equals(Namespaces.ACCUMULO_NAMESPACE) && !n.equals(Namespaces.DEFAULT_NAMESPACE))
c.namespaceOperations().delete(n);
@@ -318,7 +318,7 @@ public class NamespacesIT extends SimpleMacIT {
// verify entry is filtered out (also, verify conflict checking API)
c.namespaceOperations().checkIteratorConflicts(namespace, setting, EnumSet.allOf(IteratorScope.class));
c.namespaceOperations().attachIterator(namespace, setting);
- UtilWaitThread.sleep(2*1000);
+ UtilWaitThread.sleep(2 * 1000);
try {
c.namespaceOperations().checkIteratorConflicts(namespace, setting, EnumSet.allOf(IteratorScope.class));
fail();
@@ -334,7 +334,7 @@ public class NamespacesIT extends SimpleMacIT {
// verify can see inserted entry again
c.namespaceOperations().removeIterator(namespace, setting.getName(), EnumSet.allOf(IteratorScope.class));
- UtilWaitThread.sleep(2*1000);
+ UtilWaitThread.sleep(2 * 1000);
assertFalse(c.namespaceOperations().listIterators(namespace).containsKey(iterName));
assertFalse(c.tableOperations().listIterators(t1).containsKey(iterName));
s = c.createScanner(t1, Authorizations.EMPTY);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/test/src/test/java/org/apache/accumulo/test/ShellServerIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/ShellServerIT.java b/test/src/test/java/org/apache/accumulo/test/ShellServerIT.java
index 58042b2..4114dc8 100644
--- a/test/src/test/java/org/apache/accumulo/test/ShellServerIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/ShellServerIT.java
@@ -53,7 +53,6 @@ import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.file.FileOperations;
import org.apache.accumulo.core.file.FileSKVWriter;
import org.apache.accumulo.core.metadata.MetadataTable;
-import org.apache.accumulo.core.metadata.RootTable;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.accumulo.shell.Shell;
@@ -121,6 +120,7 @@ public class ShellServerIT extends SimpleMacIT {
private static class NoOpErrorMessageCallback extends ErrorMessageCallback {
private static final String empty = "";
+
@Override
public String getErrorMessage() {
return empty;
@@ -239,8 +239,8 @@ public class ShellServerIT extends SimpleMacIT {
@Before
public void setupShell() throws Exception {
- ts = new TestShell(ROOT_PASSWORD, getStaticCluster().getConfig().getInstanceName(), getStaticCluster().getConfig().getZooKeepers(),
- getStaticCluster().getConfig().getClientConfFile().getAbsolutePath());
+ ts = new TestShell(ROOT_PASSWORD, getStaticCluster().getConfig().getInstanceName(), getStaticCluster().getConfig().getZooKeepers(), getStaticCluster()
+ .getConfig().getClientConfFile().getAbsolutePath());
}
@AfterClass
@@ -252,7 +252,7 @@ public class ShellServerIT extends SimpleMacIT {
public void deleteTables() throws Exception {
Connector c = getConnector();
for (String table : c.tableOperations().list()) {
- if (!table.equals(MetadataTable.NAME) && !table.equals(RootTable.NAME) && !table.equals("trace"))
+ if (!table.startsWith(Namespaces.ACCUMULO_NAMESPACE + ".") && !table.equals("trace"))
try {
c.tableOperations().delete(table);
} catch (TableNotFoundException e) {
@@ -427,7 +427,7 @@ public class ShellServerIT extends SimpleMacIT {
ts.exec("deleteuser -f xyzzy", true);
ts.exec("users", true, "xyzzy", false);
}
-
+
@Test(timeout = 60 * 1000)
public void durability() throws Exception {
final String table = name.getMethodName();
@@ -436,7 +436,7 @@ public class ShellServerIT extends SimpleMacIT {
ts.exec("insert -d foo a cf cq2 2", false, "foo", true);
ts.exec("scan -r a", true, "randomGunkaASDFWEAQRd", true);
ts.exec("scan -r a", true, "foo", false);
- }
+ }
@Test
public void iter() throws Exception {
@@ -1105,9 +1105,8 @@ public class ShellServerIT extends SimpleMacIT {
FileUtils.copyURLToFile(this.getClass().getResource("/FooConstraint.jar"), fooConstraintJar);
fooConstraintJar.deleteOnExit();
- ts.exec(
- "config -s " + Property.VFS_CONTEXT_CLASSPATH_PROPERTY.getKey() + "cx1=" + fooFilterJar.toURI().toString() + "," + fooConstraintJar.toURI().toString(),
- true);
+ ts.exec("config -s " + Property.VFS_CONTEXT_CLASSPATH_PROPERTY.getKey() + "cx1=" + fooFilterJar.toURI().toString() + ","
+ + fooConstraintJar.toURI().toString(), true);
ts.exec("createtable " + table, true);
ts.exec("config -t " + table + " -s " + Property.TABLE_CLASSPATH.getKey() + "=cx1", true);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/test/src/test/java/org/apache/accumulo/test/WaitForBalanceIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/WaitForBalanceIT.java b/test/src/test/java/org/apache/accumulo/test/WaitForBalanceIT.java
index a331faa..b558715 100644
--- a/test/src/test/java/org/apache/accumulo/test/WaitForBalanceIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/WaitForBalanceIT.java
@@ -38,12 +38,13 @@ import org.apache.hadoop.io.Text;
import org.junit.Test;
public class WaitForBalanceIT extends ConfigurableMacIT {
-
+
@Test(timeout = 30 * 1000)
public void test() throws Exception {
final Connector c = getConnector();
// ensure the metadata table is online
- for (@SuppressWarnings("unused") Entry<Key,Value> unused : c.createScanner(MetadataTable.NAME, Authorizations.EMPTY))
+ for (@SuppressWarnings("unused")
+ Entry<Key,Value> unused : c.createScanner(MetadataTable.NAME, Authorizations.EMPTY))
;
c.instanceOperations().waitForBalance();
assertTrue(isBalanced());
@@ -61,10 +62,10 @@ public class WaitForBalanceIT extends ConfigurableMacIT {
}
private boolean isBalanced() throws Exception {
- final Map<String, Integer> counts = new HashMap<String, Integer>();
+ final Map<String,Integer> counts = new HashMap<String,Integer>();
int offline = 0;
final Connector c = getConnector();
- for (String tableName : new String[]{MetadataTable.NAME, RootTable.NAME}) {
+ for (String tableName : new String[] {MetadataTable.NAME, RootTable.NAME}) {
final Scanner s = c.createScanner(tableName, Authorizations.EMPTY);
s.setRange(MetadataSchema.TabletsSection.getRange());
s.fetchColumnFamily(MetadataSchema.TabletsSection.CurrentLocationColumnFamily.NAME);
@@ -88,7 +89,8 @@ public class WaitForBalanceIT extends ConfigurableMacIT {
}
}
}
- if (offline > 0) {
+ // the replication table is expected to be offline for this test, so ignore it
+ if (offline > 1) {
System.out.println("Offline tablets " + offline);
return false;
}
@@ -107,5 +109,5 @@ public class WaitForBalanceIT extends ConfigurableMacIT {
}
return true;
}
-
+
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/test/src/test/java/org/apache/accumulo/test/replication/MultiInstanceReplicationIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/replication/MultiInstanceReplicationIT.java b/test/src/test/java/org/apache/accumulo/test/replication/MultiInstanceReplicationIT.java
index 0000e7f..6ce5959 100644
--- a/test/src/test/java/org/apache/accumulo/test/replication/MultiInstanceReplicationIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/replication/MultiInstanceReplicationIT.java
@@ -54,7 +54,6 @@ import org.apache.accumulo.minicluster.ServerType;
import org.apache.accumulo.minicluster.impl.MiniAccumuloClusterImpl;
import org.apache.accumulo.minicluster.impl.MiniAccumuloConfigImpl;
import org.apache.accumulo.minicluster.impl.ProcessReference;
-import org.apache.accumulo.server.replication.ReplicationUtil;
import org.apache.accumulo.test.functional.ConfigurableMacIT;
import org.apache.accumulo.tserver.TabletServer;
import org.apache.accumulo.tserver.replication.AccumuloReplicaSystem;
@@ -164,8 +163,8 @@ public class MultiInstanceReplicationIT extends ConfigurableMacIT {
final Connector connMaster = getConnector();
final Connector connPeer = peerCluster.getConnector("root", ROOT_PASSWORD);
- ReplicationUtil.createReplicationTable(connMaster);
-
+ ReplicationTable.setOnline(connMaster);
+
String peerUserName = "peer", peerPassword = "foo";
String peerClusterName = "peer";
@@ -236,7 +235,7 @@ public class MultiInstanceReplicationIT extends ConfigurableMacIT {
log.info("");
log.info("Fetching replication records:");
- for (Entry<Key,Value> kv : connMaster.createScanner(ReplicationTable.NAME, Authorizations.EMPTY)) {
+ for (Entry<Key,Value> kv : ReplicationTable.getScanner(connMaster)) {
log.info(kv.getKey().toStringNoTruncate() + " " + ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
}
@@ -274,7 +273,7 @@ public class MultiInstanceReplicationIT extends ConfigurableMacIT {
log.info("");
log.info("Fetching replication records:");
- for (Entry<Key,Value> kv : connMaster.createScanner(ReplicationTable.NAME, Authorizations.EMPTY)) {
+ for (Entry<Key,Value> kv : ReplicationTable.getScanner(connMaster)) {
log.info(kv.getKey().toStringNoTruncate() + " " + ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
}
@@ -397,10 +396,6 @@ public class MultiInstanceReplicationIT extends ConfigurableMacIT {
Set<String> filesFor1 = connMaster.replicationOperations().referencedFiles(masterTable1), filesFor2 = connMaster.replicationOperations().referencedFiles(
masterTable2);
- while (!connMaster.tableOperations().exists(ReplicationTable.NAME)) {
- Thread.sleep(500);
- }
-
// Restart the tserver to force a close on the WAL
for (ProcessReference proc : cluster.getProcesses().get(ServerType.TABLET_SERVER)) {
cluster.killProcess(ServerType.TABLET_SERVER, proc);
@@ -523,7 +518,7 @@ public class MultiInstanceReplicationIT extends ConfigurableMacIT {
for (@SuppressWarnings("unused")
Entry<Key,Value> kv : connMaster.createScanner(masterTable, Authorizations.EMPTY)) {}
- for (Entry<Key,Value> kv : connMaster.createScanner(ReplicationTable.NAME, Authorizations.EMPTY)) {
+ for (Entry<Key,Value> kv : ReplicationTable.getScanner(connMaster)) {
log.debug(kv.getKey().toStringNoTruncate() + " " + ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
}
@@ -637,10 +632,6 @@ public class MultiInstanceReplicationIT extends ConfigurableMacIT {
log.info("Wrote all data to master cluster");
- while (!connMaster.tableOperations().exists(ReplicationTable.NAME)) {
- Thread.sleep(500);
- }
-
for (ProcessReference proc : cluster.getProcesses().get(ServerType.TABLET_SERVER)) {
cluster.killProcess(ServerType.TABLET_SERVER, proc);
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/test/src/test/java/org/apache/accumulo/test/replication/ReplicationIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/replication/ReplicationIT.java b/test/src/test/java/org/apache/accumulo/test/replication/ReplicationIT.java
index 69cd195..c514cf8 100644
--- a/test/src/test/java/org/apache/accumulo/test/replication/ReplicationIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/replication/ReplicationIT.java
@@ -21,26 +21,34 @@ import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
+import java.util.EnumSet;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
+import java.util.Map;
import java.util.Map.Entry;
import java.util.NoSuchElementException;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.BatchWriter;
import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.IteratorSetting;
+import org.apache.accumulo.core.client.IteratorSetting.Column;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.client.admin.TableOperations;
import org.apache.accumulo.core.client.replication.ReplicaSystemFactory;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
+import org.apache.accumulo.core.iterators.conf.ColumnSet;
import org.apache.accumulo.core.metadata.MetadataTable;
import org.apache.accumulo.core.metadata.schema.MetadataSchema;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.ReplicationSection;
@@ -51,6 +59,7 @@ import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
import org.apache.accumulo.core.replication.ReplicationSchema.WorkSection;
import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.ReplicationTarget;
+import org.apache.accumulo.core.replication.StatusFormatter;
import org.apache.accumulo.core.replication.StatusUtil;
import org.apache.accumulo.core.replication.proto.Replication.Status;
import org.apache.accumulo.core.security.Authorizations;
@@ -61,7 +70,7 @@ import org.apache.accumulo.gc.SimpleGarbageCollector;
import org.apache.accumulo.minicluster.ServerType;
import org.apache.accumulo.minicluster.impl.MiniAccumuloConfigImpl;
import org.apache.accumulo.minicluster.impl.ProcessReference;
-import org.apache.accumulo.server.replication.ReplicationUtil;
+import org.apache.accumulo.server.replication.StatusCombiner;
import org.apache.accumulo.server.util.ReplicationTableUtil;
import org.apache.accumulo.test.functional.ConfigurableMacIT;
import org.apache.accumulo.tserver.TabletServer;
@@ -75,6 +84,8 @@ import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.base.Function;
+import com.google.common.base.Joiner;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.Iterables;
import com.google.common.collect.Multimap;
@@ -130,6 +141,75 @@ public class ReplicationIT extends ConfigurableMacIT {
}
@Test
+ public void replicationTableCreated() throws AccumuloException, AccumuloSecurityException {
+ Assert.assertTrue(getConnector().tableOperations().exists(ReplicationTable.NAME));
+ Assert.assertEquals(ReplicationTable.ID, getConnector().tableOperations().tableIdMap().get(ReplicationTable.NAME));
+ }
+
+ @Test
+ public void verifyReplicationTableConfig() throws AccumuloException, TableNotFoundException, AccumuloSecurityException {
+ TableOperations tops = getConnector().tableOperations();
+ Map<String,EnumSet<IteratorScope>> iterators = tops.listIterators(ReplicationTable.NAME);
+
+ // verify combiners are only iterators (no versioning)
+ Assert.assertEquals(1, iterators.size());
+
+ // look for combiner
+ Assert.assertTrue(iterators.containsKey(ReplicationTable.COMBINER_NAME));
+ Assert.assertTrue(iterators.get(ReplicationTable.COMBINER_NAME).containsAll(EnumSet.allOf(IteratorScope.class)));
+ for (IteratorScope scope : EnumSet.allOf(IteratorScope.class)) {
+ IteratorSetting is = tops.getIteratorSetting(ReplicationTable.NAME, ReplicationTable.COMBINER_NAME, scope);
+ Assert.assertEquals(30, is.getPriority());
+ Assert.assertEquals(StatusCombiner.class.getName(), is.getIteratorClass());
+ Assert.assertEquals(1, is.getOptions().size());
+ Assert.assertTrue(is.getOptions().containsKey("columns"));
+ String cols = is.getOptions().get("columns");
+ Column statusSectionCol = new Column(StatusSection.NAME);
+ Column workSectionCol = new Column(WorkSection.NAME);
+ Assert.assertEquals(
+ ColumnSet.encodeColumns(statusSectionCol.getColumnFamily(), statusSectionCol.getColumnQualifier()) + ","
+ + ColumnSet.encodeColumns(workSectionCol.getColumnFamily(), workSectionCol.getColumnQualifier()), cols);
+ }
+
+ boolean foundLocalityGroups = false;
+ boolean foundLocalityGroupDef1 = false;
+ boolean foundLocalityGroupDef2 = false;
+ boolean foundFormatter = false;
+ Joiner j = Joiner.on(",");
+ Function<Text,String> textToString = new Function<Text,String>() {
+ @Override
+ public String apply(Text text) {
+ return text.toString();
+ }
+ };
+ for (Entry<String,String> p : tops.getProperties(ReplicationTable.NAME)) {
+ String key = p.getKey();
+ String val = p.getValue();
+ // STATUS_LG_NAME, STATUS_LG_COLFAMS, WORK_LG_NAME, WORK_LG_COLFAMS
+ if (key.equals(Property.TABLE_FORMATTER_CLASS.getKey()) && val.equals(StatusFormatter.class.getName())) {
+ // look for formatter
+ foundFormatter = true;
+ } else if (key.equals(Property.TABLE_LOCALITY_GROUPS.getKey()) && val.equals(j.join(ReplicationTable.LOCALITY_GROUPS.keySet()))) {
+ // look for locality groups enabled
+ foundLocalityGroups = true;
+ } else if (key.startsWith(Property.TABLE_LOCALITY_GROUP_PREFIX.getKey())) {
+ // look for locality group column family definitions
+ if (key.equals(Property.TABLE_LOCALITY_GROUP_PREFIX.getKey() + ReplicationTable.STATUS_LG_NAME)
+ && val.equals(j.join(Iterables.transform(ReplicationTable.STATUS_LG_COLFAMS, textToString)))) {
+ foundLocalityGroupDef1 = true;
+ } else if (key.equals(Property.TABLE_LOCALITY_GROUP_PREFIX.getKey() + ReplicationTable.WORK_LG_NAME)
+ && val.equals(j.join(Iterables.transform(ReplicationTable.WORK_LG_COLFAMS, textToString)))) {
+ foundLocalityGroupDef2 = true;
+ }
+ }
+ }
+ Assert.assertTrue(foundLocalityGroups);
+ Assert.assertTrue(foundLocalityGroupDef1);
+ Assert.assertTrue(foundLocalityGroupDef2);
+ Assert.assertTrue(foundFormatter);
+ }
+
+ @Test
public void correctRecordsCompleteFile() throws Exception {
Connector conn = getConnector();
String table = "table1";
@@ -146,17 +226,17 @@ public class ReplicationIT extends ConfigurableMacIT {
bw.close();
- // After writing data, we'll get a replication table
- boolean exists = conn.tableOperations().exists(ReplicationTable.NAME);
+ // After writing data, we'll get a replication table online
+ boolean online = ReplicationTable.isOnline(conn);
int attempts = 10;
do {
- if (!exists) {
+ if (!online) {
UtilWaitThread.sleep(2000);
- exists = conn.tableOperations().exists(ReplicationTable.NAME);
+ online = ReplicationTable.isOnline(conn);
attempts--;
}
- } while (!exists && attempts > 0);
- Assert.assertTrue("Replication table did not exist", exists);
+ } while (!online && attempts > 0);
+ Assert.assertTrue("Replication table was not online", online);
for (int i = 0; i < 5; i++) {
if (conn.securityOperations().hasTablePermission("root", ReplicationTable.NAME, TablePermission.READ)) {
@@ -193,7 +273,7 @@ public class ReplicationIT extends ConfigurableMacIT {
Scanner s;
attempts = 5;
while (wals.isEmpty() && attempts > 0) {
- s = conn.createScanner(MetadataTable.NAME, new Authorizations());
+ s = conn.createScanner(MetadataTable.NAME, Authorizations.EMPTY);
s.fetchColumnFamily(MetadataSchema.TabletsSection.LogColumnFamily.NAME);
for (Entry<Key,Value> entry : s) {
LogEntry logEntry = LogEntry.fromKeyValue(entry.getKey(), entry.getValue());
@@ -215,8 +295,8 @@ public class ReplicationIT extends ConfigurableMacIT {
Connector conn = getConnector();
List<String> tables = new ArrayList<>();
- // replication shouldn't exist when we begin
- Assert.assertFalse(conn.tableOperations().exists(ReplicationTable.NAME));
+ // replication shouldn't be online when we begin
+ Assert.assertFalse(ReplicationTable.isOnline(conn));
for (int i = 0; i < 5; i++) {
String name = "table" + i;
@@ -225,7 +305,7 @@ public class ReplicationIT extends ConfigurableMacIT {
}
// nor after we create some tables (that aren't being replicated)
- Assert.assertFalse(conn.tableOperations().exists(ReplicationTable.NAME));
+ Assert.assertFalse(ReplicationTable.isOnline(conn));
for (String table : tables) {
BatchWriter bw = conn.createBatchWriter(table, new BatchWriterConfig());
@@ -243,21 +323,21 @@ public class ReplicationIT extends ConfigurableMacIT {
}
// After writing data, still no replication table
- Assert.assertFalse(conn.tableOperations().exists(ReplicationTable.NAME));
+ Assert.assertFalse(ReplicationTable.isOnline(conn));
for (String table : tables) {
conn.tableOperations().compact(table, null, null, true, true);
}
// After compacting data, still no replication table
- Assert.assertFalse(conn.tableOperations().exists(ReplicationTable.NAME));
+ Assert.assertFalse(ReplicationTable.isOnline(conn));
for (String table : tables) {
conn.tableOperations().delete(table);
}
// After deleting tables, still no replication table
- Assert.assertFalse(conn.tableOperations().exists(ReplicationTable.NAME));
+ Assert.assertFalse(ReplicationTable.isOnline(conn));
}
@Test
@@ -266,7 +346,7 @@ public class ReplicationIT extends ConfigurableMacIT {
String table1 = "table1", table2 = "table2";
// replication shouldn't exist when we begin
- Assert.assertFalse("Replication table already existed at the beginning of the test", conn.tableOperations().exists(ReplicationTable.NAME));
+ Assert.assertFalse("Replication table already online at the beginning of the test", ReplicationTable.isOnline(conn));
// Create two tables
conn.tableOperations().create(table1);
@@ -275,8 +355,8 @@ public class ReplicationIT extends ConfigurableMacIT {
// Enable replication on table1
conn.tableOperations().setProperty(table1, Property.TABLE_REPLICATION.getKey(), "true");
- // Despite having replication on, we shouldn't have any need to write a record to it (and create it)
- Assert.assertFalse(conn.tableOperations().exists(ReplicationTable.NAME));
+ // Despite having replication on, we shouldn't have any need to write a record to it (and bring it online)
+ Assert.assertFalse(ReplicationTable.isOnline(conn));
// Write some data to table1
BatchWriter bw = conn.createBatchWriter(table1, new BatchWriterConfig());
@@ -295,19 +375,19 @@ public class ReplicationIT extends ConfigurableMacIT {
// After the commit for these mutations finishes, we'll get a replication entry in accumulo.metadata for table1
// Don't want to compact table1 as it ultimately cause the entry in accumulo.metadata to be removed before we can verify it's there
- // After writing data, we'll get a replication table
- boolean exists = conn.tableOperations().exists(ReplicationTable.NAME);
+ // After writing data, we'll get a replication table online
+ boolean online = ReplicationTable.isOnline(conn);
int attempts = 10;
do {
- if (!exists) {
+ if (!online) {
UtilWaitThread.sleep(5000);
- exists = conn.tableOperations().exists(ReplicationTable.NAME);
+ online = ReplicationTable.isOnline(conn);
attempts--;
}
- } while (!exists && attempts > 0);
- Assert.assertTrue("Replication table did not exist", exists);
+ } while (!online && attempts > 0);
+ Assert.assertTrue("Replication table did not exist", online);
- Assert.assertTrue(conn.tableOperations().exists(ReplicationTable.NAME));
+ Assert.assertTrue(ReplicationTable.isOnline(conn));
conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.READ);
// Verify that we found a single replication record that's for table1
@@ -353,8 +433,8 @@ public class ReplicationIT extends ConfigurableMacIT {
// After the commit on these mutations, we'll get a replication entry in accumulo.metadata for table2
// Don't want to compact table2 as it ultimately cause the entry in accumulo.metadata to be removed before we can verify it's there
- // After writing data, we'll get a replication table
- Assert.assertTrue(conn.tableOperations().exists(ReplicationTable.NAME));
+ // After writing data, we'll get a replication table online
+ Assert.assertTrue(ReplicationTable.isOnline(conn));
conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.READ);
Set<String> tableIds = Sets.newHashSet(conn.tableOperations().tableIdMap().get(table1), conn.tableOperations().tableIdMap().get(table2));
@@ -558,12 +638,9 @@ public class ReplicationIT extends ConfigurableMacIT {
public void noDeadlock() throws Exception {
final Connector conn = getConnector();
- if (conn.tableOperations().exists(ReplicationTable.NAME)) {
- conn.tableOperations().delete(ReplicationTable.NAME);
- }
-
- ReplicationUtil.createReplicationTable(conn);
+ ReplicationTable.setOnline(conn);
conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.WRITE);
+ conn.tableOperations().deleteRows(ReplicationTable.NAME, null, null);
final AtomicBoolean keepRunning = new AtomicBoolean(true);
final Set<String> metadataWals = new HashSet<>();
@@ -648,7 +725,7 @@ public class ReplicationIT extends ConfigurableMacIT {
}
for (String table : Arrays.asList(MetadataTable.NAME, table1, table2, table3)) {
- Scanner s = conn.createScanner(table, new Authorizations());
+ Scanner s = conn.createScanner(table, Authorizations.EMPTY);
for (@SuppressWarnings("unused")
Entry<Key,Value> entry : s) {}
}
@@ -712,7 +789,7 @@ public class ReplicationIT extends ConfigurableMacIT {
conn.tableOperations().flush(table, null, null, true);
- while (!conn.tableOperations().exists(ReplicationTable.NAME)) {
+ while (!ReplicationTable.isOnline(conn)) {
UtilWaitThread.sleep(2000);
}
@@ -778,8 +855,8 @@ public class ReplicationIT extends ConfigurableMacIT {
Connector conn = getConnector();
String table1 = "table1";
- // replication shouldn't exist when we begin
- Assert.assertFalse(conn.tableOperations().exists(ReplicationTable.NAME));
+ // replication shouldn't be online when we begin
+ Assert.assertFalse(ReplicationTable.isOnline(conn));
// Create a table
conn.tableOperations().create(table1);
@@ -819,17 +896,17 @@ public class ReplicationIT extends ConfigurableMacIT {
bw.close();
- // Make sure the replication table exists at this point
- boolean exists = conn.tableOperations().exists(ReplicationTable.NAME);
+ // Make sure the replication table is online at this point
+ boolean online = ReplicationTable.isOnline(conn);
attempts = 10;
do {
- if (!exists) {
+ if (!online) {
UtilWaitThread.sleep(2000);
- exists = conn.tableOperations().exists(ReplicationTable.NAME);
+ online = ReplicationTable.isOnline(conn);
attempts--;
}
- } while (!exists && attempts > 0);
- Assert.assertTrue("Replication table was never created", exists);
+ } while (!online && attempts > 0);
+ Assert.assertTrue("Replication table was never created", online);
// ACCUMULO-2743 The Observer in the tserver has to be made aware of the change to get the combiner (made by the master)
for (int i = 0; i < 10 && !conn.tableOperations().listIterators(ReplicationTable.NAME).keySet().contains(ReplicationTable.COMBINER_NAME); i++) {
@@ -963,8 +1040,8 @@ public class ReplicationIT extends ConfigurableMacIT {
Connector conn = getConnector();
String table1 = "table1";
- // replication shouldn't exist when we begin
- Assert.assertFalse(conn.tableOperations().exists(ReplicationTable.NAME));
+ // replication shouldn't be online when we begin
+ Assert.assertFalse(ReplicationTable.isOnline(conn));
// Create two tables
conn.tableOperations().create(table1);
@@ -1003,16 +1080,16 @@ public class ReplicationIT extends ConfigurableMacIT {
Assert.assertNotNull("Table ID was null", tableId);
// Make sure the replication table exists at this point
- boolean exists = conn.tableOperations().exists(ReplicationTable.NAME);
+ boolean online = ReplicationTable.isOnline(conn);
attempts = 5;
do {
- if (!exists) {
+ if (!online) {
UtilWaitThread.sleep(500);
- exists = conn.tableOperations().exists(ReplicationTable.NAME);
+ online = ReplicationTable.isOnline(conn);
attempts--;
}
- } while (!exists && attempts > 0);
- Assert.assertTrue("Replication table did not exist", exists);
+ } while (!online && attempts > 0);
+ Assert.assertTrue("Replication table did not exist", online);
for (int i = 0; i < 5 && !conn.securityOperations().hasTablePermission("root", ReplicationTable.NAME, TablePermission.READ); i++) {
Thread.sleep(1000);
@@ -1059,12 +1136,9 @@ public class ReplicationIT extends ConfigurableMacIT {
final Connector conn = getConnector();
- if (conn.tableOperations().exists(ReplicationTable.NAME)) {
- conn.tableOperations().delete(ReplicationTable.NAME);
- }
-
- ReplicationUtil.createReplicationTable(conn);
+ ReplicationTable.setOnline(conn);
conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.WRITE);
+ conn.tableOperations().deleteRows(ReplicationTable.NAME, null, null);
final AtomicBoolean keepRunning = new AtomicBoolean(true);
final Set<String> metadataWals = new HashSet<>();
@@ -1164,7 +1238,7 @@ public class ReplicationIT extends ConfigurableMacIT {
// Make sure we can read all the tables (recovery complete)
for (String table : Arrays.asList(table1, table2, table3)) {
- Scanner s = conn.createScanner(table, new Authorizations());
+ Scanner s = conn.createScanner(table, Authorizations.EMPTY);
for (@SuppressWarnings("unused")
Entry<Key,Value> entry : s) {}
}
@@ -1265,8 +1339,8 @@ public class ReplicationIT extends ConfigurableMacIT {
log.info("Got connector to MAC");
String table1 = "table1";
- // replication shouldn't exist when we begin
- Assert.assertFalse(conn.tableOperations().exists(ReplicationTable.NAME));
+ // replication shouldn't be online when we begin
+ Assert.assertFalse(ReplicationTable.isOnline(conn));
// Create two tables
conn.tableOperations().create(table1);
@@ -1308,16 +1382,16 @@ public class ReplicationIT extends ConfigurableMacIT {
bw.close();
// Make sure the replication table exists at this point
- boolean exists = conn.tableOperations().exists(ReplicationTable.NAME);
+ boolean online = ReplicationTable.isOnline(conn);
attempts = 10;
do {
- if (!exists) {
+ if (!online) {
UtilWaitThread.sleep(1000);
- exists = conn.tableOperations().exists(ReplicationTable.NAME);
+ online = ReplicationTable.isOnline(conn);
attempts--;
}
- } while (!exists && attempts > 0);
- Assert.assertTrue("Replication table did not exist", exists);
+ } while (!online && attempts > 0);
+ Assert.assertTrue("Replication table did not exist", online);
// Grant ourselves the write permission for later
conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.WRITE);
@@ -1391,7 +1465,7 @@ public class ReplicationIT extends ConfigurableMacIT {
// Make sure we can read all the tables (recovery complete)
for (String table : new String[] {MetadataTable.NAME, table1}) {
- s = conn.createScanner(table, new Authorizations());
+ s = conn.createScanner(table, Authorizations.EMPTY);
for (@SuppressWarnings("unused")
Entry<Key,Value> entry : s) {}
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e840549a/test/src/test/java/org/apache/accumulo/test/replication/StatusCombinerMacTest.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/replication/StatusCombinerMacTest.java b/test/src/test/java/org/apache/accumulo/test/replication/StatusCombinerMacTest.java
index ff64d0b..98d7676 100644
--- a/test/src/test/java/org/apache/accumulo/test/replication/StatusCombinerMacTest.java
+++ b/test/src/test/java/org/apache/accumulo/test/replication/StatusCombinerMacTest.java
@@ -22,7 +22,6 @@ import java.util.Map;
import java.util.Map.Entry;
import org.apache.accumulo.core.client.BatchWriter;
-import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.admin.TableOperations;
@@ -38,8 +37,7 @@ import org.apache.accumulo.core.replication.ReplicationSchema.StatusSection;
import org.apache.accumulo.core.replication.ReplicationTable;
import org.apache.accumulo.core.replication.StatusUtil;
import org.apache.accumulo.core.replication.proto.Replication.Status;
-import org.apache.accumulo.core.security.Authorizations;
-import org.apache.accumulo.server.replication.ReplicationUtil;
+import org.apache.accumulo.core.security.TablePermission;
import org.apache.accumulo.server.util.ReplicationTableUtil;
import org.apache.accumulo.test.functional.SimpleMacIT;
import org.apache.hadoop.io.Text;
@@ -81,13 +79,10 @@ public class StatusCombinerMacTest extends SimpleMacIT {
@Test
public void test() throws Exception {
Connector conn = getConnector();
- if (conn.tableOperations().exists(ReplicationTable.NAME)) {
- conn.tableOperations().delete(ReplicationTable.NAME);
- }
-
- ReplicationUtil.createReplicationTable(conn);
- BatchWriter bw = conn.createBatchWriter(ReplicationTable.NAME, new BatchWriterConfig());
+ ReplicationTable.setOnline(conn);
+ conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.WRITE);
+ BatchWriter bw = ReplicationTable.getBatchWriter(conn);
long createTime = System.currentTimeMillis();
try {
Mutation m = new Mutation("file:/accumulo/wal/HW10447.local+56808/93cdc17e-7521-44fa-87b5-37f45bcb92d3");
@@ -97,11 +92,11 @@ public class StatusCombinerMacTest extends SimpleMacIT {
bw.close();
}
- Scanner s = conn.createScanner(ReplicationTable.NAME, new Authorizations());
+ Scanner s = ReplicationTable.getScanner(conn);
Entry<Key,Value> entry = Iterables.getOnlyElement(s);
Assert.assertEquals(StatusUtil.fileCreatedValue(createTime), entry.getValue());
- bw = conn.createBatchWriter(ReplicationTable.NAME, new BatchWriterConfig());
+ bw = ReplicationTable.getBatchWriter(conn);
try {
Mutation m = new Mutation("file:/accumulo/wal/HW10447.local+56808/93cdc17e-7521-44fa-87b5-37f45bcb92d3");
StatusSection.add(m, new Text("1"), ProtobufUtil.toValue(StatusUtil.replicated(Long.MAX_VALUE)));
@@ -110,7 +105,7 @@ public class StatusCombinerMacTest extends SimpleMacIT {
bw.close();
}
- s = conn.createScanner(ReplicationTable.NAME, new Authorizations());
+ s = ReplicationTable.getScanner(conn);
entry = Iterables.getOnlyElement(s);
Status stat = Status.parseFrom(entry.getValue().get());
Assert.assertEquals(Long.MAX_VALUE, stat.getBegin());