org.apache.hadoop.hbase.util.EnvironmentEdgeManager#reset ( )源码实例Demo

@Test
public void testSortExtract(){
  ManualEnvironmentEdge mee = new ManualEnvironmentEdge();
  EnvironmentEdgeManager.injectEdge(mee);
  mee.setValue(1);

  DeadServer d = new DeadServer();

  d.putIfAbsent(hostname123);
  mee.incValue(1);
  d.putIfAbsent(hostname1234);
  mee.incValue(1);
  d.putIfAbsent(hostname12345);

  List<Pair<ServerName, Long>> copy = d.copyDeadServersSince(2L);
  Assert.assertEquals(2, copy.size());

  Assert.assertEquals(hostname1234, copy.get(0).getFirst());
  Assert.assertEquals(new Long(2L), copy.get(0).getSecond());

  Assert.assertEquals(hostname12345, copy.get(1).getFirst());
  Assert.assertEquals(new Long(3L), copy.get(1).getSecond());

  EnvironmentEdgeManager.reset();
}
 

protected void checkShouldFlush(Configuration conf, boolean expected) throws Exception {
  try {
    EnvironmentEdgeForMemstoreTest edge = new EnvironmentEdgeForMemstoreTest();
    EnvironmentEdgeManager.injectEdge(edge);
    HBaseTestingUtility hbaseUtility = new HBaseTestingUtility(conf);
    String cf = "foo";
    HRegion region =
        hbaseUtility.createTestRegion("foobar", ColumnFamilyDescriptorBuilder.of(cf));

    edge.setCurrentTimeMillis(1234);
    Put p = new Put(Bytes.toBytes("r"));
    p.add(KeyValueTestUtil.create("r", cf, "q", 100, "v"));
    region.put(p);
    edge.setCurrentTimeMillis(1234 + 100);
    StringBuilder sb = new StringBuilder();
    assertTrue(!region.shouldFlush(sb));
    edge.setCurrentTimeMillis(1234 + 10000);
    assertTrue(region.shouldFlush(sb) == expected);
  } finally {
    EnvironmentEdgeManager.reset();
  }
}
 

@AfterClass
public static void tearDownAfterClass() throws Exception {
  EnvironmentEdgeManager.reset();
  for (int i = 0; i < tables.length; ++i) {
    if (tables[i] != null) {
      tables[i].close();
      TEST_UTIL.deleteTable(TABLE_NAMES[i]);
    }
  }
  TEST_UTIL.deleteTable(TABLE_NAME);
  TEST_UTIL.getAdmin().deleteNamespace(NAMESPACE);
  TEST_UTIL.shutdownMiniCluster();
}
 

@Test
public void testOverconsumptionFixedIntervalRefillStrategy() throws InterruptedException {
  RateLimiter limiter = new FixedIntervalRateLimiter();
  limiter.set(10, TimeUnit.SECONDS);

  // fix the current time in order to get the precise value of interval
  EnvironmentEdge edge = new EnvironmentEdge() {
    private final long ts = System.currentTimeMillis();

    @Override
    public long currentTime() {
      return ts;
    }
  };
  EnvironmentEdgeManager.injectEdge(edge);
  // 10 resources are available, but we need to consume 20 resources
  // Verify that we have to wait at least 1.1sec to have 1 resource available
  assertTrue(limiter.canExecute());
  limiter.consume(20);
  // To consume 1 resource also wait for 1000ms
  assertEquals(1000, limiter.waitInterval(1));
  // To consume 10 resource wait for 100ms
  assertEquals(1000, limiter.waitInterval(10));
  EnvironmentEdgeManager.reset();

  limiter.setNextRefillTime(limiter.getNextRefillTime() - 900);
  // Verify that after 1sec also no resource should be available
  assertFalse(limiter.canExecute(1));
  limiter.setNextRefillTime(limiter.getNextRefillTime() - 100);

  // Verify that after 1sec the 10 resource is available
  assertTrue(limiter.canExecute());
  assertEquals(0, limiter.waitInterval());
}
 

