/*
 * 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.hadoop.hbase.client;

import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.TreeMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HBaseClassTestRule;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HBaseTestingUtil;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.Waiter;
import org.apache.hadoop.hbase.coprocessor.CoreCoprocessor;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.coprocessor.RegionObserver;
import org.apache.hadoop.hbase.regionserver.RegionServerStoppedException;
import org.apache.hadoop.hbase.regionserver.StorefileRefresherChore;
import org.apache.hadoop.hbase.regionserver.TestHRegionServerBulkLoad;
import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
import org.apache.hadoop.hbase.testclassification.ClientTests;
import org.apache.hadoop.hbase.testclassification.LargeTests;
import org.apache.hadoop.hbase.tool.BulkLoadHFiles;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.zookeeper.MiniZooKeeperCluster;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Category({ LargeTests.class, ClientTests.class })
public class TestReplicaWithCluster {

  @ClassRule
  public static final HBaseClassTestRule CLASS_RULE =
    HBaseClassTestRule.forClass(TestReplicaWithCluster.class);

  private static final Logger LOG = LoggerFactory.getLogger(TestReplicaWithCluster.class);

  private static final int NB_SERVERS = 3;
  private static final byte[] row = Bytes.toBytes(TestReplicaWithCluster.class.getName());
  private static final HBaseTestingUtil HTU = new HBaseTestingUtil();

  // second minicluster used in testing of replication
  private static HBaseTestingUtil HTU2;
  private static final byte[] f = HConstants.CATALOG_FAMILY;

  private final static int REFRESH_PERIOD = 1000;
  private final static int META_SCAN_TIMEOUT_IN_MILLISEC = 200;

  /**
   * This copro is used to synchronize the tests.
   */
  public static class SlowMeCopro implements RegionCoprocessor, RegionObserver {
    static final AtomicLong sleepTime = new AtomicLong(0);
    static final AtomicReference<CountDownLatch> cdl = new AtomicReference<>(new CountDownLatch(0));

    public SlowMeCopro() {
    }

    @Override
    public Optional<RegionObserver> getRegionObserver() {
      return Optional.of(this);
    }

    @Override
    public void preGetOp(final ObserverContext<? extends RegionCoprocessorEnvironment> e,
      final Get get, final List<Cell> results) throws IOException {

      if (e.getEnvironment().getRegion().getRegionInfo().getReplicaId() == 0) {
        CountDownLatch latch = cdl.get();
        try {
          if (sleepTime.get() > 0) {
            LOG.info("Sleeping for " + sleepTime.get() + " ms");
            Thread.sleep(sleepTime.get());
          } else if (latch.getCount() > 0) {
            LOG.info("Waiting for the counterCountDownLatch");
            latch.await(2, TimeUnit.MINUTES); // To help the tests to finish.
            if (latch.getCount() > 0) {
              throw new RuntimeException("Can't wait more");
            }
          }
        } catch (InterruptedException e1) {
          LOG.error(e1.toString(), e1);
        }
      } else {
        LOG.info("We're not the primary replicas.");
      }
    }
  }

  /**
   * This copro is used to simulate region server down exception for Get and Scan
   */
  @CoreCoprocessor
  public static class RegionServerStoppedCopro implements RegionCoprocessor, RegionObserver {

    public RegionServerStoppedCopro() {
    }

    @Override
    public Optional<RegionObserver> getRegionObserver() {
      return Optional.of(this);
    }

    @Override
    public void preGetOp(final ObserverContext<? extends RegionCoprocessorEnvironment> e,
      final Get get, final List<Cell> results) throws IOException {

      int replicaId = e.getEnvironment().getRegion().getRegionInfo().getReplicaId();

      // Fail for the primary replica and replica 1
      if (e.getEnvironment().getRegion().getRegionInfo().getReplicaId() <= 1) {
        LOG.info("Throw Region Server Stopped Exceptoin for replica id " + replicaId);
        throw new RegionServerStoppedException(
          "Server " + e.getEnvironment().getServerName() + " not running");
      } else {
        LOG.info("We're replica region " + replicaId);
      }
    }

    @Override
    public void preScannerOpen(final ObserverContext<? extends RegionCoprocessorEnvironment> e,
      final Scan scan) throws IOException {
      int replicaId = e.getEnvironment().getRegion().getRegionInfo().getReplicaId();
      // Fail for the primary replica and replica 1
      if (e.getEnvironment().getRegion().getRegionInfo().getReplicaId() <= 1) {
        LOG.info("Throw Region Server Stopped Exceptoin for replica id " + replicaId);
        throw new RegionServerStoppedException(
          "Server " + e.getEnvironment().getServerName() + " not running");
      } else {
        LOG.info("We're replica region " + replicaId);
      }
    }
  }

  /**
   * This copro is used to slow down the primary meta region scan a bit
   */
  public static class RegionServerHostingPrimayMetaRegionSlowOrStopCopro
    implements RegionCoprocessor, RegionObserver {
    static boolean slowDownPrimaryMetaScan = false;
    static boolean throwException = false;

    @Override
    public Optional<RegionObserver> getRegionObserver() {
      return Optional.of(this);
    }

    @Override
    public void preGetOp(final ObserverContext<? extends RegionCoprocessorEnvironment> e,
      final Get get, final List<Cell> results) throws IOException {

      int replicaId = e.getEnvironment().getRegion().getRegionInfo().getReplicaId();

      // Fail for the primary replica, but not for meta
      if (throwException) {
        if (!e.getEnvironment().getRegion().getRegionInfo().isMetaRegion() && (replicaId == 0)) {
          LOG.info("Get, throw Region Server Stopped Exceptoin for region "
            + e.getEnvironment().getRegion().getRegionInfo());
          throw new RegionServerStoppedException(
            "Server " + e.getEnvironment().getServerName() + " not running");
        }
      } else {
        LOG.info("Get, We're replica region " + replicaId);
      }
    }

    @Override
    public void preScannerOpen(final ObserverContext<? extends RegionCoprocessorEnvironment> e,
      final Scan scan) throws IOException {

      int replicaId = e.getEnvironment().getRegion().getRegionInfo().getReplicaId();

      // Slow down with the primary meta region scan
      if (e.getEnvironment().getRegion().getRegionInfo().isMetaRegion() && (replicaId == 0)) {
        if (slowDownPrimaryMetaScan) {
          LOG.info("Scan with primary meta region, slow down a bit");
          try {
            Thread.sleep(META_SCAN_TIMEOUT_IN_MILLISEC - 50);
          } catch (InterruptedException ie) {
            // Ingore
          }
        }

        // Fail for the primary replica
        if (throwException) {
          LOG.info("Scan, throw Region Server Stopped Exceptoin for replica "
            + e.getEnvironment().getRegion().getRegionInfo());

          throw new RegionServerStoppedException(
            "Server " + e.getEnvironment().getServerName() + " not running");
        } else {
          LOG.info("Scan, We're replica region " + replicaId);
        }
      } else {
        LOG.info("Scan, We're replica region " + replicaId);
      }
    }
  }

  @BeforeClass
  public static void beforeClass() throws Exception {
    // enable store file refreshing
    HTU.getConfiguration().setInt(StorefileRefresherChore.REGIONSERVER_STOREFILE_REFRESH_PERIOD,
      REFRESH_PERIOD);

    HTU.getConfiguration().setFloat("hbase.regionserver.logroll.multiplier", 0.0001f);
    HTU.getConfiguration().setInt("replication.source.size.capacity", 10240);
    HTU.getConfiguration().setLong("replication.source.sleepforretries", 100);
    HTU.getConfiguration().setInt("hbase.regionserver.maxlogs", 2);
    HTU.getConfiguration().setLong("hbase.master.logcleaner.ttl", 10);
    HTU.getConfiguration().setInt("zookeeper.recovery.retry", 1);
    HTU.getConfiguration().setInt("zookeeper.recovery.retry.intervalmill", 10);

    // Wait for primary call longer so make sure that it will get exception from the primary call
    HTU.getConfiguration().setInt("hbase.client.primaryCallTimeout.get", 1000000);
    HTU.getConfiguration().setInt("hbase.client.primaryCallTimeout.scan", 1000000);

    // Make sure master does not host system tables.
    HTU.getConfiguration().set("hbase.balancer.tablesOnMaster", "none");

    // Set system coprocessor so it can be applied to meta regions
    HTU.getConfiguration().set("hbase.coprocessor.region.classes",
      RegionServerHostingPrimayMetaRegionSlowOrStopCopro.class.getName());

    HTU.getConfiguration().setInt(HConstants.HBASE_CLIENT_META_REPLICA_SCAN_TIMEOUT,
      META_SCAN_TIMEOUT_IN_MILLISEC * 1000);

    HTU.startMiniCluster(NB_SERVERS);
    // Enable meta replica at server side
    HBaseTestingUtil.setReplicas(HTU.getAdmin(), TableName.META_TABLE_NAME, 2);

    HTU.getHBaseCluster().startMaster();
  }

  @AfterClass
  public static void afterClass() throws Exception {
    if (HTU2 != null) HTU2.shutdownMiniCluster();
    HTU.shutdownMiniCluster();
  }

  @Test
  public void testCreateDeleteTable() throws IOException {
    // Create table then get the single region for our new table.
    TableDescriptorBuilder builder =
      HTU.createModifyableTableDescriptor(TableName.valueOf("testCreateDeleteTable"),
        ColumnFamilyDescriptorBuilder.DEFAULT_MIN_VERSIONS, 3, HConstants.FOREVER,
        ColumnFamilyDescriptorBuilder.DEFAULT_KEEP_DELETED);
    builder.setRegionReplication(NB_SERVERS);
    builder.setCoprocessor(SlowMeCopro.class.getName());
    TableDescriptor hdt = builder.build();
    Table table = HTU.createTable(hdt, new byte[][] { f }, null);

    Put p = new Put(row);
    p.addColumn(f, row, row);
    table.put(p);

    Get g = new Get(row);
    Result r = table.get(g);
    Assert.assertFalse(r.isStale());

    try {
      // But if we ask for stale we will get it
      SlowMeCopro.cdl.set(new CountDownLatch(1));
      g = new Get(row);
      g.setConsistency(Consistency.TIMELINE);
      r = table.get(g);
      Assert.assertTrue(r.isStale());
      SlowMeCopro.cdl.get().countDown();
    } finally {
      SlowMeCopro.cdl.get().countDown();
      SlowMeCopro.sleepTime.set(0);
    }

    HTU.getAdmin().disableTable(hdt.getTableName());
    HTU.deleteTable(hdt.getTableName());
  }

  @Test
  public void testChangeTable() throws Exception {
    TableDescriptor td = TableDescriptorBuilder.newBuilder(TableName.valueOf("testChangeTable"))
      .setRegionReplication(NB_SERVERS).setCoprocessor(SlowMeCopro.class.getName())
      .setColumnFamily(ColumnFamilyDescriptorBuilder.of(f)).build();
    HTU.getAdmin().createTable(td);
    Table table = HTU.getConnection().getTable(td.getTableName());
    // basic test: it should work.
    Put p = new Put(row);
    p.addColumn(f, row, row);
    table.put(p);

    Get g = new Get(row);
    Result r = table.get(g);
    Assert.assertFalse(r.isStale());

    // Add a CF, it should work.
    TableDescriptor bHdt = HTU.getAdmin().getDescriptor(td.getTableName());
    td = TableDescriptorBuilder.newBuilder(td)
      .setColumnFamily(ColumnFamilyDescriptorBuilder.of(row)).build();
    HTU.getAdmin().disableTable(td.getTableName());
    HTU.getAdmin().modifyTable(td);
    HTU.getAdmin().enableTable(td.getTableName());
    TableDescriptor nHdt = HTU.getAdmin().getDescriptor(td.getTableName());
    Assert.assertEquals("fams=" + Arrays.toString(nHdt.getColumnFamilies()),
      bHdt.getColumnFamilyCount() + 1, nHdt.getColumnFamilyCount());

    p = new Put(row);
    p.addColumn(row, row, row);
    table.put(p);

    g = new Get(row);
    r = table.get(g);
    Assert.assertFalse(r.isStale());

    try {
      SlowMeCopro.cdl.set(new CountDownLatch(1));
      g = new Get(row);
      g.setConsistency(Consistency.TIMELINE);
      r = table.get(g);
      Assert.assertTrue(r.isStale());
    } finally {
      SlowMeCopro.cdl.get().countDown();
      SlowMeCopro.sleepTime.set(0);
    }

    Admin admin = HTU.getAdmin();
    nHdt = admin.getDescriptor(td.getTableName());
    Assert.assertEquals("fams=" + Arrays.toString(nHdt.getColumnFamilies()),
      bHdt.getColumnFamilyCount() + 1, nHdt.getColumnFamilyCount());

    admin.disableTable(td.getTableName());
    admin.deleteTable(td.getTableName());
    admin.close();
  }

  @SuppressWarnings("deprecation")
  @Test
  public void testReplicaAndReplication() throws Exception {
    TableDescriptorBuilder builder =
      HTU.createModifyableTableDescriptor("testReplicaAndReplication");
    builder.setRegionReplication(NB_SERVERS);
    builder.setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(row)
      .setScope(HConstants.REPLICATION_SCOPE_GLOBAL).build());

    builder.setCoprocessor(SlowMeCopro.class.getName());
    TableDescriptor tableDescriptor = builder.build();
    HTU.getAdmin().createTable(tableDescriptor, HBaseTestingUtil.KEYS_FOR_HBA_CREATE_TABLE);

    Configuration conf2 = HBaseConfiguration.create(HTU.getConfiguration());
    conf2.set(HConstants.HBASE_CLIENT_INSTANCE_ID, String.valueOf(-1));
    conf2.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/2");
    MiniZooKeeperCluster miniZK = HTU.getZkCluster();

    HTU2 = new HBaseTestingUtil(conf2);
    HTU2.setZkCluster(miniZK);
    HTU2.startMiniCluster(NB_SERVERS);
    LOG.info("Setup second Zk");
    HTU2.getAdmin().createTable(tableDescriptor, HBaseTestingUtil.KEYS_FOR_HBA_CREATE_TABLE);

    try (Connection connection = ConnectionFactory.createConnection(HTU.getConfiguration());
      Admin admin = connection.getAdmin()) {
      ReplicationPeerConfig rpc =
        ReplicationPeerConfig.newBuilder().setClusterKey(HTU2.getRpcConnnectionURI()).build();
      admin.addReplicationPeer("2", rpc);
    }

    Put p = new Put(row);
    p.addColumn(row, row, row);
    final Table table = HTU.getConnection().getTable(tableDescriptor.getTableName());
    table.put(p);

    HTU.getAdmin().flush(table.getName());
    LOG.info("Put & flush done on the first cluster. Now doing a get on the same cluster.");

    Waiter.waitFor(HTU.getConfiguration(), 1000, new Waiter.Predicate<Exception>() {
      @Override
      public boolean evaluate() throws Exception {
        try {
          SlowMeCopro.cdl.set(new CountDownLatch(1));
          Get g = new Get(row);
          g.setConsistency(Consistency.TIMELINE);
          Result r = table.get(g);
          Assert.assertTrue(r.isStale());
          return !r.isEmpty();
        } finally {
          SlowMeCopro.cdl.get().countDown();
          SlowMeCopro.sleepTime.set(0);
        }
      }
    });
    table.close();
    LOG.info("stale get on the first cluster done. Now for the second.");

    final Table table2 = HTU.getConnection().getTable(tableDescriptor.getTableName());
    Waiter.waitFor(HTU.getConfiguration(), 1000, new Waiter.Predicate<Exception>() {
      @Override
      public boolean evaluate() throws Exception {
        try {
          SlowMeCopro.cdl.set(new CountDownLatch(1));
          Get g = new Get(row);
          g.setConsistency(Consistency.TIMELINE);
          Result r = table2.get(g);
          Assert.assertTrue(r.isStale());
          return !r.isEmpty();
        } finally {
          SlowMeCopro.cdl.get().countDown();
          SlowMeCopro.sleepTime.set(0);
        }
      }
    });
    table2.close();

    HTU.getAdmin().disableTable(tableDescriptor.getTableName());
    HTU.deleteTable(tableDescriptor.getTableName());

    HTU2.getAdmin().disableTable(tableDescriptor.getTableName());
    HTU2.deleteTable(tableDescriptor.getTableName());

    // We shutdown HTU2 minicluster later, in afterClass(), as shutting down
    // the minicluster has negative impact of deleting all HConnections in JVM.
  }

  @Test
  public void testBulkLoad() throws IOException {
    // Create table then get the single region for our new table.
    LOG.debug("Creating test table");
    TableDescriptorBuilder builder = HTU.createModifyableTableDescriptor(
      TableName.valueOf("testBulkLoad"), ColumnFamilyDescriptorBuilder.DEFAULT_MIN_VERSIONS, 3,
      HConstants.FOREVER, ColumnFamilyDescriptorBuilder.DEFAULT_KEEP_DELETED);
    builder.setRegionReplication(NB_SERVERS);
    builder.setCoprocessor(SlowMeCopro.class.getName());
    TableDescriptor hdt = builder.build();
    Table table = HTU.createTable(hdt, new byte[][] { f }, null);

    // create hfiles to load.
    LOG.debug("Creating test data");
    Path dir = HTU.getDataTestDirOnTestFS("testBulkLoad");
    final int numRows = 10;
    final byte[] qual = Bytes.toBytes("qual");
    final byte[] val = Bytes.toBytes("val");
    Map<byte[], List<Path>> family2Files = new TreeMap<>(Bytes.BYTES_COMPARATOR);
    for (ColumnFamilyDescriptor col : hdt.getColumnFamilies()) {
      Path hfile = new Path(dir, col.getNameAsString());
      TestHRegionServerBulkLoad.createHFile(HTU.getTestFileSystem(), hfile, col.getName(), qual,
        val, numRows);
      family2Files.put(col.getName(), Collections.singletonList(hfile));
    }

    // bulk load HFiles
    LOG.debug("Loading test data");
    BulkLoadHFiles.create(HTU.getConfiguration()).bulkLoad(hdt.getTableName(), family2Files);

    // verify we can read them from the primary
    LOG.debug("Verifying data load");
    for (int i = 0; i < numRows; i++) {
      byte[] row = TestHRegionServerBulkLoad.rowkey(i);
      Get g = new Get(row);
      Result r = table.get(g);
      Assert.assertFalse(r.isStale());
    }

    // verify we can read them from the replica
    LOG.debug("Verifying replica queries");
    try {
      SlowMeCopro.cdl.set(new CountDownLatch(1));
      for (int i = 0; i < numRows; i++) {
        byte[] row = TestHRegionServerBulkLoad.rowkey(i);
        Get g = new Get(row);
        g.setConsistency(Consistency.TIMELINE);
        Result r = table.get(g);
        Assert.assertTrue(r.isStale());
      }
      SlowMeCopro.cdl.get().countDown();
    } finally {
      SlowMeCopro.cdl.get().countDown();
      SlowMeCopro.sleepTime.set(0);
    }

    HTU.getAdmin().disableTable(hdt.getTableName());
    HTU.deleteTable(hdt.getTableName());
  }

  @Test
  public void testReplicaGetWithPrimaryDown() throws IOException {
    // Create table then get the single region for our new table.
    TableDescriptorBuilder builder =
      HTU.createModifyableTableDescriptor(TableName.valueOf("testCreateDeleteTable"),
        ColumnFamilyDescriptorBuilder.DEFAULT_MIN_VERSIONS, 3, HConstants.FOREVER,
        ColumnFamilyDescriptorBuilder.DEFAULT_KEEP_DELETED);
    builder.setRegionReplication(NB_SERVERS);
    builder.setCoprocessor(RegionServerStoppedCopro.class.getName());
    TableDescriptor hdt = builder.build();
    try {
      Table table = HTU.createTable(hdt, new byte[][] { f }, null);

      Put p = new Put(row);
      p.addColumn(f, row, row);
      table.put(p);

      // Flush so it can be picked by the replica refresher thread
      HTU.flush(table.getName());

      // Sleep for some time until data is picked up by replicas
      try {
        Thread.sleep(2 * REFRESH_PERIOD);
      } catch (InterruptedException e1) {
        LOG.error(e1.toString(), e1);
      }

      // But if we ask for stale we will get it
      Get g = new Get(row);
      g.setConsistency(Consistency.TIMELINE);
      Result r = table.get(g);
      Assert.assertTrue(r.isStale());
    } finally {
      HTU.getAdmin().disableTable(hdt.getTableName());
      HTU.deleteTable(hdt.getTableName());
    }
  }

  @Test
  public void testReplicaScanWithPrimaryDown() throws IOException {
    // Create table then get the single region for our new table.
    TableDescriptorBuilder builder =
      HTU.createModifyableTableDescriptor(TableName.valueOf("testCreateDeleteTable"),
        ColumnFamilyDescriptorBuilder.DEFAULT_MIN_VERSIONS, 3, HConstants.FOREVER,
        ColumnFamilyDescriptorBuilder.DEFAULT_KEEP_DELETED);
    builder.setRegionReplication(NB_SERVERS);
    builder.setCoprocessor(RegionServerStoppedCopro.class.getName());
    TableDescriptor hdt = builder.build();
    try {
      Table table = HTU.createTable(hdt, new byte[][] { f }, null);

      Put p = new Put(row);
      p.addColumn(f, row, row);
      table.put(p);

      // Flush so it can be picked by the replica refresher thread
      HTU.flush(table.getName());

      // Sleep for some time until data is picked up by replicas
      try {
        Thread.sleep(2 * REFRESH_PERIOD);
      } catch (InterruptedException e1) {
        LOG.error(e1.toString(), e1);
      }

      // But if we ask for stale we will get it
      // Instantiating the Scan class
      Scan scan = new Scan();

      // Scanning the required columns
      scan.addFamily(f);
      scan.setConsistency(Consistency.TIMELINE);

      // Getting the scan result
      ResultScanner scanner = table.getScanner(scan);

      Result r = scanner.next();

      Assert.assertTrue(r.isStale());
    } finally {
      HTU.getAdmin().disableTable(hdt.getTableName());
      HTU.deleteTable(hdt.getTableName());
    }
  }

  @Test
  public void testReplicaGetWithAsyncRpcClientImpl() throws IOException {
    HTU.getConfiguration().setBoolean("hbase.ipc.client.specificThreadForWriting", true);
    HTU.getConfiguration().set("hbase.rpc.client.impl",
      "org.apache.hadoop.hbase.ipc.AsyncRpcClient");
    // Create table then get the single region for our new table.
    TableDescriptorBuilder builder =
      HTU.createModifyableTableDescriptor(TableName.valueOf("testReplicaGetWithAsyncRpcClientImpl"),
        ColumnFamilyDescriptorBuilder.DEFAULT_MIN_VERSIONS, 3, HConstants.FOREVER,
        ColumnFamilyDescriptorBuilder.DEFAULT_KEEP_DELETED);
    builder.setRegionReplication(NB_SERVERS);
    builder.setCoprocessor(SlowMeCopro.class.getName());
    TableDescriptor hdt = builder.build();
    try {
      Table table = HTU.createTable(hdt, new byte[][] { f }, null);

      Put p = new Put(row);
      p.addColumn(f, row, row);
      table.put(p);

      // Flush so it can be picked by the replica refresher thread
      HTU.flush(table.getName());

      // Sleep for some time until data is picked up by replicas
      try {
        Thread.sleep(2 * REFRESH_PERIOD);
      } catch (InterruptedException e1) {
        LOG.error(e1.toString(), e1);
      }

      try {
        // Create the new connection so new config can kick in
        Connection connection = ConnectionFactory.createConnection(HTU.getConfiguration());
        Table t = connection.getTable(hdt.getTableName());

        // But if we ask for stale we will get it
        SlowMeCopro.cdl.set(new CountDownLatch(1));
        Get g = new Get(row);
        g.setConsistency(Consistency.TIMELINE);
        Result r = t.get(g);
        Assert.assertTrue(r.isStale());
        SlowMeCopro.cdl.get().countDown();
      } finally {
        SlowMeCopro.cdl.get().countDown();
        SlowMeCopro.sleepTime.set(0);
      }
    } finally {
      HTU.getConfiguration().unset("hbase.ipc.client.specificThreadForWriting");
      HTU.getConfiguration().unset("hbase.rpc.client.impl");
      HTU.getAdmin().disableTable(hdt.getTableName());
      HTU.deleteTable(hdt.getTableName());
    }
  }
}