HBASE-24079 [Flakey Tests] Misc fixes and debug; fix BindException in…

hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientAsyncPrefetchScanner.java

@@ -17,10 +17,7 @@
  */
 package org.apache.hadoop.hbase.client;
 
-import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
-
 import static org.apache.hadoop.hbase.client.ConnectionUtils.calcEstimatedSize;
-
 import java.io.IOException;
 import java.io.InterruptedIOException;
 import java.util.Queue;
@@ -32,12 +29,12 @@
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Consumer;
-
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.TableName;
-import org.apache.yetus.audience.InterfaceAudience;
 import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
 import org.apache.hadoop.hbase.util.Threads;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
 
 /**
  * ClientAsyncPrefetchScanner implements async scanner behaviour.
@@ -55,9 +52,7 @@ public class ClientAsyncPrefetchScanner extends ClientSimpleScanner {
   private long maxCacheSize;
   private AtomicLong cacheSizeInBytes;
   // exception queue (from prefetch to main scan execution)
-  private Queue<Exception> exceptionsQueue;
-  // prefetch thread to be executed asynchronously
-  private Thread prefetcher;
+  private final Queue<Exception> exceptionsQueue;
   // used for testing
   private Consumer<Boolean> prefetchListener;
 
@@ -71,6 +66,8 @@ public ClientAsyncPrefetchScanner(Configuration configuration, Scan scan, TableN
       int replicaCallTimeoutMicroSecondScan) throws IOException {
     super(configuration, scan, name, connection, rpcCallerFactory, rpcControllerFactory, pool,
         replicaCallTimeoutMicroSecondScan);
+    exceptionsQueue = new ConcurrentLinkedQueue<>();
+    Threads.setDaemonThreadRunning(new Thread(new PrefetchRunnable()), name + ".asyncPrefetcher");
   }
 
   @VisibleForTesting
@@ -80,13 +77,10 @@ void setPrefetchListener(Consumer<Boolean> prefetchListener) {
 
   @Override
   protected void initCache() {
-    // concurrent cache
-    maxCacheSize = resultSize2CacheSize(maxScannerResultSize);
+    // Override to put a different cache in place of the super's -- a concurrent one.
     cache = new LinkedBlockingQueue<>();
+    maxCacheSize = resultSize2CacheSize(maxScannerResultSize);
     cacheSizeInBytes = new AtomicLong(0);
-    exceptionsQueue = new ConcurrentLinkedQueue<>();
-    prefetcher = new Thread(new PrefetchRunnable());
-    Threads.setDaemonThreadRunning(prefetcher, tableName + ".asyncPrefetcher");
   }
 
   private long resultSize2CacheSize(long maxResultSize) {

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java

@@ -174,9 +174,13 @@ public class BucketCache implements BlockCache, HeapSize {
   private final AtomicLong accessCount = new AtomicLong();
 
   private static final int DEFAULT_CACHE_WAIT_TIME = 50;
-  // Used in test now. If the flag is false and the cache speed is very fast,
-  // bucket cache will skip some blocks when caching. If the flag is true, we
-  // will wait blocks flushed to IOEngine for some time when caching
+
+  /**
+   * Used in tests. If this flag is false and the cache speed is very fast,
+   * bucket cache will skip some blocks when caching. If the flag is true, we
+   * will wait until blocks are flushed to IOEngine.
+   */
+  @VisibleForTesting
   boolean wait_when_cache = false;
 
   private final BucketCacheStats cacheStats = new BucketCacheStats();

hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/RSProcedureDispatcher.java

@@ -39,12 +39,10 @@
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-
 import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
 import org.apache.hbase.thirdparty.com.google.common.collect.ArrayListMultimap;
 import org.apache.hbase.thirdparty.com.google.protobuf.ByteString;
 import org.apache.hbase.thirdparty.com.google.protobuf.ServiceException;
-
 import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
 import org.apache.hadoop.hbase.shaded.protobuf.RequestConverter;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.AdminService;