@Test
public void testUnconfiguredLimiters() throws InterruptedException {

  ManualEnvironmentEdge testEdge = new ManualEnvironmentEdge();
  EnvironmentEdgeManager.injectEdge(testEdge);
  long limit = Long.MAX_VALUE;

  // For unconfigured limiters, it is supposed to use as much as possible
  RateLimiter avgLimiter = new AverageIntervalRateLimiter();
  RateLimiter fixLimiter = new FixedIntervalRateLimiter();

  assertEquals(limit, avgLimiter.getAvailable());
  assertEquals(limit, fixLimiter.getAvailable());

  assertTrue(avgLimiter.canExecute(limit));
  avgLimiter.consume(limit);

  assertTrue(fixLimiter.canExecute(limit));
  fixLimiter.consume(limit);

  // Make sure that available is Long.MAX_VALUE
  assertTrue(limit == avgLimiter.getAvailable());
  assertTrue(limit == fixLimiter.getAvailable());

  // after 100 millseconds, it should be able to execute limit as well
  testEdge.incValue(100);

  assertTrue(avgLimiter.canExecute(limit));
  avgLimiter.consume(limit);

  assertTrue(fixLimiter.canExecute(limit));
  fixLimiter.consume(limit);

  // Make sure that available is Long.MAX_VALUE
  assertTrue(limit == avgLimiter.getAvailable());
  assertTrue(limit == fixLimiter.getAvailable());

  EnvironmentEdgeManager.reset();
}
 

@AfterClass
public static void tearDownAfterClass() throws Exception {
  EnvironmentEdgeManager.reset();
  for (int i = 0; i < tables.length; ++i) {
    if (tables[i] != null) {
      tables[i].close();
      TEST_UTIL.deleteTable(TABLE_NAMES[i]);
    }
  }

  TEST_UTIL.shutdownMiniCluster();
}
 

/**
 * Tests that the timeOfOldestEdit is updated correctly for the
 * various edit operations in memstore.
 * @throws Exception
 */
@Test
public void testUpdateToTimeOfOldestEdit() throws Exception {
  try {
    EnvironmentEdgeForMemstoreTest edge = new EnvironmentEdgeForMemstoreTest();
    EnvironmentEdgeManager.injectEdge(edge);
    DefaultMemStore memstore = new DefaultMemStore();
    long t = memstore.timeOfOldestEdit();
    assertEquals(Long.MAX_VALUE, t);

    // test the case that the timeOfOldestEdit is updated after a KV add
    memstore.add(KeyValueTestUtil.create("r", "f", "q", 100, "v"), null);
    t = memstore.timeOfOldestEdit();
    assertTrue(t == 1234);
    // snapshot() will reset timeOfOldestEdit. The method will also assert the
    // value is reset to Long.MAX_VALUE
    t = runSnapshot(memstore);

    // test the case that the timeOfOldestEdit is updated after a KV delete
    memstore.add(KeyValueTestUtil.create("r", "f", "q", 100, KeyValue.Type.Delete, "v"), null);
    t = memstore.timeOfOldestEdit();
    assertTrue(t == 1234);
    t = runSnapshot(memstore);

    // test the case that the timeOfOldestEdit is updated after a KV upsert
    List<Cell> l = new ArrayList<>();
    KeyValue kv1 = KeyValueTestUtil.create("r", "f", "q", 100, "v");
    kv1.setSequenceId(100);
    l.add(kv1);
    memstore.upsert(l, 1000, null);
    t = memstore.timeOfOldestEdit();
    assertTrue(t == 1234);
  } finally {
    EnvironmentEdgeManager.reset();
  }
}
 

