accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From e..@apache.org
Subject [03/43] accumulo git commit: ACCUMULO-3871 move ITs into distro jar, stop building test jar
Date Thu, 04 Jun 2015 18:52:44 GMT
http://git-wip-us.apache.org/repos/asf/accumulo/blob/01ae5b85/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
deleted file mode 100644
index 77198df..0000000
--- a/test/src/test/java/org/apache/accumulo/test/replication/ReplicationIT.java
+++ /dev/null
@@ -1,1436 +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.test.replication;
-
-import static java.nio.charset.StandardCharsets.UTF_8;
-
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.util.ArrayList;
-import java.util.Arrays;
-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.UUID;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import org.apache.accumulo.core.Constants;
-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.Instance;
-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.TableOfflineException;
-import org.apache.accumulo.core.client.ZooKeeperInstance;
-import org.apache.accumulo.core.client.admin.TableOperations;
-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.data.impl.KeyExtent;
-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;
-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.ReplicationSchema.WorkSection;
-import org.apache.accumulo.core.replication.ReplicationTable;
-import org.apache.accumulo.core.replication.ReplicationTarget;
-import org.apache.accumulo.core.security.Authorizations;
-import org.apache.accumulo.core.security.TablePermission;
-import org.apache.accumulo.core.tabletserver.log.LogEntry;
-import org.apache.accumulo.core.util.Pair;
-import org.apache.accumulo.core.util.UtilWaitThread;
-import org.apache.accumulo.core.zookeeper.ZooUtil;
-import org.apache.accumulo.fate.zookeeper.ZooCache;
-import org.apache.accumulo.fate.zookeeper.ZooCacheFactory;
-import org.apache.accumulo.fate.zookeeper.ZooLock;
-import org.apache.accumulo.gc.SimpleGarbageCollector;
-import org.apache.accumulo.minicluster.ServerType;
-import org.apache.accumulo.minicluster.impl.MiniAccumuloConfigImpl;
-import org.apache.accumulo.server.log.WalStateManager;
-import org.apache.accumulo.server.log.WalStateManager.WalState;
-import org.apache.accumulo.server.master.state.TServerInstance;
-import org.apache.accumulo.server.replication.ReplicaSystemFactory;
-import org.apache.accumulo.server.replication.StatusCombiner;
-import org.apache.accumulo.server.replication.StatusFormatter;
-import org.apache.accumulo.server.replication.StatusUtil;
-import org.apache.accumulo.server.replication.proto.Replication.Status;
-import org.apache.accumulo.server.util.ReplicationTableUtil;
-import org.apache.accumulo.server.zookeeper.ZooReaderWriter;
-import org.apache.accumulo.test.functional.ConfigurableMacBase;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.RawLocalFileSystem;
-import org.apache.hadoop.io.Text;
-import org.junit.Assert;
-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.Iterators;
-import com.google.common.collect.Multimap;
-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.
- */
-public class ReplicationIT extends ConfigurableMacBase {
-  private static final Logger log = LoggerFactory.getLogger(ReplicationIT.class);
-
-  @Override
-  public int defaultTimeoutSeconds() {
-    return 60 * 10;
-  }
-
-  @Override
-  public void configure(MiniAccumuloConfigImpl cfg, Configuration hadoopCoreSite) {
-    // Run the master replication loop run frequently
-    cfg.setProperty(Property.INSTANCE_ZK_TIMEOUT, "10s");
-    cfg.setProperty(Property.MASTER_REPLICATION_SCAN_INTERVAL, "1s");
-    cfg.setProperty(Property.REPLICATION_WORK_ASSIGNMENT_SLEEP, "1s");
-    cfg.setProperty(Property.TSERV_WALOG_MAX_SIZE, "1M");
-    cfg.setProperty(Property.GC_CYCLE_START, "1s");
-    cfg.setProperty(Property.GC_CYCLE_DELAY, "0");
-    cfg.setProperty(Property.REPLICATION_NAME, "master");
-    cfg.setProperty(Property.REPLICATION_WORK_PROCESSOR_DELAY, "1s");
-    cfg.setProperty(Property.REPLICATION_WORK_PROCESSOR_PERIOD, "1s");
-    cfg.setProperty(Property.TSERV_TOTAL_MUTATION_QUEUE_MAX, "1M");
-    cfg.setProperty(Property.INSTANCE_ZK_TIMEOUT, "5s");
-    cfg.setNumTservers(1);
-    hadoopCoreSite.set("fs.file.impl", RawLocalFileSystem.class.getName());
-  }
-
-  private Multimap<String,String> getLogs(Connector conn) throws Exception {
-    // Map of server to tableId
-    Multimap<TServerInstance,String> serverToTableID = HashMultimap.create();
-    Scanner scanner = conn.createScanner(MetadataTable.NAME, Authorizations.EMPTY);
-    scanner.setRange(MetadataSchema.TabletsSection.getRange());
-    scanner.fetchColumnFamily(MetadataSchema.TabletsSection.CurrentLocationColumnFamily.NAME);
-    for (Entry<Key,Value> entry : scanner) {
-      TServerInstance key = new TServerInstance(entry.getValue(), entry.getKey().getColumnQualifier());
-      byte[] tableId = KeyExtent.tableOfMetadataRow(entry.getKey().getRow());
-      serverToTableID.put(key, new String(tableId, UTF_8));
-    }
-    // Map of logs to tableId
-    Multimap<String,String> logs = HashMultimap.create();
-    Instance i = conn.getInstance();
-    ZooReaderWriter zk = new ZooReaderWriter(i.getZooKeepers(), i.getZooKeepersSessionTimeOut(), "");
-    WalStateManager wals = new WalStateManager(conn.getInstance(), zk);
-    for (Entry<TServerInstance,List<UUID>> entry : wals.getAllMarkers().entrySet()) {
-      for (UUID id : entry.getValue()) {
-        Pair<WalState,Path> state = wals.state(entry.getKey(), id);
-        for (String tableId : serverToTableID.get(entry.getKey())) {
-          logs.put(state.getSecond().toString(), tableId);
-        }
-      }
-    }
-    return logs;
-  }
-
-  private Multimap<String,String> getAllLogs(Connector conn) throws Exception {
-    Multimap<String,String> logs = getLogs(conn);
-    try {
-      Scanner scanner = conn.createScanner(ReplicationTable.NAME, Authorizations.EMPTY);
-      StatusSection.limit(scanner);
-      Text buff = new Text();
-      for (Entry<Key,Value> entry : scanner) {
-        if (Thread.interrupted()) {
-          Thread.currentThread().interrupt();
-          return logs;
-        }
-
-        StatusSection.getFile(entry.getKey(), buff);
-        String file = buff.toString();
-        StatusSection.getTableId(entry.getKey(), buff);
-        String tableId = buff.toString();
-
-        logs.put(file, tableId);
-      }
-    } catch (TableOfflineException e) {
-      log.debug("Replication table isn't online yet");
-    }
-    return logs;
-  }
-
-  private void waitForGCLock(Connector conn) throws InterruptedException {
-    // Check if the GC process has the lock before wasting our retry attempts
-    ZooKeeperInstance zki = (ZooKeeperInstance) conn.getInstance();
-    ZooCacheFactory zcf = new ZooCacheFactory();
-    ZooCache zcache = zcf.getZooCache(zki.getZooKeepers(), zki.getZooKeepersSessionTimeOut());
-    String zkPath = ZooUtil.getRoot(conn.getInstance()) + Constants.ZGC_LOCK;
-    log.info("Looking for GC lock at {}", zkPath);
-    byte[] data = ZooLock.getLockData(zcache, zkPath, null);
-    while (null == data) {
-      log.info("Waiting for GC ZooKeeper lock to be acquired");
-      Thread.sleep(1000);
-      data = ZooLock.getLockData(zcache, zkPath, null);
-    }
-  }
-
-  @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";
-    conn.tableOperations().create(table);
-    // If we have more than one tserver, this is subject to a race condition.
-    conn.tableOperations().setProperty(table, Property.TABLE_REPLICATION.getKey(), "true");
-
-    BatchWriter bw = conn.createBatchWriter(table, new BatchWriterConfig());
-    for (int i = 0; i < 10; i++) {
-      Mutation m = new Mutation(Integer.toString(i));
-      m.put(new byte[0], new byte[0], new byte[0]);
-      bw.addMutation(m);
-    }
-
-    bw.close();
-
-    // After writing data, we'll get a replication table online
-    boolean online = ReplicationTable.isOnline(conn);
-    int attempts = 10;
-    do {
-      if (!online) {
-        UtilWaitThread.sleep(2000);
-        online = ReplicationTable.isOnline(conn);
-        attempts--;
-      }
-    } 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)) {
-        break;
-      }
-      log.info("Could not read replication table, waiting and will retry");
-      Thread.sleep(2000);
-    }
-
-    Assert.assertTrue("'root' user could not read the replication table",
-        conn.securityOperations().hasTablePermission("root", ReplicationTable.NAME, TablePermission.READ));
-
-    Set<String> replRows = Sets.newHashSet();
-    Scanner scanner;
-    attempts = 5;
-    while (replRows.isEmpty() && attempts > 0) {
-      scanner = ReplicationTable.getScanner(conn);
-      StatusSection.limit(scanner);
-      for (Entry<Key,Value> entry : scanner) {
-        Key k = entry.getKey();
-
-        String fileUri = k.getRow().toString();
-        try {
-          new URI(fileUri);
-        } catch (URISyntaxException e) {
-          Assert.fail("Expected a valid URI: " + fileUri);
-        }
-
-        replRows.add(fileUri);
-      }
-    }
-
-    Set<String> wals = Sets.newHashSet();
-    attempts = 5;
-    Instance i = conn.getInstance();
-    ZooReaderWriter zk = new ZooReaderWriter(i.getZooKeepers(), i.getZooKeepersSessionTimeOut(), "");
-    while (wals.isEmpty() && attempts > 0) {
-      WalStateManager markers = new WalStateManager(i, zk);
-      for (Entry<Path,WalState> entry : markers.getAllState().entrySet()) {
-        wals.add(entry.getKey().toString());
-      }
-      attempts--;
-    }
-
-    // We only have one file that should need replication (no trace table)
-    // We should find an entry in tablet and in the repl row
-    Assert.assertEquals("Rows found: " + replRows, 1, replRows.size());
-
-    // There should only be one extra WALog that replication doesn't know about
-    replRows.removeAll(wals);
-    Assert.assertEquals(2, wals.size());
-    Assert.assertEquals(0, replRows.size());
-  }
-
-  @Test
-  public void noRecordsWithoutReplication() throws Exception {
-    Connector conn = getConnector();
-    List<String> tables = new ArrayList<>();
-
-    // replication shouldn't be online when we begin
-    Assert.assertFalse(ReplicationTable.isOnline(conn));
-
-    for (int i = 0; i < 5; i++) {
-      String name = "table" + i;
-      tables.add(name);
-      conn.tableOperations().create(name);
-    }
-
-    // nor after we create some tables (that aren't being replicated)
-    Assert.assertFalse(ReplicationTable.isOnline(conn));
-
-    for (String table : tables) {
-      writeSomeData(conn, table, 5, 5);
-    }
-
-    // After writing data, still no replication table
-    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(ReplicationTable.isOnline(conn));
-
-    for (String table : tables) {
-      conn.tableOperations().delete(table);
-    }
-
-    // After deleting tables, still no replication table
-    Assert.assertFalse(ReplicationTable.isOnline(conn));
-  }
-
-  @Test
-  public void twoEntriesForTwoTables() throws Exception {
-    Connector conn = getConnector();
-    String table1 = "table1", table2 = "table2";
-
-    // replication shouldn't exist when we begin
-    Assert.assertFalse("Replication table already online at the beginning of the test", ReplicationTable.isOnline(conn));
-
-    // Create two tables
-    conn.tableOperations().create(table1);
-    conn.tableOperations().create(table2);
-
-    // 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 bring it online)
-    Assert.assertFalse(ReplicationTable.isOnline(conn));
-
-    // Write some data to table1
-    writeSomeData(conn, table1, 50, 50);
-
-    // 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 online
-    boolean online = ReplicationTable.isOnline(conn);
-    int attempts = 10;
-    do {
-      if (!online) {
-        UtilWaitThread.sleep(5000);
-        online = ReplicationTable.isOnline(conn);
-        attempts--;
-      }
-    } while (!online && attempts > 0);
-    Assert.assertTrue("Replication table did not exist", online);
-
-    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
-    Scanner s = ReplicationTable.getScanner(conn);
-    StatusSection.limit(s);
-    Iterator<Entry<Key,Value>> iter = s.iterator();
-    attempts = 5;
-    while (attempts > 0) {
-      if (!iter.hasNext()) {
-        s.close();
-        Thread.sleep(1000);
-        s = ReplicationTable.getScanner(conn);
-        iter = s.iterator();
-        attempts--;
-      } else {
-        break;
-      }
-    }
-    Assert.assertTrue(iter.hasNext());
-    Entry<Key,Value> entry = iter.next();
-    // We should at least find one status record for this table, we might find a second if another log was started from ingesting the data
-    Assert.assertEquals("Expected to find replication entry for " + table1, conn.tableOperations().tableIdMap().get(table1), entry.getKey()
-        .getColumnQualifier().toString());
-    s.close();
-
-    // Enable replication on table2
-    conn.tableOperations().setProperty(table2, Property.TABLE_REPLICATION.getKey(), "true");
-
-    // Write some data to table2
-    writeSomeData(conn, table2, 50, 50);
-
-    // 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 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));
-    Set<String> tableIdsForMetadata = Sets.newHashSet(tableIds);
-
-    // Wait to make sure the table permission propagate
-    Thread.sleep(5000);
-
-    s = conn.createScanner(MetadataTable.NAME, Authorizations.EMPTY);
-    s.setRange(MetadataSchema.ReplicationSection.getRange());
-
-    List<Entry<Key,Value>> records = new ArrayList<>();
-    for (Entry<Key,Value> metadata : s) {
-      records.add(metadata);
-    }
-
-    Assert.assertEquals("Expected to find 2 records, but actually found " + records, 2, records.size());
-
-    for (Entry<Key,Value> metadata : records) {
-      Assert.assertTrue("Expected record to be in metadata but wasn't " + metadata.getKey().toStringNoTruncate() + ", tableIds remaining "
-          + tableIdsForMetadata, tableIdsForMetadata.remove(metadata.getKey().getColumnQualifier().toString()));
-    }
-
-    Assert.assertTrue("Expected that we had removed all metadata entries " + tableIdsForMetadata, tableIdsForMetadata.isEmpty());
-
-    // Should be creating these records in replication table from metadata table every second
-    Thread.sleep(5000);
-
-    // Verify that we found two replication records: one for table1 and one for table2
-    s = ReplicationTable.getScanner(conn);
-    StatusSection.limit(s);
-    iter = s.iterator();
-    Assert.assertTrue("Found no records in replication table", iter.hasNext());
-    entry = iter.next();
-    Assert.assertTrue("Expected to find element in replication table", tableIds.remove(entry.getKey().getColumnQualifier().toString()));
-    Assert.assertTrue("Expected to find two elements in replication table, only found one ", iter.hasNext());
-    entry = iter.next();
-    Assert.assertTrue("Expected to find element in replication table", tableIds.remove(entry.getKey().getColumnQualifier().toString()));
-    Assert.assertFalse("Expected to only find two elements in replication table", iter.hasNext());
-  }
-
-  private void writeSomeData(Connector conn, String table, int rows, int cols) throws Exception {
-    BatchWriter bw = conn.createBatchWriter(table, new BatchWriterConfig());
-    for (int row = 0; row < rows; row++) {
-      Mutation m = new Mutation(Integer.toString(row));
-      for (int col = 0; col < cols; col++) {
-        String value = Integer.toString(col);
-        m.put(value, "", value);
-      }
-      bw.addMutation(m);
-    }
-    bw.close();
-  }
-
-  @Test
-  public void replicationEntriesPrecludeWalDeletion() throws Exception {
-    final Connector conn = getConnector();
-    String table1 = "table1", table2 = "table2", table3 = "table3";
-    final Multimap<String,String> logs = HashMultimap.create();
-    final AtomicBoolean keepRunning = new AtomicBoolean(true);
-
-    Thread t = new Thread(new Runnable() {
-      @Override
-      public void run() {
-        // Should really be able to interrupt here, but the Scanner throws a fit to the logger
-        // when that happens
-        while (keepRunning.get()) {
-          try {
-            logs.putAll(getAllLogs(conn));
-          } catch (Exception e) {
-            log.error("Error getting logs", e);
-          }
-        }
-      }
-
-    });
-
-    t.start();
-
-    conn.tableOperations().create(table1);
-    conn.tableOperations().setProperty(table1, Property.TABLE_REPLICATION.getKey(), "true");
-    conn.tableOperations().setProperty(table1, Property.TABLE_REPLICATION_TARGET.getKey() + "cluster1", "1");
-    Thread.sleep(2000);
-
-    // Write some data to table1
-    writeSomeData(conn, table1, 200, 500);
-
-    conn.tableOperations().create(table2);
-    conn.tableOperations().setProperty(table2, Property.TABLE_REPLICATION.getKey(), "true");
-    conn.tableOperations().setProperty(table2, Property.TABLE_REPLICATION_TARGET.getKey() + "cluster1", "1");
-    Thread.sleep(2000);
-
-    writeSomeData(conn, table2, 200, 500);
-
-    conn.tableOperations().create(table3);
-    conn.tableOperations().setProperty(table3, Property.TABLE_REPLICATION.getKey(), "true");
-    conn.tableOperations().setProperty(table3, Property.TABLE_REPLICATION_TARGET.getKey() + "cluster1", "1");
-    Thread.sleep(2000);
-
-    writeSomeData(conn, table3, 200, 500);
-
-    // Force a write to metadata for the data written
-    for (String table : Arrays.asList(table1, table2, table3)) {
-      conn.tableOperations().flush(table, null, null, true);
-    }
-
-    keepRunning.set(false);
-    t.join(5000);
-
-    // The master is only running every second to create records in the replication table from the metadata table
-    // Sleep a sufficient amount of time to ensure that we get the straggling WALs that might have been created at the end
-    Thread.sleep(5000);
-
-    Scanner s = ReplicationTable.getScanner(conn);
-    StatusSection.limit(s);
-    Set<String> replFiles = new HashSet<>();
-    for (Entry<Key,Value> entry : s) {
-      replFiles.add(entry.getKey().getRow().toString());
-    }
-
-    // We might have a WAL that was use solely for the replication table
-    // We want to remove that from our list as it should not appear in the replication table
-    String replicationTableId = conn.tableOperations().tableIdMap().get(ReplicationTable.NAME);
-    Iterator<Entry<String,String>> observedLogs = logs.entries().iterator();
-    while (observedLogs.hasNext()) {
-      Entry<String,String> observedLog = observedLogs.next();
-      if (replicationTableId.equals(observedLog.getValue())) {
-        log.info("Removing {} because its tableId is for the replication table", observedLog);
-        observedLogs.remove();
-      }
-    }
-
-    // We should have *some* reference to each log that was seen in the metadata table
-    // They might not yet all be closed though (might be newfile)
-    Assert.assertTrue("Metadata log distribution: " + logs + "replFiles " + replFiles, logs.keySet().containsAll(replFiles));
-    Assert.assertTrue("Difference between replication entries and current logs is bigger than one", logs.keySet().size() - replFiles.size() <= 1);
-
-    final Configuration conf = new Configuration();
-    for (String replFile : replFiles) {
-      Path p = new Path(replFile);
-      FileSystem fs = p.getFileSystem(conf);
-      Assert.assertTrue("File does not exist anymore, it was likely incorrectly garbage collected: " + p, fs.exists(p));
-    }
-  }
-
-  @Test
-  public void combinerWorksOnMetadata() throws Exception {
-    Connector conn = getConnector();
-
-    conn.securityOperations().grantTablePermission("root", MetadataTable.NAME, TablePermission.WRITE);
-
-    ReplicationTableUtil.configureMetadataTable(conn, MetadataTable.NAME);
-
-    Status stat1 = StatusUtil.fileCreated(100);
-    Status stat2 = StatusUtil.fileClosed();
-
-    BatchWriter bw = conn.createBatchWriter(MetadataTable.NAME, new BatchWriterConfig());
-    Mutation m = new Mutation(ReplicationSection.getRowPrefix() + "file:/accumulo/wals/tserver+port/uuid");
-    m.put(ReplicationSection.COLF, new Text("1"), ProtobufUtil.toValue(stat1));
-    bw.addMutation(m);
-    bw.close();
-
-    Scanner s = conn.createScanner(MetadataTable.NAME, Authorizations.EMPTY);
-    s.setRange(ReplicationSection.getRange());
-
-    Status actual = Status.parseFrom(Iterables.getOnlyElement(s).getValue().get());
-    Assert.assertEquals(stat1, actual);
-
-    bw = conn.createBatchWriter(MetadataTable.NAME, new BatchWriterConfig());
-    m = new Mutation(ReplicationSection.getRowPrefix() + "file:/accumulo/wals/tserver+port/uuid");
-    m.put(ReplicationSection.COLF, new Text("1"), ProtobufUtil.toValue(stat2));
-    bw.addMutation(m);
-    bw.close();
-
-    s = conn.createScanner(MetadataTable.NAME, Authorizations.EMPTY);
-    s.setRange(ReplicationSection.getRange());
-
-    actual = Status.parseFrom(Iterables.getOnlyElement(s).getValue().get());
-    Status expected = Status.newBuilder().setBegin(0).setEnd(0).setClosed(true).setInfiniteEnd(true).setCreatedTime(100).build();
-
-    Assert.assertEquals(expected, actual);
-  }
-
-  @Test
-  public void noDeadlock() throws Exception {
-    final Connector conn = getConnector();
-
-    ReplicationTable.setOnline(conn);
-    conn.securityOperations().grantTablePermission("root", ReplicationTable.NAME, TablePermission.WRITE);
-    conn.tableOperations().deleteRows(ReplicationTable.NAME, null, null);
-
-    String table1 = "table1", table2 = "table2", table3 = "table3";
-    conn.tableOperations().create(table1);
-    conn.tableOperations().setProperty(table1, Property.TABLE_REPLICATION.getKey(), "true");
-    conn.tableOperations().setProperty(table1, Property.TABLE_REPLICATION_TARGET.getKey() + "cluster1", "1");
-    conn.tableOperations().create(table2);
-    conn.tableOperations().setProperty(table2, Property.TABLE_REPLICATION.getKey(), "true");
-    conn.tableOperations().setProperty(table2, Property.TABLE_REPLICATION_TARGET.getKey() + "cluster1", "1");
-    conn.tableOperations().create(table3);
-    conn.tableOperations().setProperty(table3, Property.TABLE_REPLICATION.getKey(), "true");
-    conn.tableOperations().setProperty(table3, Property.TABLE_REPLICATION_TARGET.getKey() + "cluster1", "1");
-
-    writeSomeData(conn, table1, 200, 500);
-
-    writeSomeData(conn, table2, 200, 500);
-
-    writeSomeData(conn, table3, 200, 500);
-
-    // Flush everything to try to make the replication records
-    for (String table : Arrays.asList(table1, table2, table3)) {
-      conn.tableOperations().flush(table, null, null, true);
-    }
-
-    // Flush everything to try to make the replication records
-    for (String table : Arrays.asList(table1, table2, table3)) {
-      conn.tableOperations().flush(table, null, null, true);
-    }
-
-    for (String table : Arrays.asList(MetadataTable.NAME, table1, table2, table3)) {
-      Iterators.size(conn.createScanner(table, Authorizations.EMPTY).iterator());
-    }
-  }
-
-  @Test
-  public void filesClosedAfterUnused() throws Exception {
-    Connector conn = getConnector();
-
-    String table = "table";
-    conn.tableOperations().create(table);
-    String tableId = conn.tableOperations().tableIdMap().get(table);
-
-    Assert.assertNotNull(tableId);
-
-    conn.tableOperations().setProperty(table, Property.TABLE_REPLICATION.getKey(), "true");
-    conn.tableOperations().setProperty(table, Property.TABLE_REPLICATION_TARGET.getKey() + "cluster1", "1");
-    // just sleep
-    conn.instanceOperations().setProperty(Property.REPLICATION_PEERS.getKey() + "cluster1",
-        ReplicaSystemFactory.getPeerConfigurationValue(MockReplicaSystem.class, "50000"));
-
-    // Write a mutation to make a log file
-    BatchWriter bw = conn.createBatchWriter(table, new BatchWriterConfig());
-    Mutation m = new Mutation("one");
-    m.put("", "", "");
-    bw.addMutation(m);
-    bw.close();
-
-    // Write another to make sure the logger rolls itself?
-    bw = conn.createBatchWriter(table, new BatchWriterConfig());
-    m = new Mutation("three");
-    m.put("", "", "");
-    bw.addMutation(m);
-    bw.close();
-
-    Scanner s = conn.createScanner(MetadataTable.NAME, Authorizations.EMPTY);
-    s.fetchColumnFamily(TabletsSection.LogColumnFamily.NAME);
-    s.setRange(TabletsSection.getRange(tableId));
-    Set<String> wals = new HashSet<>();
-    for (Entry<Key,Value> entry : s) {
-      LogEntry logEntry = LogEntry.fromKeyValue(entry.getKey(), entry.getValue());
-      wals.add(new Path(logEntry.filename).toString());
-    }
-
-    log.warn("Found wals {}", wals);
-
-    bw = conn.createBatchWriter(table, new BatchWriterConfig());
-    m = new Mutation("three");
-    byte[] bytes = new byte[1024 * 1024];
-    m.put("1".getBytes(), new byte[0], bytes);
-    m.put("2".getBytes(), new byte[0], bytes);
-    m.put("3".getBytes(), new byte[0], bytes);
-    m.put("4".getBytes(), new byte[0], bytes);
-    m.put("5".getBytes(), new byte[0], bytes);
-    bw.addMutation(m);
-    bw.close();
-
-    conn.tableOperations().flush(table, null, null, true);
-
-    while (!ReplicationTable.isOnline(conn)) {
-      UtilWaitThread.sleep(2000);
-    }
-
-    for (int i = 0; i < 10; i++) {
-      s = conn.createScanner(MetadataTable.NAME, Authorizations.EMPTY);
-      s.fetchColumnFamily(LogColumnFamily.NAME);
-      s.setRange(TabletsSection.getRange(tableId));
-      for (Entry<Key,Value> entry : s) {
-        log.info(entry.getKey().toStringNoTruncate() + "=" + entry.getValue());
-      }
-
-      try {
-        s = ReplicationTable.getScanner(conn);
-        StatusSection.limit(s);
-        Text buff = new Text();
-        boolean allReferencedLogsClosed = true;
-        int recordsFound = 0;
-        for (Entry<Key,Value> e : s) {
-          recordsFound++;
-          allReferencedLogsClosed = true;
-          StatusSection.getFile(e.getKey(), buff);
-          String file = buff.toString();
-          if (wals.contains(file)) {
-            Status stat = Status.parseFrom(e.getValue().get());
-            if (!stat.getClosed()) {
-              log.info("{} wasn't closed", file);
-              allReferencedLogsClosed = false;
-            }
-          }
-        }
-
-        if (recordsFound > 0 && allReferencedLogsClosed) {
-          return;
-        }
-        Thread.sleep(2000);
-      } catch (RuntimeException e) {
-        Throwable cause = e.getCause();
-        if (cause instanceof AccumuloSecurityException) {
-          AccumuloSecurityException ase = (AccumuloSecurityException) cause;
-          switch (ase.getSecurityErrorCode()) {
-            case PERMISSION_DENIED:
-              // We tried to read the replication table before the GRANT went through
-              Thread.sleep(2000);
-              break;
-            default:
-              throw e;
-          }
-        }
-      }
-    }
-
-    Assert.fail("We had a file that was referenced but didn't get closed");
-  }
-
-  @Test
-  public void singleTableWithSingleTarget() throws Exception {
-    // We want to kill the GC so it doesn't come along and close Status records and mess up the comparisons
-    // against expected Status messages.
-    getCluster().getClusterControl().stop(ServerType.GARBAGE_COLLECTOR);
-
-    Connector conn = getConnector();
-    String table1 = "table1";
-
-    // replication shouldn't be online when we begin
-    Assert.assertFalse(ReplicationTable.isOnline(conn));
-
-    // Create a table
-    conn.tableOperations().create(table1);
-
-    int attempts = 10;
-
-    // Might think the table doesn't yet exist, retry
-    while (attempts > 0) {
-      try {
-        // Enable replication on table1
-        conn.tableOperations().setProperty(table1, Property.TABLE_REPLICATION.getKey(), "true");
-        // Replicate table1 to cluster1 in the table with id of '4'
-        conn.tableOperations().setProperty(table1, Property.TABLE_REPLICATION_TARGET.getKey() + "cluster1", "4");
-        // Sleep for 100 seconds before saying something is replicated
-        conn.instanceOperations().setProperty(Property.REPLICATION_PEERS.getKey() + "cluster1",
-            ReplicaSystemFactory.getPeerConfigurationValue(MockReplicaSystem.class, "100000"));
-        break;
-      } catch (Exception e) {
-        attempts--;
-        if (attempts <= 0) {
-          throw e;
-        }
-        UtilWaitThread.sleep(2000);
-      }
-    }
-
-    // Write some data to table1
-    writeSomeData(conn, table1, 2000, 50);
-
-    // Make sure the replication table is online at this point
-    boolean online = ReplicationTable.isOnline(conn);
-    attempts = 10;
-    do {
-      if (!online) {
-        UtilWaitThread.sleep(2000);
-        online = ReplicationTable.isOnline(conn);
-        attempts--;
-      }
-    } 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++) {
-      UtilWaitThread.sleep(2000);
-    }
-
-    Assert.assertTrue("Combiner was never set on replication table",
-        conn.tableOperations().listIterators(ReplicationTable.NAME).keySet().contains(ReplicationTable.COMBINER_NAME));
-
-    // Trigger the minor compaction, waiting for it to finish.
-    // This should write the entry to metadata that the file has data
-    conn.tableOperations().flush(table1, null, null, true);
-
-    // Make sure that we have one status element, should be a new file
-    Scanner s = ReplicationTable.getScanner(conn);
-    StatusSection.limit(s);
-    Entry<Key,Value> entry = null;
-    Status expectedStatus = StatusUtil.openWithUnknownLength();
-    attempts = 10;
-    // This record will move from new to new with infinite length because of the minc (flush)
-    while (null == entry && attempts > 0) {
-      try {
-        entry = Iterables.getOnlyElement(s);
-        Status actual = Status.parseFrom(entry.getValue().get());
-        if (actual.getInfiniteEnd() != expectedStatus.getInfiniteEnd()) {
-          entry = null;
-          // the master process didn't yet fire and write the new mutation, wait for it to do
-          // so and try to read it again
-          Thread.sleep(1000);
-        }
-      } catch (NoSuchElementException e) {
-        entry = null;
-        Thread.sleep(500);
-      } catch (IllegalArgumentException e) {
-        // saw this contain 2 elements once
-        s = ReplicationTable.getScanner(conn);
-        StatusSection.limit(s);
-        for (Entry<Key,Value> content : s) {
-          log.info(content.getKey().toStringNoTruncate() + " => " + content.getValue());
-        }
-        throw e;
-      } finally {
-        attempts--;
-      }
-    }
-
-    Assert.assertNotNull("Could not find expected entry in replication table", entry);
-    Status actual = Status.parseFrom(entry.getValue().get());
-    Assert.assertTrue("Expected to find a replication entry that is open with infinite length: " + ProtobufUtil.toString(actual),
-        !actual.getClosed() && actual.getInfiniteEnd());
-
-    // Try a couple of times to watch for the work record to be created
-    boolean notFound = true;
-    for (int i = 0; i < 10 && notFound; i++) {
-      s = ReplicationTable.getScanner(conn);
-      WorkSection.limit(s);
-      int elementsFound = Iterables.size(s);
-      if (0 < elementsFound) {
-        Assert.assertEquals(1, elementsFound);
-        notFound = false;
-      }
-      Thread.sleep(500);
-    }
-
-    // If we didn't find the work record, print the contents of the table
-    if (notFound) {
-      s = ReplicationTable.getScanner(conn);
-      for (Entry<Key,Value> content : s) {
-        log.info(content.getKey().toStringNoTruncate() + " => " + content.getValue());
-      }
-      Assert.assertFalse("Did not find the work entry for the status entry", notFound);
-    }
-
-    // Write some more data so that we over-run the single WAL
-    writeSomeData(conn, table1, 3000, 50);
-
-    log.info("Issued compaction for table");
-    conn.tableOperations().compact(table1, null, null, true, true);
-    log.info("Compaction completed");
-
-    // Master is creating entries in the replication table from the metadata table every second.
-    // Compaction should trigger the record to be written to metadata. Wait a bit to ensure
-    // that the master has time to work.
-    Thread.sleep(5000);
-
-    s = ReplicationTable.getScanner(conn);
-    StatusSection.limit(s);
-    int numRecords = 0;
-    for (Entry<Key,Value> e : s) {
-      numRecords++;
-      log.info("Found status record {}\t{}", e.getKey().toStringNoTruncate(), ProtobufUtil.toString(Status.parseFrom(e.getValue().get())));
-    }
-
-    Assert.assertEquals(2, numRecords);
-
-    // We should eventually get 2 work records recorded, need to account for a potential delay though
-    // might see: status1 -> work1 -> status2 -> (our scans) -> work2
-    notFound = true;
-    for (int i = 0; i < 10 && notFound; i++) {
-      s = ReplicationTable.getScanner(conn);
-      WorkSection.limit(s);
-      int elementsFound = Iterables.size(s);
-      if (2 == elementsFound) {
-        notFound = false;
-      }
-      Thread.sleep(500);
-    }
-
-    // If we didn't find the work record, print the contents of the table
-    if (notFound) {
-      s = ReplicationTable.getScanner(conn);
-      for (Entry<Key,Value> content : s) {
-        log.info(content.getKey().toStringNoTruncate() + " => " + content.getValue());
-      }
-      Assert.assertFalse("Did not find the work entries for the status entries", notFound);
-    }
-  }
-
-  @Test
-  public void correctClusterNameInWorkEntry() throws Exception {
-    Connector conn = getConnector();
-    String table1 = "table1";
-
-    // replication shouldn't be online when we begin
-    Assert.assertFalse(ReplicationTable.isOnline(conn));
-
-    // Create two tables
-    conn.tableOperations().create(table1);
-
-    int attempts = 5;
-    while (attempts > 0) {
-      try {
-        // Enable replication on table1
-        conn.tableOperations().setProperty(table1, Property.TABLE_REPLICATION.getKey(), "true");
-        // Replicate table1 to cluster1 in the table with id of '4'
-        conn.tableOperations().setProperty(table1, Property.TABLE_REPLICATION_TARGET.getKey() + "cluster1", "4");
-        attempts = 0;
-      } catch (Exception e) {
-        attempts--;
-        if (attempts <= 0) {
-          throw e;
-        }
-        UtilWaitThread.sleep(500);
-      }
-    }
-
-    // Write some data to table1
-    writeSomeData(conn, table1, 2000, 50);
-    conn.tableOperations().flush(table1, null, null, true);
-
-    String tableId = conn.tableOperations().tableIdMap().get(table1);
-    Assert.assertNotNull("Table ID was null", tableId);
-
-    // Make sure the replication table exists at this point
-    boolean online = ReplicationTable.isOnline(conn);
-    attempts = 5;
-    do {
-      if (!online) {
-        UtilWaitThread.sleep(500);
-        online = ReplicationTable.isOnline(conn);
-        attempts--;
-      }
-    } 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);
-    }
-
-    Assert.assertTrue(conn.securityOperations().hasTablePermission("root", ReplicationTable.NAME, TablePermission.READ));
-
-    boolean notFound = true;
-    Scanner s;
-    for (int i = 0; i < 10 && notFound; i++) {
-      s = ReplicationTable.getScanner(conn);
-      WorkSection.limit(s);
-      try {
-        Entry<Key,Value> e = Iterables.getOnlyElement(s);
-        Text expectedColqual = new ReplicationTarget("cluster1", "4", tableId).toText();
-        Assert.assertEquals(expectedColqual, e.getKey().getColumnQualifier());
-        notFound = false;
-      } catch (NoSuchElementException e) {} catch (IllegalArgumentException e) {
-        s = ReplicationTable.getScanner(conn);
-        for (Entry<Key,Value> content : s) {
-          log.info(content.getKey().toStringNoTruncate() + " => " + content.getValue());
-        }
-        Assert.fail("Found more than one work section entry");
-      }
-
-      Thread.sleep(500);
-    }
-
-    if (notFound) {
-      s = ReplicationTable.getScanner(conn);
-      for (Entry<Key,Value> content : s) {
-        log.info(content.getKey().toStringNoTruncate() + " => " + content.getValue());
-      }
-      Assert.assertFalse("Did not find the work entry for the status entry", notFound);
-    }
-  }
-
-  @Test
-  public void replicationRecordsAreClosedAfterGarbageCollection() throws Exception {
-    getCluster().getClusterControl().stop(ServerType.GARBAGE_COLLECTOR);
-
-    final Connector conn = getConnector();
-
-    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<>();
-
-    Thread t = new Thread(new Runnable() {
-      @Override
-      public void run() {
-        // Should really be able to interrupt here, but the Scanner throws a fit to the logger
-        // when that happens
-        while (keepRunning.get()) {
-          try {
-            metadataWals.addAll(getLogs(conn).keySet());
-          } catch (Exception e) {
-            log.error("Metadata table doesn't exist");
-          }
-        }
-      }
-
-    });
-
-    t.start();
-
-    String table1 = "table1", table2 = "table2", table3 = "table3";
-
-    try {
-      conn.tableOperations().create(table1);
-      conn.tableOperations().setProperty(table1, Property.TABLE_REPLICATION.getKey(), "true");
-      conn.tableOperations().setProperty(table1, Property.TABLE_REPLICATION_TARGET.getKey() + "cluster1", "1");
-      conn.instanceOperations().setProperty(Property.REPLICATION_PEERS.getKey() + "cluster1",
-          ReplicaSystemFactory.getPeerConfigurationValue(MockReplicaSystem.class, null));
-
-      // Write some data to table1
-      writeSomeData(conn, table1, 200, 500);
-
-      conn.tableOperations().create(table2);
-      conn.tableOperations().setProperty(table2, Property.TABLE_REPLICATION.getKey(), "true");
-      conn.tableOperations().setProperty(table2, Property.TABLE_REPLICATION_TARGET.getKey() + "cluster1", "1");
-
-      writeSomeData(conn, table2, 200, 500);
-
-      conn.tableOperations().create(table3);
-      conn.tableOperations().setProperty(table3, Property.TABLE_REPLICATION.getKey(), "true");
-      conn.tableOperations().setProperty(table3, Property.TABLE_REPLICATION_TARGET.getKey() + "cluster1", "1");
-
-      writeSomeData(conn, table3, 200, 500);
-
-      // Flush everything to try to make the replication records
-      for (String table : Arrays.asList(table1, table2, table3)) {
-        conn.tableOperations().compact(table, null, null, true, true);
-      }
-    } finally {
-      keepRunning.set(false);
-      t.join(5000);
-      Assert.assertFalse(t.isAlive());
-    }
-
-    // Kill the tserver(s) and restart them
-    // to ensure that the WALs we previously observed all move to closed.
-    cluster.getClusterControl().stop(ServerType.TABLET_SERVER);
-    cluster.getClusterControl().start(ServerType.TABLET_SERVER);
-
-    // Make sure we can read all the tables (recovery complete)
-    for (String table : Arrays.asList(table1, table2, table3)) {
-      Iterators.size(conn.createScanner(table, Authorizations.EMPTY).iterator());
-    }
-
-    // Starting the gc will run CloseWriteAheadLogReferences which will first close Statuses
-    // in the metadata table, and then in the replication table
-    Process gc = cluster.exec(SimpleGarbageCollector.class);
-
-    waitForGCLock(conn);
-
-    Thread.sleep(1000);
-
-    log.info("GC is up and should have had time to run at least once by now");
-
-    try {
-      boolean allClosed = true;
-
-      // We should either find all closed records or no records
-      // After they're closed, they are candidates for deletion
-      for (int i = 0; i < 10; i++) {
-        Scanner s = conn.createScanner(MetadataTable.NAME, Authorizations.EMPTY);
-        s.setRange(Range.prefix(ReplicationSection.getRowPrefix()));
-        Iterator<Entry<Key,Value>> iter = s.iterator();
-
-        long recordsFound = 0l;
-        while (allClosed && iter.hasNext()) {
-          Entry<Key,Value> entry = iter.next();
-          String wal = entry.getKey().getRow().toString();
-          if (metadataWals.contains(wal)) {
-            Status status = Status.parseFrom(entry.getValue().get());
-            log.info("{}={}", entry.getKey().toStringNoTruncate(), ProtobufUtil.toString(status));
-            allClosed &= status.getClosed();
-            recordsFound++;
-          }
-        }
-
-        log.info("Found {} records from the metadata table", recordsFound);
-        if (allClosed) {
-          break;
-        }
-
-        UtilWaitThread.sleep(2000);
-      }
-
-      if (!allClosed) {
-        Scanner s = conn.createScanner(MetadataTable.NAME, Authorizations.EMPTY);
-        s.setRange(Range.prefix(ReplicationSection.getRowPrefix()));
-        for (Entry<Key,Value> entry : s) {
-          log.info(entry.getKey().toStringNoTruncate() + " " + ProtobufUtil.toString(Status.parseFrom(entry.getValue().get())));
-        }
-        Assert.fail("Expected all replication records in the metadata table to be closed");
-      }
-
-      for (int i = 0; i < 10; i++) {
-        allClosed = true;
-
-        Scanner s = ReplicationTable.getScanner(conn);
-        Iterator<Entry<Key,Value>> iter = s.iterator();
-
-        long recordsFound = 0l;
-        while (allClosed && iter.hasNext()) {
-          Entry<Key,Value> entry = iter.next();
-          String wal = entry.getKey().getRow().toString();
-          if (metadataWals.contains(wal)) {
-            Status status = Status.parseFrom(entry.getValue().get());
-            log.info("{}={}", entry.getKey().toStringNoTruncate(), ProtobufUtil.toString(status));
-            allClosed &= status.getClosed();
-            recordsFound++;
-          }
-        }
-
-        log.info("Found {} records from the replication table", recordsFound);
-        if (allClosed) {
-          break;
-        }
-
-        UtilWaitThread.sleep(3000);
-      }
-
-      if (!allClosed) {
-        Scanner s = ReplicationTable.getScanner(conn);
-        StatusSection.limit(s);
-        for (Entry<Key,Value> entry : s) {
-          log.info(entry.getKey().toStringNoTruncate() + " " + TextFormat.shortDebugString(Status.parseFrom(entry.getValue().get())));
-        }
-        Assert.fail("Expected all replication records in the replication table to be closed");
-      }
-
-    } finally {
-      gc.destroy();
-      gc.waitFor();
-    }
-
-  }
-
-  @Test
-  public void replicatedStatusEntriesAreDeleted() throws Exception {
-    // Just stop it now, we'll restart it after we restart the tserver
-    getCluster().getClusterControl().stop(ServerType.GARBAGE_COLLECTOR);
-
-    final Connector conn = getConnector();
-    log.info("Got connector to MAC");
-    String table1 = "table1";
-
-    // replication shouldn't be online when we begin
-    Assert.assertFalse(ReplicationTable.isOnline(conn));
-
-    // Create two tables
-    conn.tableOperations().create(table1);
-
-    int attempts = 5;
-    while (attempts > 0) {
-      try {
-        // Enable replication on table1
-        conn.tableOperations().setProperty(table1, Property.TABLE_REPLICATION.getKey(), "true");
-        // Replicate table1 to cluster1 in the table with id of '4'
-        conn.tableOperations().setProperty(table1, Property.TABLE_REPLICATION_TARGET.getKey() + "cluster1", "4");
-        // Use the MockReplicaSystem impl and sleep for 5seconds
-        conn.instanceOperations().setProperty(Property.REPLICATION_PEERS.getKey() + "cluster1",
-            ReplicaSystemFactory.getPeerConfigurationValue(MockReplicaSystem.class, "1000"));
-        attempts = 0;
-      } catch (Exception e) {
-        attempts--;
-        if (attempts <= 0) {
-          throw e;
-        }
-        UtilWaitThread.sleep(500);
-      }
-    }
-
-    String tableId = conn.tableOperations().tableIdMap().get(table1);
-    Assert.assertNotNull("Could not determine table id for " + table1, tableId);
-
-    // Write some data to table1
-    writeSomeData(conn, table1, 2000, 50);
-    conn.tableOperations().flush(table1, null, null, true);
-
-    // Make sure the replication table exists at this point
-    boolean online = ReplicationTable.isOnline(conn);
-    attempts = 10;
-    do {
-      if (!online) {
-        UtilWaitThread.sleep(1000);
-        online = ReplicationTable.isOnline(conn);
-        attempts--;
-      }
-    } 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);
-
-    log.info("Checking for replication entries in replication");
-    // Then we need to get those records over to the replication table
-    Scanner s;
-    Set<String> entries = new HashSet<>();
-    for (int i = 0; i < 5; i++) {
-      s = conn.createScanner(MetadataTable.NAME, Authorizations.EMPTY);
-      s.setRange(ReplicationSection.getRange());
-      entries.clear();
-      for (Entry<Key,Value> entry : s) {
-        entries.add(entry.getKey().getRow().toString());
-        log.info("{}={}", entry.getKey().toStringNoTruncate(), entry.getValue());
-      }
-      if (!entries.isEmpty()) {
-        log.info("Replication entries {}", entries);
-        break;
-      }
-      Thread.sleep(1000);
-    }
-
-    Assert.assertFalse("Did not find any replication entries in the replication table", entries.isEmpty());
-
-    // Find the WorkSection record that will be created for that data we ingested
-    boolean notFound = true;
-    for (int i = 0; i < 10 && notFound; i++) {
-      try {
-        s = ReplicationTable.getScanner(conn);
-        WorkSection.limit(s);
-        Entry<Key,Value> e = Iterables.getOnlyElement(s);
-        log.info("Found entry: " + e.getKey().toStringNoTruncate());
-        Text expectedColqual = new ReplicationTarget("cluster1", "4", tableId).toText();
-        Assert.assertEquals(expectedColqual, e.getKey().getColumnQualifier());
-        notFound = false;
-      } catch (NoSuchElementException e) {
-
-      } catch (IllegalArgumentException e) {
-        // Somehow we got more than one element. Log what they were
-        s = ReplicationTable.getScanner(conn);
-        for (Entry<Key,Value> content : s) {
-          log.info(content.getKey().toStringNoTruncate() + " => " + content.getValue());
-        }
-        Assert.fail("Found more than one work section entry");
-      } catch (RuntimeException e) {
-        // Catch a propagation issue, fail if it's not what we expect
-        Throwable cause = e.getCause();
-        if (cause instanceof AccumuloSecurityException) {
-          AccumuloSecurityException sec = (AccumuloSecurityException) cause;
-          switch (sec.getSecurityErrorCode()) {
-            case PERMISSION_DENIED:
-              // retry -- the grant didn't happen yet
-              log.warn("Sleeping because permission was denied");
-              break;
-            default:
-              throw e;
-          }
-        } else {
-          throw e;
-        }
-      }
-
-      Thread.sleep(2000);
-    }
-
-    if (notFound) {
-      s = ReplicationTable.getScanner(conn);
-      for (Entry<Key,Value> content : s) {
-        log.info(content.getKey().toStringNoTruncate() + " => " + ProtobufUtil.toString(Status.parseFrom(content.getValue().get())));
-      }
-      Assert.assertFalse("Did not find the work entry for the status entry", notFound);
-    }
-
-    /**
-     * By this point, we should have data ingested into a table, with at least one WAL as a candidate for replication. Compacting the table should close all
-     * open WALs, which should ensure all records we're going to replicate have entries in the replication table, and nothing will exist in the metadata table
-     * anymore
-     */
-
-    log.info("Killing tserver");
-    // Kill the tserver(s) and restart them
-    // to ensure that the WALs we previously observed all move to closed.
-    cluster.getClusterControl().stop(ServerType.TABLET_SERVER);
-
-    log.info("Starting tserver");
-    cluster.getClusterControl().start(ServerType.TABLET_SERVER);
-
-    log.info("Waiting to read tables");
-    UtilWaitThread.sleep(2 * 3 * 1000);
-
-    // Make sure we can read all the tables (recovery complete)
-    for (String table : new String[] {MetadataTable.NAME, table1}) {
-      Iterators.size(conn.createScanner(table, Authorizations.EMPTY).iterator());
-    }
-
-    log.info("Recovered metadata:");
-    s = conn.createScanner(MetadataTable.NAME, Authorizations.EMPTY);
-    for (Entry<Key,Value> entry : s) {
-      log.info("{}={}", entry.getKey().toStringNoTruncate(), entry.getValue());
-    }
-
-    cluster.getClusterControl().start(ServerType.GARBAGE_COLLECTOR);
-
-    // Wait for a bit since the GC has to run (should be running after a one second delay)
-    waitForGCLock(conn);
-
-    Thread.sleep(1000);
-
-    log.info("After GC");
-    s = conn.createScanner(MetadataTable.NAME, Authorizations.EMPTY);
-    for (Entry<Key,Value> entry : s) {
-      log.info("{}={}", entry.getKey().toStringNoTruncate(), entry.getValue());
-    }
-
-    // We expect no records in the metadata table after compaction. We have to poll
-    // because we have to wait for the StatusMaker's next iteration which will clean
-    // up the dangling *closed* records after we create the record in the replication table.
-    // We need the GC to close the file (CloseWriteAheadLogReferences) before we can remove the record
-    log.info("Checking metadata table for replication entries");
-    Set<String> remaining = new HashSet<>();
-    for (int i = 0; i < 10; i++) {
-      s = conn.createScanner(MetadataTable.NAME, Authorizations.EMPTY);
-      s.setRange(ReplicationSection.getRange());
-      remaining.clear();
-      for (Entry<Key,Value> e : s) {
-        remaining.add(e.getKey().getRow().toString());
-      }
-      remaining.retainAll(entries);
-      if (remaining.isEmpty()) {
-        break;
-      }
-      log.info("remaining {}", remaining);
-      Thread.sleep(2000);
-      log.info("");
-    }
-
-    Assert.assertTrue("Replication status messages were not cleaned up from metadata table", remaining.isEmpty());
-
-    /**
-     * After we close out and subsequently delete the metadata record, this will propagate to the replication table, which will cause those records to be
-     * deleted after replication occurs
-     */
-
-    int recordsFound = 0;
-    for (int i = 0; i < 30; i++) {
-      s = ReplicationTable.getScanner(conn);
-      recordsFound = 0;
-      for (Entry<Key,Value> entry : s) {
-        recordsFound++;
-        log.info("{} {}", entry.getKey().toStringNoTruncate(), ProtobufUtil.toString(Status.parseFrom(entry.getValue().get())));
-      }
-
-      if (recordsFound <= 2) {
-        break;
-      } else {
-        Thread.sleep(1000);
-        log.info("");
-      }
-    }
-
-    Assert.assertTrue("Found unexpected replication records in the replication table", recordsFound <= 2);
-  }
-}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/01ae5b85/test/src/test/java/org/apache/accumulo/test/replication/ReplicationRandomWalkIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/replication/ReplicationRandomWalkIT.java b/test/src/test/java/org/apache/accumulo/test/replication/ReplicationRandomWalkIT.java
deleted file mode 100644
index 80bc69d..0000000
--- a/test/src/test/java/org/apache/accumulo/test/replication/ReplicationRandomWalkIT.java
+++ /dev/null
@@ -1,67 +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.test.replication;
-
-import static org.apache.accumulo.core.conf.Property.TSERV_ARCHIVE_WALOGS;
-import static org.apache.accumulo.core.conf.Property.TSERV_WALOG_MAX_SIZE;
-
-import java.util.Properties;
-
-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.minicluster.impl.MiniAccumuloConfigImpl;
-import org.apache.accumulo.test.functional.ConfigurableMacBase;
-import org.apache.accumulo.test.randomwalk.Environment;
-import org.apache.accumulo.test.randomwalk.concurrent.Replication;
-import org.apache.hadoop.conf.Configuration;
-import org.junit.Test;
-
-public class ReplicationRandomWalkIT extends ConfigurableMacBase {
-
-  @Override
-  protected void configure(MiniAccumuloConfigImpl cfg, Configuration hadoopCoreSite) {
-    cfg.setProperty(TSERV_ARCHIVE_WALOGS, "false");
-    cfg.setProperty(TSERV_WALOG_MAX_SIZE, "1M");
-    cfg.setNumTservers(1);
-  }
-
-  @Test(timeout = 5 * 60 * 1000)
-  public void runReplicationRandomWalkStep() throws Exception {
-    Replication r = new Replication();
-
-    Environment env = new Environment(new Properties()) {
-      @Override
-      public String getUserName() {
-        return "root";
-      }
-
-      @Override
-      public String getPassword() {
-        return ROOT_PASSWORD;
-      }
-
-      @Override
-      public Connector getConnector() throws AccumuloException, AccumuloSecurityException {
-        return ReplicationRandomWalkIT.this.getConnector();
-      }
-
-    };
-    r.visit(null, env, null);
-  }
-
-}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/01ae5b85/test/src/test/java/org/apache/accumulo/test/replication/StatusCombinerMacIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/replication/StatusCombinerMacIT.java b/test/src/test/java/org/apache/accumulo/test/replication/StatusCombinerMacIT.java
deleted file mode 100644
index b072aa7..0000000
--- a/test/src/test/java/org/apache/accumulo/test/replication/StatusCombinerMacIT.java
+++ /dev/null
@@ -1,118 +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.test.replication;
-
-import java.util.EnumSet;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Map.Entry;
-
-import org.apache.accumulo.cluster.ClusterUser;
-import org.apache.accumulo.core.client.BatchWriter;
-import org.apache.accumulo.core.client.Connector;
-import org.apache.accumulo.core.client.Scanner;
-import org.apache.accumulo.core.client.admin.TableOperations;
-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.Value;
-import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
-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.security.TablePermission;
-import org.apache.accumulo.harness.SharedMiniClusterBase;
-import org.apache.accumulo.server.replication.StatusUtil;
-import org.apache.accumulo.server.replication.proto.Replication.Status;
-import org.apache.accumulo.server.util.ReplicationTableUtil;
-import org.apache.hadoop.io.Text;
-import org.junit.Assert;
-import org.junit.Test;
-
-import com.google.common.collect.Iterables;
-
-public class StatusCombinerMacIT extends SharedMiniClusterBase {
-
-  @Override
-  public int defaultTimeoutSeconds() {
-    return 60;
-  }
-
-  @Test
-  public void testCombinerSetOnMetadata() throws Exception {
-    TableOperations tops = getConnector().tableOperations();
-    Map<String,EnumSet<IteratorScope>> iterators = tops.listIterators(MetadataTable.NAME);
-
-    Assert.assertTrue(iterators.containsKey(ReplicationTableUtil.COMBINER_NAME));
-    EnumSet<IteratorScope> scopes = iterators.get(ReplicationTableUtil.COMBINER_NAME);
-    Assert.assertEquals(3, scopes.size());
-    Assert.assertTrue(scopes.contains(IteratorScope.scan));
-    Assert.assertTrue(scopes.contains(IteratorScope.minc));
-    Assert.assertTrue(scopes.contains(IteratorScope.majc));
-
-    Iterable<Entry<String,String>> propIter = tops.getProperties(MetadataTable.NAME);
-    HashMap<String,String> properties = new HashMap<String,String>();
-    for (Entry<String,String> entry : propIter) {
-      properties.put(entry.getKey(), entry.getValue());
-    }
-
-    for (IteratorScope scope : scopes) {
-      String key = Property.TABLE_ITERATOR_PREFIX.getKey() + scope.name() + "." + ReplicationTableUtil.COMBINER_NAME + ".opt.columns";
-      Assert.assertTrue("Properties did not contain key : " + key, properties.containsKey(key));
-      Assert.assertEquals(MetadataSchema.ReplicationSection.COLF.toString(), properties.get(key));
-    }
-  }
-
-  @Test
-  public void test() throws Exception {
-    Connector conn = getConnector();
-    ClusterUser user = getAdminUser();
-
-    ReplicationTable.setOnline(conn);
-    conn.securityOperations().grantTablePermission(user.getPrincipal(), 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");
-      StatusSection.add(m, new Text("1"), StatusUtil.fileCreatedValue(createTime));
-      bw.addMutation(m);
-    } finally {
-      bw.close();
-    }
-
-    Scanner s = ReplicationTable.getScanner(conn);
-    Entry<Key,Value> entry = Iterables.getOnlyElement(s);
-    Assert.assertEquals(StatusUtil.fileCreatedValue(createTime), entry.getValue());
-
-    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)));
-      bw.addMutation(m);
-    } finally {
-      bw.close();
-    }
-
-    s = ReplicationTable.getScanner(conn);
-    entry = Iterables.getOnlyElement(s);
-    Status stat = Status.parseFrom(entry.getValue().get());
-    Assert.assertEquals(Long.MAX_VALUE, stat.getBegin());
-  }
-
-}


Mime
View raw message