@@ -95,19 +93,23 @@ public boolean start() {
     if (!super.start()) {
       return false;
     }
+    if (master.isStopped()) {
+      LOG.debug("Stopped");
+      return false;
+    }
     // Around startup, if failed, some of the below may be set back to null so NPE is possible.
     ServerManager sm = master.getServerManager();
     if (sm == null) {
-      LOG.debug("ServerManager is null; stopping={}", master.isStopping());
+      LOG.debug("ServerManager is null");
       return false;
     }
     sm.registerListener(this);
     ProcedureExecutor<MasterProcedureEnv> pe = master.getMasterProcedureExecutor();
     if (pe == null) {
-      LOG.debug("ProcedureExecutor is null; stopping={}", master.isStopping());
+      LOG.debug("ProcedureExecutor is null");
       return false;
     }
-    procedureEnv = pe.getEnvironment();
+    this.procedureEnv = pe.getEnvironment();
     if (this.procedureEnv == null) {
       LOG.debug("ProcedureEnv is null; stopping={}", master.isStopping());
       return false;

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactSplit.java

@@ -20,7 +20,6 @@
 
 import static org.apache.hadoop.hbase.regionserver.Store.NO_PRIORITY;
 import static org.apache.hadoop.hbase.regionserver.Store.PRIORITY_USER;
-
 import java.io.IOException;
 import java.io.PrintWriter;
 import java.io.StringWriter;
@@ -35,7 +34,6 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.IntSupplier;
-
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.conf.ConfigurationManager;
 import org.apache.hadoop.hbase.conf.PropagatingConfigurationObserver;
@@ -55,7 +53,6 @@
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-
 import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
 import org.apache.hbase.thirdparty.com.google.common.base.Preconditions;
 import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
@@ -103,7 +100,6 @@ public class CompactSplit implements CompactionRequester, PropagatingConfigurati
    */
   private int regionSplitLimit;
 
-  /** @param server */
   CompactSplit(HRegionServer server) {
     this.server = server;
     this.conf = server.getConfiguration();
@@ -192,12 +188,19 @@ public String dumpQueue() {
 
   public synchronized boolean requestSplit(final Region r) {
     // don't split regions that are blocking
-    if (shouldSplitRegion() && ((HRegion)r).getCompactPriority() >= PRIORITY_USER) {
-      byte[] midKey = ((HRegion)r).checkSplit();
-      if (midKey != null) {
-        requestSplit(r, midKey);
-        return true;
+    HRegion hr = (HRegion)r;
+    try {
+      if (shouldSplitRegion() && hr.getCompactPriority() >= PRIORITY_USER) {
+        byte[] midKey = hr.checkSplit();
+        if (midKey != null) {
+          requestSplit(r, midKey);
+          return true;
+        }
       }
+    } catch (IndexOutOfBoundsException e) {
+      // We get this sometimes. Not sure why. Catch and return false; no split request.
+      LOG.warn("Catching out-of-bounds; region={}, policy={}", hr == null? null: hr.getRegionInfo(),
+        hr == null? "null": hr.getCompactPriority(), e);
     }
     return false;
   }
@@ -244,8 +247,7 @@ default void completed(Store store) {
   }
 
   private static final CompactionCompleteTracker DUMMY_COMPLETE_TRACKER =
-      new CompactionCompleteTracker() {
-      };
+    new CompactionCompleteTracker() {};
 
   private static final class AggregatingCompleteTracker implements CompactionCompleteTracker {
 
@@ -340,7 +342,8 @@ private void requestCompactionInternal(HRegion region, HStore store, String why,
 
     CompactionContext compaction;
     if (selectNow) {
-      Optional<CompactionContext> c = selectCompaction(region, store, priority, tracker, completeTracker, user);
+      Optional<CompactionContext> c =
+        selectCompaction(region, store, priority, tracker, completeTracker, user);
       if (!c.isPresent()) {
         // message logged inside
         return;
@@ -650,8 +653,8 @@ private void doCompaction(User user) {
     @Override
     public void run() {
       Preconditions.checkNotNull(server);
-      if (server.isStopped()
-          || (region.getTableDescriptor() != null && !region.getTableDescriptor().isCompactionEnabled())) {
+      if (server.isStopped() || (region.getTableDescriptor() != null &&
+          !region.getTableDescriptor().isCompactionEnabled())) {
         region.decrementCompactionsQueuedCount();
         return;
       }

hbase-server/src/test/java/org/apache/hadoop/hbase/TestCachedClusterId.java

@@ -78,7 +78,7 @@ public void testMultiThreadedGetClusterId() throws Exception {
     Configuration conf = TEST_UTIL.getConfiguration();
     CachedClusterId cachedClusterId = new CachedClusterId(conf);
     TestContext context = new TestContext(conf);
-    int numThreads = 100;
+    int numThreads = 16;
     for (int i = 0; i < numThreads; i++) {
       context.addThread(new GetClusterIdThread(context, cachedClusterId));
     }

hbase-server/src/test/java/org/apache/hadoop/hbase/TestServerSideScanMetricsFromClientSide.java

@@ -19,7 +19,6 @@
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
-
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
@@ -46,9 +45,13 @@
 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)
 public class TestServerSideScanMetricsFromClientSide {
+  private static final Logger LOG =
+    LoggerFactory.getLogger(TestServerSideScanMetricsFromClientSide.class);
 
   @ClassRule
   public static final HBaseClassTestRule CLASS_RULE =
@@ -178,22 +181,27 @@ private void testRowsSeenMetric(boolean async) throws Exception {
     baseScan = new Scan();
     baseScan.setScanMetricsEnabled(true);
     baseScan.setAsyncPrefetch(async);
-    testRowsSeenMetric(baseScan);
-
-    // Test case that only a single result will be returned per RPC to the serer
-    baseScan.setCaching(1);
-    testRowsSeenMetric(baseScan);
-
-    // Test case that partial results are returned from the server. At most one cell will be
-    // contained in each response
-    baseScan.setMaxResultSize(1);
-    testRowsSeenMetric(baseScan);
-
-    // Test case that size limit is set such that a few cells are returned per partial result from
-    // the server
-    baseScan.setCaching(NUM_ROWS);
-    baseScan.setMaxResultSize(getCellHeapSize() * (NUM_COLS - 1));
-    testRowsSeenMetric(baseScan);
+    try {
+      testRowsSeenMetric(baseScan);
+
+      // Test case that only a single result will be returned per RPC to the serer
+      baseScan.setCaching(1);
+      testRowsSeenMetric(baseScan);
+
+      // Test case that partial results are returned from the server. At most one cell will be
+      // contained in each response
+      baseScan.setMaxResultSize(1);
+      testRowsSeenMetric(baseScan);
+
+      // Test case that size limit is set such that a few cells are returned per partial result from
+      // the server
+      baseScan.setCaching(NUM_ROWS);
+      baseScan.setMaxResultSize(getCellHeapSize() * (NUM_COLS - 1));
+      testRowsSeenMetric(baseScan);
+    } catch (Throwable t) {
+      LOG.error("FAIL", t);
+      throw t;
+    }
   }
 
   public void testRowsSeenMetric(Scan baseScan) throws Exception {
@@ -212,7 +220,8 @@ public void testRowsSeenMetric(Scan baseScan) throws Exception {
       scan = new Scan(baseScan);
       scan.withStartRow(ROWS[i - 1]);
       scan.withStopRow(ROWS[ROWS.length - 1]);
-      testMetric(scan, ServerSideScanMetrics.COUNT_OF_ROWS_SCANNED_KEY_METRIC_NAME, ROWS.length - i);
+      testMetric(scan, ServerSideScanMetrics.COUNT_OF_ROWS_SCANNED_KEY_METRIC_NAME,
+        ROWS.length - i);
     }
 
     // The filter should filter out all rows, but we still expect to see every row.
@@ -318,8 +327,11 @@ public void testRowsFilteredMetric(Scan baseScan) throws Exception {
   public void testRowsFilteredMetric(Scan baseScan, Filter filter, int expectedNumFiltered)
       throws Exception {
     Scan scan = new Scan(baseScan);
-    if (filter != null) scan.setFilter(filter);
-    testMetric(scan, ServerSideScanMetrics.COUNT_OF_ROWS_FILTERED_KEY_METRIC_NAME, expectedNumFiltered);
+    if (filter != null) {
+      scan.setFilter(filter);
+    }
+    testMetric(scan, ServerSideScanMetrics.COUNT_OF_ROWS_FILTERED_KEY_METRIC_NAME,
+      expectedNumFiltered);
   }
 
   /**
@@ -334,7 +346,7 @@ public void testMetric(Scan scan, String metricKey, long expectedValue) throws E
     ResultScanner scanner = TABLE.getScanner(scan);
     // Iterate through all the results
     while (scanner.next() != null) {
-
+      continue;
     }
     scanner.close();
     ScanMetrics metrics = scanner.getScanMetrics();

hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncRegionAdminApi.java

@@ -24,7 +24,6 @@
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
-
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
@@ -410,11 +409,11 @@ private void compactionTest(final TableName tableName, final int flushes,
     }
 
     long curt = System.currentTimeMillis();
-    long waitTime = 5000;
+    long waitTime = 10000;
     long endt = curt + waitTime;
     CompactionState state = admin.getCompactionState(tableName).get();
     while (state == CompactionState.NONE && curt < endt) {
-      Thread.sleep(10);
+      Thread.sleep(1);
       state = admin.getCompactionState(tableName).get();
       curt = System.currentTimeMillis();
     }

hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestFromClientSide3.java

@@ -234,19 +234,20 @@ public void testScanAfterDeletingSpecifiedRowV2() throws IOException, Interrupte
       byte[] row = Bytes.toBytes("SpecifiedRow");
       byte[] qual0 = Bytes.toBytes("qual0");
       byte[] qual1 = Bytes.toBytes("qual1");
-      Delete d = new Delete(row);
+      long now = System.currentTimeMillis();
+      Delete d = new Delete(row, now);
       table.delete(d);
 
       Put put = new Put(row);
-      put.addColumn(FAMILY, null, VALUE);
+      put.addColumn(FAMILY, null, now + 1, VALUE);
       table.put(put);
 
       put = new Put(row);
-      put.addColumn(FAMILY, qual1, qual1);
+      put.addColumn(FAMILY, qual1, now + 2, qual1);
       table.put(put);
 
       put = new Put(row);
-      put.addColumn(FAMILY, qual0, qual0);
+      put.addColumn(FAMILY, qual0, now + 3, qual0);
       table.put(put);
 
       Result r = table.get(new Get(row));

hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestMasterRegistry.java

@@ -103,7 +103,17 @@ private static String generateDummyMastersList(int size) {
   @Test public void testRegistryRPCs() throws Exception {
     Configuration conf = new Configuration(TEST_UTIL.getConfiguration());
     HMaster activeMaster = TEST_UTIL.getHBaseCluster().getMaster();
-    for (int numHedgedReqs = 1; numHedgedReqs <=3; numHedgedReqs++) {
+    final int size = activeMaster.getMetaRegionLocationCache().
+      getMetaRegionLocations().get().size();
+    // Add wait on all replicas being assigned before proceeding w/ test. Failed on occasion
+    // because not all replicas had made it up before test started.
+    TEST_UTIL.waitFor(10000,
+      () -> {
+        try (MasterRegistry registry = new MasterRegistry(conf)) {
+          return registry.getMetaRegionLocations().get().size() >= size;
+        }
+      });
+    for (int numHedgedReqs = 1; numHedgedReqs <= 3; numHedgedReqs++) {
       if (numHedgedReqs == 1) {
         conf.setBoolean(HConstants.MASTER_REGISTRY_ENABLE_HEDGED_READS_KEY, false);
       } else {

hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestScannersFromClientSide.java

@@ -935,10 +935,11 @@ public void testReadExpiredDataForRawScan() throws IOException {
   @Test
   public void testScanWithColumnsAndFilterAndVersion() throws IOException {
     TableName tableName = name.getTableName();
+    long now = System.currentTimeMillis();
     try (Table table = TEST_UTIL.createTable(tableName, FAMILY, 4)) {
       for (int i = 0; i < 4; i++) {
         Put put = new Put(ROW);
-        put.addColumn(FAMILY, QUALIFIER, VALUE);
+        put.addColumn(FAMILY, QUALIFIER, now + i, VALUE);
         table.put(put);
       }