@SuppressWarnings("unchecked")
@Test
public void testMergeExpiredFiles() throws Exception {
  ManualEnvironmentEdge edge = new ManualEnvironmentEdge();
  long now = defaultTtl + 2;
  edge.setValue(now);
  EnvironmentEdgeManager.injectEdge(edge);
  try {
    HStoreFile expiredFile = createFile(), notExpiredFile = createFile();
    when(expiredFile.getReader().getMaxTimestamp()).thenReturn(now - defaultTtl - 1);
    when(notExpiredFile.getReader().getMaxTimestamp()).thenReturn(now - defaultTtl + 1);
    List<HStoreFile> expired = Lists.newArrayList(expiredFile, expiredFile);
    List<HStoreFile> notExpired = Lists.newArrayList(notExpiredFile, notExpiredFile);
    List<HStoreFile> mixed = Lists.newArrayList(expiredFile, notExpiredFile);

    StripeCompactionPolicy policy = createPolicy(HBaseConfiguration.create(),
        defaultSplitSize, defaultSplitCount, defaultInitialCount, true);
    // Merge expired if there are eligible stripes.
    StripeCompactionPolicy.StripeInformationProvider si =
        createStripesWithFiles(expired, expired, expired);
    verifyWholeStripesCompaction(policy, si, 0, 2, null, 1, Long.MAX_VALUE, false);
    // Don't merge if nothing expired.
    si = createStripesWithFiles(notExpired, notExpired, notExpired);
    assertNull(policy.selectCompaction(si, al(), false));
    // Merge one expired stripe with next.
    si = createStripesWithFiles(notExpired, expired, notExpired);
    verifyWholeStripesCompaction(policy, si, 1, 2, null, 1, Long.MAX_VALUE, false);
    // Merge the biggest run out of multiple options.
    // Merge one expired stripe with next.
    si = createStripesWithFiles(notExpired, expired, notExpired, expired, expired, notExpired);
    verifyWholeStripesCompaction(policy, si, 3, 4, null, 1, Long.MAX_VALUE, false);
    // Stripe with a subset of expired files is not merged.
    si = createStripesWithFiles(expired, expired, notExpired, expired, mixed);
    verifyWholeStripesCompaction(policy, si, 0, 1, null, 1, Long.MAX_VALUE, false);
  } finally {
    EnvironmentEdgeManager.reset();
  }
}
 

@SuppressWarnings("unchecked")
@Test
public void testMergeExpiredStripes() throws Exception {
  // HBASE-11397
  ManualEnvironmentEdge edge = new ManualEnvironmentEdge();
  long now = defaultTtl + 2;
  edge.setValue(now);
  EnvironmentEdgeManager.injectEdge(edge);
  try {
    HStoreFile expiredFile = createFile(), notExpiredFile = createFile();
    when(expiredFile.getReader().getMaxTimestamp()).thenReturn(now - defaultTtl - 1);
    when(notExpiredFile.getReader().getMaxTimestamp()).thenReturn(now - defaultTtl + 1);
    List<HStoreFile> expired = Lists.newArrayList(expiredFile, expiredFile);
    List<HStoreFile> notExpired = Lists.newArrayList(notExpiredFile, notExpiredFile);

    StripeCompactionPolicy policy =
        createPolicy(HBaseConfiguration.create(), defaultSplitSize, defaultSplitCount,
          defaultInitialCount, true);

    // Merge all three expired stripes into one.
    StripeCompactionPolicy.StripeInformationProvider si =
        createStripesWithFiles(expired, expired, expired);
    verifyMergeCompatcion(policy, si, 0, 2);

    // Merge two adjacent expired stripes into one.
    si = createStripesWithFiles(notExpired, expired, notExpired, expired, expired, notExpired);
    verifyMergeCompatcion(policy, si, 3, 4);
  } finally {
    EnvironmentEdgeManager.reset();
  }
}
 

