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 2020/07/16 17:07:31 UTC
[accumulo] 01/03: Merge branch '1.9'
This is an automated email from the ASF dual-hosted git repository.
ctubbsii pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/accumulo.git
commit 1caf7c7927aa12a7d8380f210cca4ab8afb497c8
Merge: d879f58 1b2946f
Author: Christopher Tubbs <ct...@apache.org>
AuthorDate: Thu Jul 16 03:13:28 2020 -0400
Merge branch '1.9'
.../accumulo/test/functional/CloneTestIT.java | 6 +-
.../test/replication/KerberosReplicationIT.java | 12 ++--
.../replication/MultiInstanceReplicationIT.java | 74 +++++++++++++---------
.../UnorderedWorkAssignerReplicationIT.java | 55 +++++++++-------
4 files changed, 86 insertions(+), 61 deletions(-)
diff --cc test/src/main/java/org/apache/accumulo/test/functional/CloneTestIT.java
index 53db5f4,5201bcf..117367c
--- a/test/src/main/java/org/apache/accumulo/test/functional/CloneTestIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/CloneTestIT.java
@@@ -299,39 -311,40 +299,41 @@@ public class CloneTestIT extends Accumu
@Test
public void testCloneWithSplits() throws Exception {
- Connector conn = getConnector();
-
- List<Mutation> mutations = new ArrayList<>();
- TreeSet<Text> splits = new TreeSet<>();
- for (int i = 0; i < 10; i++) {
- splits.add(new Text(Integer.toString(i)));
- Mutation m = new Mutation(Integer.toString(i));
- m.put("", "", "");
- mutations.add(m);
- }
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+
+ List<Mutation> mutations = new ArrayList<>();
+ TreeSet<Text> splits = new TreeSet<>();
+ for (int i = 0; i < 10; i++) {
+ splits.add(new Text(Integer.toString(i)));
+ Mutation m = new Mutation(Integer.toString(i));
+ m.put("", "", "");
+ mutations.add(m);
+ }
- String[] tables = getUniqueNames(2);
+ String[] tables = getUniqueNames(2);
- conn.tableOperations().create(tables[0]);
+ client.tableOperations().create(tables[0]);
- conn.tableOperations().addSplits(tables[0], splits);
+ client.tableOperations().addSplits(tables[0], splits);
- BatchWriter bw = conn.createBatchWriter(tables[0], new BatchWriterConfig());
- bw.addMutations(mutations);
- bw.close();
+ try (BatchWriter bw = client.createBatchWriter(tables[0])) {
+ bw.addMutations(mutations);
+ }
- conn.tableOperations().clone(tables[0], tables[1], true, null, null);
+ client.tableOperations().clone(tables[0], tables[1], true, null, null);
- conn.tableOperations().deleteRows(tables[1], new Text("4"), new Text("8"));
+ client.tableOperations().deleteRows(tables[1], new Text("4"), new Text("8"));
- List<String> rows = Arrays.asList("0", "1", "2", "3", "4", "9");
- List<String> actualRows = new ArrayList<>();
- try (Scanner s = conn.createScanner(tables[1], Authorizations.EMPTY)) {
- for (Entry<Key,Value> entry : s) {
- actualRows.add(entry.getKey().getRow().toString());
+ List<String> rows = Arrays.asList("0", "1", "2", "3", "4", "9");
+ List<String> actualRows = new ArrayList<>();
- for (Entry<Key,Value> entry : client.createScanner(tables[1], Authorizations.EMPTY)) {
- actualRows.add(entry.getKey().getRow().toString());
++ try (var scanner = client.createScanner(tables[1], Authorizations.EMPTY)) {
++ for (Entry<Key,Value> entry : scanner) {
++ actualRows.add(entry.getKey().getRow().toString());
++ }
}
- }
- assertEquals(rows, actualRows);
+ assertEquals(rows, actualRows);
+ }
}
@Test(expected = AccumuloException.class)
diff --cc test/src/main/java/org/apache/accumulo/test/replication/KerberosReplicationIT.java
index 816ffa2,805085e..56a3d92
--- a/test/src/main/java/org/apache/accumulo/test/replication/KerberosReplicationIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/replication/KerberosReplicationIT.java
@@@ -250,13 -243,17 +250,17 @@@ public class KerberosReplicationIT exte
// Wait for both tables to be replicated
log.info("Waiting for {} for {}", filesFor1, primaryTable1);
- primaryConn.replicationOperations().drain(primaryTable1, filesFor1);
+ primaryclient.replicationOperations().drain(primaryTable1, filesFor1);
- long countTable = 0l;
- try (Scanner s = peerConn.createScanner(peerTable1, Authorizations.EMPTY)) {
- for (Entry<Key,Value> entry : s) {
+ long countTable = 0L;
- for (Entry<Key,Value> entry : peerclient.createScanner(peerTable1, Authorizations.EMPTY)) {
- countTable++;
- assertTrue("Found unexpected key-value" + entry.getKey().toStringNoTruncate() + " "
- + entry.getValue(), entry.getKey().getRow().toString().startsWith(primaryTable1));
++ try (var scanner = peerclient.createScanner(peerTable1, Authorizations.EMPTY)) {
++ for (Entry<Key,Value> entry : scanner) {
+ countTable++;
+ assertTrue(
+ "Found unexpected key-value" + entry.getKey().toStringNoTruncate() + " "
+ + entry.getValue(),
+ entry.getKey().getRow().toString().startsWith(primaryTable1));
+ }
}
log.info("Found {} records in {}", countTable, peerTable1);
diff --cc test/src/main/java/org/apache/accumulo/test/replication/MultiInstanceReplicationIT.java
index b77344e,6e22b31..9733308
--- a/test/src/main/java/org/apache/accumulo/test/replication/MultiInstanceReplicationIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/replication/MultiInstanceReplicationIT.java
@@@ -259,29 -255,37 +259,32 @@@ public class MultiInstanceReplicationI
log.info("");
log.info("Fetching metadata records:");
- for (Entry<Key,Value> kv : clientMaster.createScanner(MetadataTable.NAME,
- Authorizations.EMPTY)) {
- if (ReplicationSection.COLF.equals(kv.getKey().getColumnFamily())) {
- log.info("{} {}", kv.getKey().toStringNoTruncate(),
- ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
- } else {
- log.info("{} {}", kv.getKey().toStringNoTruncate(), kv.getValue());
- try (Scanner s = connMaster.createScanner(MetadataTable.NAME, Authorizations.EMPTY)) {
- for (Entry<Key,Value> kv : s) {
++ try (var scanner = clientMaster.createScanner(MetadataTable.NAME, Authorizations.EMPTY)) {
++ for (Entry<Key,Value> kv : scanner) {
+ if (ReplicationSection.COLF.equals(kv.getKey().getColumnFamily())) {
- log.info(kv.getKey().toStringNoTruncate() + " "
- + ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
++ log.info("{} {}", kv.getKey().toStringNoTruncate(),
++ ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
+ } else {
- log.info(kv.getKey().toStringNoTruncate() + " " + kv.getValue());
++ log.info("{} {}", kv.getKey().toStringNoTruncate(), kv.getValue());
+ }
}
}
log.info("");
log.info("Fetching replication records:");
- for (Entry<Key,Value> kv : ReplicationTable.getScanner(clientMaster)) {
- log.info("{} {}", kv.getKey().toStringNoTruncate(),
- ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
- try (Scanner s = ReplicationTable.getScanner(connMaster)) {
- for (Entry<Key,Value> kv : s) {
- log.info(kv.getKey().toStringNoTruncate() + " "
- + ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
++ try (var scanner = ReplicationTable.getScanner(clientMaster)) {
++ for (Entry<Key,Value> kv : scanner) {
++ log.info("{} {}", kv.getKey().toStringNoTruncate(),
++ ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
+ }
}
- Future<Boolean> future = executor.submit(new Callable<Boolean>() {
-
- @Override
- public Boolean call() throws Exception {
- long then = System.currentTimeMillis();
- connMaster.replicationOperations().drain(masterTable, filesNeedingReplication);
- long now = System.currentTimeMillis();
- log.info("Drain completed in " + (now - then) + "ms");
- return true;
- }
-
+ Future<Boolean> future = executor.submit(() -> {
+ long then = System.currentTimeMillis();
+ clientMaster.replicationOperations().drain(masterTable, filesNeedingReplication);
+ long now = System.currentTimeMillis();
+ log.info("Drain completed in " + (now - then) + "ms");
+ return true;
});
try {
@@@ -297,42 -301,44 +300,45 @@@
log.info("");
log.info("Fetching metadata records:");
- for (Entry<Key,Value> kv : clientMaster.createScanner(MetadataTable.NAME,
- Authorizations.EMPTY)) {
- if (ReplicationSection.COLF.equals(kv.getKey().getColumnFamily())) {
- log.info("{} {}", kv.getKey().toStringNoTruncate(),
- ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
- } else {
- log.info("{} {}", kv.getKey().toStringNoTruncate(), kv.getValue());
- try (Scanner s = connMaster.createScanner(MetadataTable.NAME, Authorizations.EMPTY)) {
- for (Entry<Key,Value> kv : s) {
++ try (var scanner = clientMaster.createScanner(MetadataTable.NAME, Authorizations.EMPTY)) {
++ for (Entry<Key,Value> kv : scanner) {
+ if (ReplicationSection.COLF.equals(kv.getKey().getColumnFamily())) {
- log.info(kv.getKey().toStringNoTruncate() + " "
- + ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
++ log.info("{} {}", kv.getKey().toStringNoTruncate(),
++ ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
+ } else {
- log.info(kv.getKey().toStringNoTruncate() + " " + kv.getValue());
++ log.info("{} {}", kv.getKey().toStringNoTruncate(), kv.getValue());
+ }
}
}
log.info("");
log.info("Fetching replication records:");
- for (Entry<Key,Value> kv : ReplicationTable.getScanner(clientMaster)) {
- log.info("{} {}", kv.getKey().toStringNoTruncate(),
- ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
- try (Scanner s = ReplicationTable.getScanner(connMaster)) {
- for (Entry<Key,Value> kv : s) {
- log.info(kv.getKey().toStringNoTruncate() + " "
- + ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
++ try (var scanner = ReplicationTable.getScanner(clientMaster)) {
++ for (Entry<Key,Value> kv : scanner) {
++ log.info("{} {}", kv.getKey().toStringNoTruncate(),
++ ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
+ }
}
- Scanner master = connMaster.createScanner(masterTable, Authorizations.EMPTY),
- peer = connPeer.createScanner(peerTable, Authorizations.EMPTY);
- Iterator<Entry<Key,Value>> masterIter = master.iterator(), peerIter = peer.iterator();
- Entry<Key,Value> masterEntry = null, peerEntry = null;
- while (masterIter.hasNext() && peerIter.hasNext()) {
- masterEntry = masterIter.next();
- peerEntry = peerIter.next();
- assertEquals(masterEntry.getKey() + " was not equal to " + peerEntry.getKey(), 0,
- masterEntry.getKey().compareTo(peerEntry.getKey(),
- PartialKey.ROW_COLFAM_COLQUAL_COLVIS));
- assertEquals(masterEntry.getValue(), peerEntry.getValue());
- }
+ try (Scanner master = clientMaster.createScanner(masterTable, Authorizations.EMPTY);
+ Scanner peer = clientPeer.createScanner(peerTable, Authorizations.EMPTY)) {
+ Iterator<Entry<Key,Value>> masterIter = master.iterator(), peerIter = peer.iterator();
+ Entry<Key,Value> masterEntry = null, peerEntry = null;
+ while (masterIter.hasNext() && peerIter.hasNext()) {
+ masterEntry = masterIter.next();
+ peerEntry = peerIter.next();
+ assertEquals(masterEntry.getKey() + " was not equal to " + peerEntry.getKey(), 0,
+ masterEntry.getKey().compareTo(peerEntry.getKey(),
+ PartialKey.ROW_COLFAM_COLQUAL_COLVIS));
+ assertEquals(masterEntry.getValue(), peerEntry.getValue());
+ }
- log.info("Last master entry: " + masterEntry);
- log.info("Last peer entry: " + peerEntry);
+ log.info("Last master entry: {}", masterEntry);
+ log.info("Last peer entry: {}", peerEntry);
- assertFalse("Had more data to read from the master", masterIter.hasNext());
- assertFalse("Had more data to read from the peer", peerIter.hasNext());
+ assertFalse("Had more data to read from the master", masterIter.hasNext());
+ assertFalse("Had more data to read from the peer", peerIter.hasNext());
+ }
} finally {
peerCluster.stop();
}
@@@ -464,26 -474,30 +470,30 @@@
// Wait for both tables to be replicated
log.info("Waiting for {} for {}", filesFor1, masterTable1);
- connMaster.replicationOperations().drain(masterTable1, filesFor1);
+ clientMaster.replicationOperations().drain(masterTable1, filesFor1);
log.info("Waiting for {} for {}", filesFor2, masterTable2);
- connMaster.replicationOperations().drain(masterTable2, filesFor2);
+ clientMaster.replicationOperations().drain(masterTable2, filesFor2);
- long countTable = 0l;
- try (Scanner s = connPeer.createScanner(peerTable1, Authorizations.EMPTY)) {
- for (Entry<Key,Value> entry : s) {
+ long countTable = 0L;
- for (Entry<Key,Value> entry : clientPeer.createScanner(peerTable1, Authorizations.EMPTY)) {
- countTable++;
- assertTrue("Found unexpected key-value" + entry.getKey().toStringNoTruncate() + " "
- + entry.getValue(), entry.getKey().getRow().toString().startsWith(masterTable1));
++ try (var scanner = clientPeer.createScanner(peerTable1, Authorizations.EMPTY)) {
++ for (Entry<Key,Value> entry : scanner) {
+ countTable++;
+ assertTrue("Found unexpected key-value" + entry.getKey().toStringNoTruncate() + " "
+ + entry.getValue(), entry.getKey().getRow().toString().startsWith(masterTable1));
+ }
}
log.info("Found {} records in {}", countTable, peerTable1);
assertEquals(masterTable1Records, countTable);
- countTable = 0l;
- try (Scanner s = connPeer.createScanner(peerTable2, Authorizations.EMPTY)) {
- for (Entry<Key,Value> entry : s) {
+ countTable = 0L;
- for (Entry<Key,Value> entry : clientPeer.createScanner(peerTable2, Authorizations.EMPTY)) {
- countTable++;
- assertTrue("Found unexpected key-value" + entry.getKey().toStringNoTruncate() + " "
- + entry.getValue(), entry.getKey().getRow().toString().startsWith(masterTable2));
++ try (var scanner = clientPeer.createScanner(peerTable2, Authorizations.EMPTY)) {
++ for (Entry<Key,Value> entry : scanner) {
+ countTable++;
+ assertTrue("Found unexpected key-value" + entry.getKey().toStringNoTruncate() + " "
+ + entry.getValue(), entry.getKey().getRow().toString().startsWith(masterTable2));
+ }
}
log.info("Found {} records in {}", countTable, peerTable2);
@@@ -509,106 -523,105 +519,108 @@@
peerCluster.start();
- Connector connMaster = getConnector();
- Connector connPeer = peerCluster.getConnector("root", new PasswordToken(ROOT_PASSWORD));
-
- String peerUserName = "repl";
- String peerPassword = "passwd";
-
- // Create a user on the peer for replication to use
- connPeer.securityOperations().createLocalUser(peerUserName, new PasswordToken(peerPassword));
-
- String peerClusterName = "peer";
-
- // ...peer = AccumuloReplicaSystem,instanceName,zookeepers
- connMaster.instanceOperations().setProperty(
- Property.REPLICATION_PEERS.getKey() + peerClusterName,
- ReplicaSystemFactory.getPeerConfigurationValue(AccumuloReplicaSystem.class,
- AccumuloReplicaSystem.buildConfiguration(peerCluster.getInstanceName(),
- peerCluster.getZooKeepers())));
-
- // Configure the credentials we should use to authenticate ourselves to the peer for replication
- connMaster.instanceOperations()
- .setProperty(Property.REPLICATION_PEER_USER.getKey() + peerClusterName, peerUserName);
- connMaster.instanceOperations()
- .setProperty(Property.REPLICATION_PEER_PASSWORD.getKey() + peerClusterName, peerPassword);
-
- String masterTable = "master", peerTable = "peer";
-
- connMaster.tableOperations().create(masterTable);
- String masterTableId = connMaster.tableOperations().tableIdMap().get(masterTable);
- assertNotNull(masterTableId);
-
- connPeer.tableOperations().create(peerTable);
- String peerTableId = connPeer.tableOperations().tableIdMap().get(peerTable);
- assertNotNull(peerTableId);
-
- // Give our replication user the ability to write to the table
- connPeer.securityOperations().grantTablePermission(peerUserName, peerTable,
- TablePermission.WRITE);
-
- // Replicate this table to the peerClusterName in a table with the peerTableId table id
- connMaster.tableOperations().setProperty(masterTable, Property.TABLE_REPLICATION.getKey(),
- "true");
- connMaster.tableOperations().setProperty(masterTable,
- Property.TABLE_REPLICATION_TARGET.getKey() + peerClusterName, peerTableId);
-
- // Write some data to table1
- BatchWriter bw = connMaster.createBatchWriter(masterTable, new BatchWriterConfig());
- for (int rows = 0; rows < 5000; rows++) {
- Mutation m = new Mutation(Integer.toString(rows));
- for (int cols = 0; cols < 100; cols++) {
- String value = Integer.toString(cols);
- m.put(value, "", value);
- }
- bw.addMutation(m);
- }
+ try (AccumuloClient clientMaster = Accumulo.newClient().from(getClientProperties()).build();
+ AccumuloClient clientPeer =
+ peerCluster.createAccumuloClient("root", new PasswordToken(ROOT_PASSWORD))) {
- bw.close();
+ String peerUserName = "repl";
+ String peerPassword = "passwd";
- log.info("Wrote all data to master cluster");
+ // Create a user on the peer for replication to use
+ clientPeer.securityOperations().createLocalUser(peerUserName,
+ new PasswordToken(peerPassword));
- Set<String> files = connMaster.replicationOperations().referencedFiles(masterTable);
+ String peerClusterName = "peer";
- log.info("Files to replicate:" + files);
+ // ...peer = AccumuloReplicaSystem,instanceName,zookeepers
+ clientMaster.instanceOperations().setProperty(
+ Property.REPLICATION_PEERS.getKey() + peerClusterName,
+ ReplicaSystemFactory.getPeerConfigurationValue(AccumuloReplicaSystem.class,
+ AccumuloReplicaSystem.buildConfiguration(peerCluster.getInstanceName(),
+ peerCluster.getZooKeepers())));
- for (ProcessReference proc : cluster.getProcesses().get(ServerType.TABLET_SERVER)) {
- cluster.killProcess(ServerType.TABLET_SERVER, proc);
- }
+ // Configure the credentials we should use to authenticate ourselves to the peer for
+ // replication
+ clientMaster.instanceOperations()
+ .setProperty(Property.REPLICATION_PEER_USER.getKey() + peerClusterName, peerUserName);
+ clientMaster.instanceOperations()
+ .setProperty(Property.REPLICATION_PEER_PASSWORD.getKey() + peerClusterName, peerPassword);
- cluster.exec(TabletServer.class);
+ String masterTable = "master", peerTable = "peer";
+ clientPeer.tableOperations().create(peerTable, new NewTableConfiguration());
+ String peerTableId = clientPeer.tableOperations().tableIdMap().get(peerTable);
+ assertNotNull(peerTableId);
- while (!ReplicationTable.isOnline(connMaster)) {
- log.info("Replication table still offline, waiting");
- Thread.sleep(5000);
- }
+ // Give our replication user the ability to write to the table
+ clientPeer.securityOperations().grantTablePermission(peerUserName, peerTable,
+ TablePermission.WRITE);
- Iterators.size(connMaster.createScanner(masterTable, Authorizations.EMPTY).iterator());
+ Map<String,String> props = new HashMap<>();
+ props.put(Property.TABLE_REPLICATION.getKey(), "true");
+ // Replicate this table to the peerClusterName in a table with the peerTableId table id
+ props.put(Property.TABLE_REPLICATION_TARGET.getKey() + peerClusterName, peerTableId);
+ clientMaster.tableOperations().create(masterTable,
+ new NewTableConfiguration().setProperties(props));
+ String masterTableId = clientMaster.tableOperations().tableIdMap().get(masterTable);
+ assertNotNull(masterTableId);
- try (Scanner s = ReplicationTable.getScanner(connMaster)) {
- for (Entry<Key,Value> kv : s) {
- log.debug(kv.getKey().toStringNoTruncate() + " "
- + ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
+ // Write some data to table1
+ try (BatchWriter bw = clientMaster.createBatchWriter(masterTable)) {
+ for (int rows = 0; rows < 5000; rows++) {
+ Mutation m = new Mutation(Integer.toString(rows));
+ for (int cols = 0; cols < 100; cols++) {
+ String value = Integer.toString(cols);
+ m.put(value, "", value);
+ }
+ bw.addMutation(m);
+ }
}
- }
- connMaster.replicationOperations().drain(masterTable, files);
+ log.info("Wrote all data to master cluster");
- Scanner master = connMaster.createScanner(masterTable, Authorizations.EMPTY),
- peer = connPeer.createScanner(peerTable, Authorizations.EMPTY);
- Iterator<Entry<Key,Value>> masterIter = master.iterator(), peerIter = peer.iterator();
- while (masterIter.hasNext() && peerIter.hasNext()) {
- Entry<Key,Value> masterEntry = masterIter.next(), peerEntry = peerIter.next();
- assertEquals(peerEntry.getKey() + " was not equal to " + peerEntry.getKey(), 0,
- masterEntry.getKey().compareTo(peerEntry.getKey(), PartialKey.ROW_COLFAM_COLQUAL_COLVIS));
- assertEquals(masterEntry.getValue(), peerEntry.getValue());
- }
+ Set<String> files = clientMaster.replicationOperations().referencedFiles(masterTable);
- assertFalse("Had more data to read from the master", masterIter.hasNext());
- assertFalse("Had more data to read from the peer", peerIter.hasNext());
+ log.info("Files to replicate:" + files);
- peerCluster.stop();
+ for (ProcessReference proc : cluster.getProcesses().get(ServerType.TABLET_SERVER)) {
+ cluster.killProcess(ServerType.TABLET_SERVER, proc);
+ }
+
+ cluster.exec(TabletServer.class);
+
+ while (!ReplicationTable.isOnline(clientMaster)) {
+ log.info("Replication table still offline, waiting");
+ Thread.sleep(5000);
+ }
+
+ Iterators.size(clientMaster.createScanner(masterTable, Authorizations.EMPTY).iterator());
+
- for (Entry<Key,Value> kv : ReplicationTable.getScanner(clientMaster)) {
- log.debug("{} {}", kv.getKey().toStringNoTruncate(),
- ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
++ try (var scanner = ReplicationTable.getScanner(clientMaster)) {
++ for (Entry<Key,Value> kv : scanner) {
++ log.debug("{} {}", kv.getKey().toStringNoTruncate(),
++ ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
++ }
+ }
+
+ clientMaster.replicationOperations().drain(masterTable, files);
+
+ try (Scanner master = clientMaster.createScanner(masterTable, Authorizations.EMPTY);
+ Scanner peer = clientPeer.createScanner(peerTable, Authorizations.EMPTY)) {
+ Iterator<Entry<Key,Value>> masterIter = master.iterator(), peerIter = peer.iterator();
+ while (masterIter.hasNext() && peerIter.hasNext()) {
+ Entry<Key,Value> masterEntry = masterIter.next(), peerEntry = peerIter.next();
+ assertEquals(peerEntry.getKey() + " was not equal to " + peerEntry.getKey(), 0,
+ masterEntry.getKey().compareTo(peerEntry.getKey(),
+ PartialKey.ROW_COLFAM_COLQUAL_COLVIS));
+ assertEquals(masterEntry.getValue(), peerEntry.getValue());
+ }
+
+ assertFalse("Had more data to read from the master", masterIter.hasNext());
+ assertFalse("Had more data to read from the peer", peerIter.hasNext());
+ }
+ } finally {
+ peerCluster.stop();
+ }
}
@Test
diff --cc test/src/main/java/org/apache/accumulo/test/replication/UnorderedWorkAssignerReplicationIT.java
index beb29e4,889334b..62e2567
--- a/test/src/main/java/org/apache/accumulo/test/replication/UnorderedWorkAssignerReplicationIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/replication/UnorderedWorkAssignerReplicationIT.java
@@@ -259,27 -256,35 +259,30 @@@ public class UnorderedWorkAssignerRepli
log.info("");
log.info("Fetching metadata records:");
- for (Entry<Key,Value> kv : clientMaster.createScanner(MetadataTable.NAME,
- Authorizations.EMPTY)) {
- if (ReplicationSection.COLF.equals(kv.getKey().getColumnFamily())) {
- log.info("{} {}", kv.getKey().toStringNoTruncate(),
- ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
- } else {
- log.info("{} {}", kv.getKey().toStringNoTruncate(), kv.getValue());
- try (Scanner s = connMaster.createScanner(MetadataTable.NAME, Authorizations.EMPTY)) {
- for (Entry<Key,Value> kv : s) {
++ try (var scanner = clientMaster.createScanner(MetadataTable.NAME, Authorizations.EMPTY)) {
++ for (Entry<Key,Value> kv : scanner) {
+ if (ReplicationSection.COLF.equals(kv.getKey().getColumnFamily())) {
- log.info(kv.getKey().toStringNoTruncate() + " "
- + ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
++ log.info("{} {}", kv.getKey().toStringNoTruncate(),
++ ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
+ } else {
- log.info(kv.getKey().toStringNoTruncate() + " " + kv.getValue());
++ log.info("{} {}", kv.getKey().toStringNoTruncate(), kv.getValue());
+ }
}
}
log.info("");
log.info("Fetching replication records:");
- for (Entry<Key,Value> kv : ReplicationTable.getScanner(clientMaster)) {
- log.info("{} {}", kv.getKey().toStringNoTruncate(),
- ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
- try (Scanner s = ReplicationTable.getScanner(connMaster)) {
- for (Entry<Key,Value> kv : s) {
- log.info(kv.getKey().toStringNoTruncate() + " "
- + ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
++ try (var scanner = ReplicationTable.getScanner(clientMaster)) {
++ for (Entry<Key,Value> kv : scanner) {
++ log.info("{} {}", kv.getKey().toStringNoTruncate(),
++ ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
+ }
}
- Future<Boolean> future = executor.submit(new Callable<Boolean>() {
-
- @Override
- public Boolean call() throws Exception {
- connMaster.replicationOperations().drain(masterTable, filesNeedingReplication);
- log.info("Drain completed");
- return true;
- }
-
+ Future<Boolean> future = executor.submit(() -> {
+ clientMaster.replicationOperations().drain(masterTable, filesNeedingReplication);
+ log.info("Drain completed");
+ return true;
});
long timeoutSeconds = timeoutFactor * 30;
@@@ -294,42 -299,44 +297,45 @@@
log.info("");
log.info("Fetching metadata records:");
- for (Entry<Key,Value> kv : clientMaster.createScanner(MetadataTable.NAME,
- Authorizations.EMPTY)) {
- if (ReplicationSection.COLF.equals(kv.getKey().getColumnFamily())) {
- log.info("{} {}", kv.getKey().toStringNoTruncate(),
- ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
- } else {
- log.info("{} {}", kv.getKey().toStringNoTruncate(), kv.getValue());
- try (Scanner s = connMaster.createScanner(MetadataTable.NAME, Authorizations.EMPTY)) {
- for (Entry<Key,Value> kv : s) {
++ try (var scanner = clientMaster.createScanner(MetadataTable.NAME, Authorizations.EMPTY)) {
++ for (Entry<Key,Value> kv : scanner) {
+ if (ReplicationSection.COLF.equals(kv.getKey().getColumnFamily())) {
- log.info(kv.getKey().toStringNoTruncate() + " "
- + ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
++ log.info("{} {}", kv.getKey().toStringNoTruncate(),
++ ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
+ } else {
- log.info(kv.getKey().toStringNoTruncate() + " " + kv.getValue());
++ log.info("{} {}", kv.getKey().toStringNoTruncate(), kv.getValue());
+ }
}
}
log.info("");
log.info("Fetching replication records:");
- for (Entry<Key,Value> kv : ReplicationTable.getScanner(clientMaster)) {
- log.info("{} {}", kv.getKey().toStringNoTruncate(),
- ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
- try (Scanner s = ReplicationTable.getScanner(connMaster)) {
- for (Entry<Key,Value> kv : s) {
- log.info(kv.getKey().toStringNoTruncate() + " "
- + ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
++ try (var scanner = ReplicationTable.getScanner(clientMaster)) {
++ for (Entry<Key,Value> kv : scanner) {
++ log.info("{} {}", kv.getKey().toStringNoTruncate(),
++ ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
+ }
}
- Scanner master = connMaster.createScanner(masterTable, Authorizations.EMPTY),
- peer = connPeer.createScanner(peerTable, Authorizations.EMPTY);
- Iterator<Entry<Key,Value>> masterIter = master.iterator(), peerIter = peer.iterator();
- Entry<Key,Value> masterEntry = null, peerEntry = null;
- while (masterIter.hasNext() && peerIter.hasNext()) {
- masterEntry = masterIter.next();
- peerEntry = peerIter.next();
- assertEquals(masterEntry.getKey() + " was not equal to " + peerEntry.getKey(), 0,
- masterEntry.getKey().compareTo(peerEntry.getKey(),
- PartialKey.ROW_COLFAM_COLQUAL_COLVIS));
- assertEquals(masterEntry.getValue(), peerEntry.getValue());
- }
+ try (Scanner master = clientMaster.createScanner(masterTable, Authorizations.EMPTY);
+ Scanner peer = clientPeer.createScanner(peerTable, Authorizations.EMPTY)) {
+ Iterator<Entry<Key,Value>> masterIter = master.iterator(), peerIter = peer.iterator();
+ Entry<Key,Value> masterEntry = null, peerEntry = null;
+ while (masterIter.hasNext() && peerIter.hasNext()) {
+ masterEntry = masterIter.next();
+ peerEntry = peerIter.next();
+ assertEquals(masterEntry.getKey() + " was not equal to " + peerEntry.getKey(), 0,
+ masterEntry.getKey().compareTo(peerEntry.getKey(),
+ PartialKey.ROW_COLFAM_COLQUAL_COLVIS));
+ assertEquals(masterEntry.getValue(), peerEntry.getValue());
+ }
- log.info("Last master entry: " + masterEntry);
- log.info("Last peer entry: " + peerEntry);
+ log.info("Last master entry: {}", masterEntry);
+ log.info("Last peer entry: {}", peerEntry);
- assertFalse("Had more data to read from the master", masterIter.hasNext());
- assertFalse("Had more data to read from the peer", peerIter.hasNext());
+ assertFalse("Had more data to read from the master", masterIter.hasNext());
+ assertFalse("Had more data to read from the peer", peerIter.hasNext());
+ }
} finally {
peerCluster.stop();
}
@@@ -521,105 -529,103 +527,106 @@@
peerCluster.start();
- Connector connMaster = getConnector();
- Connector connPeer = peerCluster.getConnector("root", new PasswordToken(ROOT_PASSWORD));
-
- String peerUserName = "repl";
- String peerPassword = "passwd";
-
- // Create a user on the peer for replication to use
- connPeer.securityOperations().createLocalUser(peerUserName, new PasswordToken(peerPassword));
-
- String peerClusterName = "peer";
-
- // ...peer = AccumuloReplicaSystem,instanceName,zookeepers
- connMaster.instanceOperations().setProperty(
- Property.REPLICATION_PEERS.getKey() + peerClusterName,
- ReplicaSystemFactory.getPeerConfigurationValue(AccumuloReplicaSystem.class,
- AccumuloReplicaSystem.buildConfiguration(peerCluster.getInstanceName(),
- peerCluster.getZooKeepers())));
-
- // Configure the credentials we should use to authenticate ourselves to the peer for replication
- connMaster.instanceOperations()
- .setProperty(Property.REPLICATION_PEER_USER.getKey() + peerClusterName, peerUserName);
- connMaster.instanceOperations()
- .setProperty(Property.REPLICATION_PEER_PASSWORD.getKey() + peerClusterName, peerPassword);
-
- String masterTable = "master", peerTable = "peer";
-
- connMaster.tableOperations().create(masterTable);
- String masterTableId = connMaster.tableOperations().tableIdMap().get(masterTable);
- assertNotNull(masterTableId);
-
- connPeer.tableOperations().create(peerTable);
- String peerTableId = connPeer.tableOperations().tableIdMap().get(peerTable);
- assertNotNull(peerTableId);
-
- // Give our replication user the ability to write to the table
- connPeer.securityOperations().grantTablePermission(peerUserName, peerTable,
- TablePermission.WRITE);
-
- // Replicate this table to the peerClusterName in a table with the peerTableId table id
- connMaster.tableOperations().setProperty(masterTable, Property.TABLE_REPLICATION.getKey(),
- "true");
- connMaster.tableOperations().setProperty(masterTable,
- Property.TABLE_REPLICATION_TARGET.getKey() + peerClusterName, peerTableId);
-
- // Write some data to table1
- BatchWriter bw = connMaster.createBatchWriter(masterTable, new BatchWriterConfig());
- for (int rows = 0; rows < 5000; rows++) {
- Mutation m = new Mutation(Integer.toString(rows));
- for (int cols = 0; cols < 100; cols++) {
- String value = Integer.toString(cols);
- m.put(value, "", value);
- }
- bw.addMutation(m);
- }
+ try (AccumuloClient clientMaster = Accumulo.newClient().from(getClientProperties()).build();
+ AccumuloClient clientPeer =
+ peerCluster.createAccumuloClient("root", new PasswordToken(ROOT_PASSWORD))) {
- bw.close();
+ String peerUserName = "repl";
+ String peerPassword = "passwd";
- log.info("Wrote all data to master cluster");
+ // Create a user on the peer for replication to use
+ clientPeer.securityOperations().createLocalUser(peerUserName,
+ new PasswordToken(peerPassword));
- Set<String> files = connMaster.replicationOperations().referencedFiles(masterTable);
- for (String s : files) {
- log.info("Found referenced file for " + masterTable + ": " + s);
- }
+ String peerClusterName = "peer";
- for (ProcessReference proc : cluster.getProcesses().get(ServerType.TABLET_SERVER)) {
- cluster.killProcess(ServerType.TABLET_SERVER, proc);
- }
+ // ...peer = AccumuloReplicaSystem,instanceName,zookeepers
+ clientMaster.instanceOperations().setProperty(
+ Property.REPLICATION_PEERS.getKey() + peerClusterName,
+ ReplicaSystemFactory.getPeerConfigurationValue(AccumuloReplicaSystem.class,
+ AccumuloReplicaSystem.buildConfiguration(peerCluster.getInstanceName(),
+ peerCluster.getZooKeepers())));
+
+ // Configure the credentials we should use to authenticate ourselves to the peer for
+ // replication
+ clientMaster.instanceOperations()
+ .setProperty(Property.REPLICATION_PEER_USER.getKey() + peerClusterName, peerUserName);
+ clientMaster.instanceOperations()
+ .setProperty(Property.REPLICATION_PEER_PASSWORD.getKey() + peerClusterName, peerPassword);
+
+ String masterTable = "master", peerTable = "peer";
+
+ clientPeer.tableOperations().create(peerTable);
+ String peerTableId = clientPeer.tableOperations().tableIdMap().get(peerTable);
+ assertNotNull(peerTableId);
+
+ Map<String,String> props = new HashMap<>();
+ props.put(Property.TABLE_REPLICATION.getKey(), "true");
+ props.put(Property.TABLE_REPLICATION_TARGET.getKey() + peerClusterName, peerTableId);
- cluster.exec(TabletServer.class);
+ clientMaster.tableOperations().create(masterTable,
+ new NewTableConfiguration().setProperties(props));
+ String masterTableId = clientMaster.tableOperations().tableIdMap().get(masterTable);
+ assertNotNull(masterTableId);
- Iterators.size(connMaster.createScanner(masterTable, Authorizations.EMPTY).iterator());
+ // Give our replication user the ability to write to the table
+ clientPeer.securityOperations().grantTablePermission(peerUserName, peerTable,
+ TablePermission.WRITE);
- try (Scanner s = connMaster.createScanner(ReplicationTable.NAME, Authorizations.EMPTY)) {
- for (Entry<Key,Value> kv : s) {
- log.debug(kv.getKey().toStringNoTruncate() + " "
- + ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
+ // Write some data to table1
+ try (BatchWriter bw = clientMaster.createBatchWriter(masterTable)) {
+ for (int rows = 0; rows < 5000; rows++) {
+ Mutation m = new Mutation(Integer.toString(rows));
+ for (int cols = 0; cols < 100; cols++) {
+ String value = Integer.toString(cols);
+ m.put(value, "", value);
+ }
+ bw.addMutation(m);
+ }
}
- }
- connMaster.replicationOperations().drain(masterTable, files);
-
- Scanner master = connMaster.createScanner(masterTable, Authorizations.EMPTY),
- peer = connPeer.createScanner(peerTable, Authorizations.EMPTY);
- Iterator<Entry<Key,Value>> masterIter = master.iterator(), peerIter = peer.iterator();
- assertTrue("No data in master table", masterIter.hasNext());
- assertTrue("No data in peer table", peerIter.hasNext());
- while (masterIter.hasNext() && peerIter.hasNext()) {
- Entry<Key,Value> masterEntry = masterIter.next(), peerEntry = peerIter.next();
- assertEquals(peerEntry.getKey() + " was not equal to " + peerEntry.getKey(), 0,
- masterEntry.getKey().compareTo(peerEntry.getKey(), PartialKey.ROW_COLFAM_COLQUAL_COLVIS));
- assertEquals(masterEntry.getValue(), peerEntry.getValue());
- }
+ log.info("Wrote all data to master cluster");
+
+ Set<String> files = clientMaster.replicationOperations().referencedFiles(masterTable);
+ for (String s : files) {
+ log.info("Found referenced file for {}: {}", masterTable, s);
+ }
- assertFalse("Had more data to read from the master", masterIter.hasNext());
- assertFalse("Had more data to read from the peer", peerIter.hasNext());
+ for (ProcessReference proc : cluster.getProcesses().get(ServerType.TABLET_SERVER)) {
+ cluster.killProcess(ServerType.TABLET_SERVER, proc);
+ }
- peerCluster.stop();
+ cluster.exec(TabletServer.class);
+
+ Iterators.size(clientMaster.createScanner(masterTable, Authorizations.EMPTY).iterator());
+
- for (Entry<Key,Value> kv : clientMaster.createScanner(ReplicationTable.NAME,
- Authorizations.EMPTY)) {
- log.debug("{} {}", kv.getKey().toStringNoTruncate(),
- ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
++ try (var scanner = clientMaster.createScanner(ReplicationTable.NAME, Authorizations.EMPTY)) {
++ for (Entry<Key,Value> kv : scanner) {
++ log.debug("{} {}", kv.getKey().toStringNoTruncate(),
++ ProtobufUtil.toString(Status.parseFrom(kv.getValue().get())));
++ }
+ }
+
+ clientMaster.replicationOperations().drain(masterTable, files);
+
+ try (Scanner master = clientMaster.createScanner(masterTable, Authorizations.EMPTY);
+ Scanner peer = clientPeer.createScanner(peerTable, Authorizations.EMPTY)) {
+ Iterator<Entry<Key,Value>> masterIter = master.iterator(), peerIter = peer.iterator();
+ assertTrue("No data in master table", masterIter.hasNext());
+ assertTrue("No data in peer table", peerIter.hasNext());
+ while (masterIter.hasNext() && peerIter.hasNext()) {
+ Entry<Key,Value> masterEntry = masterIter.next(), peerEntry = peerIter.next();
+ assertEquals(peerEntry.getKey() + " was not equal to " + peerEntry.getKey(), 0,
+ masterEntry.getKey().compareTo(peerEntry.getKey(),
+ PartialKey.ROW_COLFAM_COLQUAL_COLVIS));
+ assertEquals(masterEntry.getValue(), peerEntry.getValue());
+ }
+
+ assertFalse("Had more data to read from the master", masterIter.hasNext());
+ assertFalse("Had more data to read from the peer", peerIter.hasNext());
+ }
+ peerCluster.stop();
+ }
}
@Test