@Test
public void testWalNonces() throws Exception {
  ManualEnvironmentEdge edge = new ManualEnvironmentEdge();
  EnvironmentEdgeManager.injectEdge(edge);
  try {
    ServerNonceManager nm = createManager(6);
    ScheduledChore cleanup = nm.createCleanupScheduledChore(Mockito.mock(Stoppable.class));
    // Add nonces from WAL, including dups.
    edge.setValue(12);
    nm.reportOperationFromWal(NO_NONCE, 1, 8);
    nm.reportOperationFromWal(NO_NONCE, 2, 2);
    nm.reportOperationFromWal(NO_NONCE, 3, 5);
    nm.reportOperationFromWal(NO_NONCE, 3, 6);
    // WAL nonces should prevent cross-server conflicts.
    assertFalse(nm.startOperation(NO_NONCE, 1, createStoppable()));
    // Make sure we ignore very old nonces, but not borderline old nonces.
    assertTrue(nm.startOperation(NO_NONCE, 2, createStoppable()));
    assertFalse(nm.startOperation(NO_NONCE, 3, createStoppable()));
    // Make sure grace period is counted from recovery time.
    edge.setValue(17);
    cleanup.choreForTesting();
    assertFalse(nm.startOperation(NO_NONCE, 1, createStoppable()));
    assertFalse(nm.startOperation(NO_NONCE, 3, createStoppable()));
    edge.setValue(19);
    cleanup.choreForTesting();
    assertTrue(nm.startOperation(NO_NONCE, 1, createStoppable()));
    assertTrue(nm.startOperation(NO_NONCE, 3, createStoppable()));
  } finally {
    EnvironmentEdgeManager.reset();
  }
}
 

/**
 * Tests that the timeOfOldestEdit is updated correctly for the
 * various edit operations in memstore.
 */
@Override
@Test
public void testUpdateToTimeOfOldestEdit() throws Exception {
  try {
    EnvironmentEdgeForMemstoreTest edge = new EnvironmentEdgeForMemstoreTest();
    EnvironmentEdgeManager.injectEdge(edge);
    long t = memstore.timeOfOldestEdit();
    assertEquals(Long.MAX_VALUE, t);

    // test the case that the timeOfOldestEdit is updated after a KV add
    memstore.add(KeyValueTestUtil.create("r", "f", "q", 100, "v"), null);
    t = memstore.timeOfOldestEdit();
    assertTrue(t == 1234);
    // The method will also assert
    // the value is reset to Long.MAX_VALUE
    t = runSnapshot(memstore, true);

    // test the case that the timeOfOldestEdit is updated after a KV delete
    memstore.add(KeyValueTestUtil.create("r", "f", "q", 100, KeyValue.Type.Delete, "v"), null);
    t = memstore.timeOfOldestEdit();
    assertTrue(t == 1234);
   t = runSnapshot(memstore, true);

    // test the case that the timeOfOldestEdit is updated after a KV upsert
    List<Cell> l = new ArrayList<>();
    KeyValue kv1 = KeyValueTestUtil.create("r", "f", "q", 100, "v");
    kv1.setSequenceId(100);
    l.add(kv1);
    memstore.upsert(l, 1000, null);
    t = memstore.timeOfOldestEdit();
    assertTrue(t == 1234);
  } finally {
    EnvironmentEdgeManager.reset();
  }
}
 

@AfterClass
public static void tearDownAfterClass() throws Exception {
  TEST_UTIL.shutdownMiniCluster();
  EnvironmentEdgeManager.reset();
}
 

@AfterClass
public static void tearDownAfterClass() throws Exception {
  TEST_UTIL.shutdownMiniCluster();
  EnvironmentEdgeManager.reset();
}
 

@Test
public void testMastersSystemTimeIsUsedInMergeRegions() throws IOException {
  long regionId = System.currentTimeMillis();

  RegionInfo regionInfoA = RegionInfoBuilder.newBuilder(TableName.valueOf(name.getMethodName()))
    .setStartKey(HConstants.EMPTY_START_ROW).setEndKey(new byte[] { 'a' }).setSplit(false)
    .setRegionId(regionId).setReplicaId(0).build();

  RegionInfo regionInfoB = RegionInfoBuilder.newBuilder(TableName.valueOf(name.getMethodName()))
    .setStartKey(new byte[] { 'a' }).setEndKey(HConstants.EMPTY_END_ROW).setSplit(false)
    .setRegionId(regionId).setReplicaId(0).build();
  RegionInfo mergedRegionInfo =
    RegionInfoBuilder.newBuilder(TableName.valueOf(name.getMethodName()))
      .setStartKey(HConstants.EMPTY_START_ROW).setEndKey(HConstants.EMPTY_END_ROW).setSplit(false)
      .setRegionId(regionId).setReplicaId(0).build();

  ServerName sn = ServerName.valueOf("bar", 0, 0);
  try (Table meta = MetaTableAccessor.getMetaHTable(connection)) {
    List<RegionInfo> regionInfos = Lists.newArrayList(regionInfoA, regionInfoB);
    MetaTableAccessor.addRegionsToMeta(connection, regionInfos, 1);

    // write the serverName column with a big current time, but set the masters time as even
    // bigger. When region merge deletes the rows for regionA and regionB, the serverName columns
    // should not be seen by the following get
    long serverNameTime = EnvironmentEdgeManager.currentTime() + 100000000;
    long masterSystemTime = EnvironmentEdgeManager.currentTime() + 123456789;

    // write the serverName columns
    MetaTableAccessor.updateRegionLocation(connection, regionInfoA, sn, 1, serverNameTime);

    // assert that we have the serverName column with expected ts
    Get get = new Get(mergedRegionInfo.getRegionName());
    Result result = meta.get(get);
    Cell serverCell = result.getColumnLatestCell(HConstants.CATALOG_FAMILY,
      CatalogFamilyFormat.getServerColumn(0));
    assertNotNull(serverCell);
    assertEquals(serverNameTime, serverCell.getTimestamp());

    ManualEnvironmentEdge edge = new ManualEnvironmentEdge();
    edge.setValue(masterSystemTime);
    EnvironmentEdgeManager.injectEdge(edge);
    try {
      // now merge the regions, effectively deleting the rows for region a and b.
      MetaTableAccessor.mergeRegions(connection, mergedRegionInfo,
        getMapOfRegionsToSeqNum(regionInfoA, regionInfoB), sn, 1);
    } finally {
      EnvironmentEdgeManager.reset();
    }

    result = meta.get(get);
    serverCell = result.getColumnLatestCell(HConstants.CATALOG_FAMILY,
      CatalogFamilyFormat.getServerColumn(0));
    Cell startCodeCell = result.getColumnLatestCell(HConstants.CATALOG_FAMILY,
      CatalogFamilyFormat.getStartCodeColumn(0));
    Cell seqNumCell = result.getColumnLatestCell(HConstants.CATALOG_FAMILY,
      CatalogFamilyFormat.getSeqNumColumn(0));
    assertNull(serverCell);
    assertNull(startCodeCell);
    assertNull(seqNumCell);
  }
}
 

/**
 * Test that connection can become idle without breaking everything.
 */
@Test
public void testConnectionIdle() throws Exception {
  final TableName tableName = TableName.valueOf(name.getMethodName());
  TEST_UTIL.createTable(tableName, FAM_NAM).close();
  int idleTime = 20000;
  boolean previousBalance = TEST_UTIL.getAdmin().balancerSwitch(false, true);

  Configuration c2 = new Configuration(TEST_UTIL.getConfiguration());
  // We want to work on a separate connection.
  c2.set(HConstants.HBASE_CLIENT_INSTANCE_ID, String.valueOf(-1));
  c2.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 1); // Don't retry: retry = test failed
  c2.setInt(RpcClient.IDLE_TIME, idleTime);

  Connection connection = ConnectionFactory.createConnection(c2);
  final Table table = connection.getTable(tableName);

  Put put = new Put(ROW);
  put.addColumn(FAM_NAM, ROW, ROW);
  table.put(put);

  ManualEnvironmentEdge mee = new ManualEnvironmentEdge();
  mee.setValue(System.currentTimeMillis());
  EnvironmentEdgeManager.injectEdge(mee);
  LOG.info("first get");
  table.get(new Get(ROW));

  LOG.info("first get - changing the time & sleeping");
  mee.incValue(idleTime + 1000);
  Thread.sleep(1500); // we need to wait a little for the connection to be seen as idle.
                      // 1500 = sleep time in RpcClient#waitForWork + a margin

  LOG.info("second get - connection has been marked idle in the middle");
  // To check that the connection actually became idle would need to read some private
  // fields of RpcClient.
  table.get(new Get(ROW));
  mee.incValue(idleTime + 1000);

  LOG.info("third get - connection is idle, but the reader doesn't know yet");
  // We're testing here a special case:
  // time limit reached BUT connection not yet reclaimed AND a new call.
  // in this situation, we don't close the connection, instead we use it immediately.
  // If we're very unlucky we can have a race condition in the test: the connection is already
  // under closing when we do the get, so we have an exception, and we don't retry as the
  // retry number is 1. The probability is very very low, and seems acceptable for now. It's
  // a test issue only.
  table.get(new Get(ROW));

  LOG.info("we're done - time will change back");

  table.close();

  connection.close();
  EnvironmentEdgeManager.reset();
  TEST_UTIL.getAdmin().balancerSwitch(previousBalance, true);
}
 

@Test
public void testExtremeLimiters() throws InterruptedException {

  ManualEnvironmentEdge testEdge = new ManualEnvironmentEdge();
  EnvironmentEdgeManager.injectEdge(testEdge);
  long limit = Long.MAX_VALUE - 1;

  RateLimiter avgLimiter = new AverageIntervalRateLimiter();
  avgLimiter.set(limit, TimeUnit.SECONDS);
  RateLimiter fixLimiter = new FixedIntervalRateLimiter();
  fixLimiter.set(limit, TimeUnit.SECONDS);

  assertEquals(limit, avgLimiter.getAvailable());
  assertEquals(limit, fixLimiter.getAvailable());

  assertTrue(avgLimiter.canExecute(limit / 2));
  avgLimiter.consume(limit / 2);

  assertTrue(fixLimiter.canExecute(limit / 2));
  fixLimiter.consume(limit / 2);

  // Make sure that available is whatever left
  assertTrue((limit - (limit / 2)) == avgLimiter.getAvailable());
  assertTrue((limit - (limit / 2)) == fixLimiter.getAvailable());

  // after 100 millseconds, both should not be able to execute the limit
  testEdge.incValue(100);

  assertFalse(avgLimiter.canExecute(limit));
  assertFalse(fixLimiter.canExecute(limit));

  // after 500 millseconds, average interval limiter should be able to execute the limit
  testEdge.incValue(500);
  assertTrue(avgLimiter.canExecute(limit));
  assertFalse(fixLimiter.canExecute(limit));

  // Make sure that available is correct
  assertTrue(limit == avgLimiter.getAvailable());
  assertTrue((limit - (limit / 2)) == fixLimiter.getAvailable());

  // after 500 millseconds, both should be able to execute
  testEdge.incValue(500);
  assertTrue(avgLimiter.canExecute(limit));
  assertTrue(fixLimiter.canExecute(limit));

  // Make sure that available is Long.MAX_VALUE
  assertTrue(limit == avgLimiter.getAvailable());
  assertTrue(limit == fixLimiter.getAvailable());

  EnvironmentEdgeManager.reset();
}
 

@After
public void tearDown() throws Exception {
  EnvironmentEdgeManager.reset();
}
 

@AfterClass
public static void resetEnvironmentEdge() throws Exception {
  TEST_UTIL.shutdownMiniCluster();
  EnvironmentEdgeManager.reset();
}
 

/**
 * Cleanup the test based on the passed state.
 * @param state
 */
private void cleanup(TestState state) throws IOException {
  EnvironmentEdgeManager.reset();
  state.table.close();
  UTIL.deleteTable(state.table.getTableName());
}
 

@AfterClass
public static void teardown() {
  EnvironmentEdgeManager.reset();
}