hbase.storescanner.parallel.seek.threads
10
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ScanInfo.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ScanInfo.java
index b0c497110328..d87ec0f02d41 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ScanInfo.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ScanInfo.java
@@ -45,6 +45,7 @@ public class ScanInfo {
private boolean usePread;
private long cellsPerTimeoutCheck;
private boolean parallelSeekEnabled;
+ private boolean adaptiveParallelSeekEnabled;
private final long preadMaxBytes;
private final boolean newVersionBehavior;
@@ -92,13 +93,14 @@ public ScanInfo(Configuration conf, byte[] family, int minVersions, int maxVersi
conf.getLong(HConstants.TABLE_MAX_ROWSIZE_KEY, HConstants.TABLE_MAX_ROWSIZE_DEFAULT),
conf.getBoolean("hbase.storescanner.use.pread", false), getCellsPerTimeoutCheck(conf),
conf.getBoolean(StoreScanner.STORESCANNER_PARALLEL_SEEK_ENABLE, false),
- conf.getLong(StoreScanner.STORESCANNER_PREAD_MAX_BYTES, 4 * blockSize), newVersionBehavior);
+ conf.getLong(StoreScanner.STORESCANNER_PREAD_MAX_BYTES, 4 * blockSize), newVersionBehavior,
+ conf.getBoolean(StoreScanner.STORESCANNER_ADAPTIVE_PARALLEL_SEEK_ENABLE, false));
}
private ScanInfo(byte[] family, int minVersions, int maxVersions, long ttl,
KeepDeletedCells keepDeletedCells, long timeToPurgeDeletes, CellComparator comparator,
long tableMaxRowSize, boolean usePread, long cellsPerTimeoutCheck, boolean parallelSeekEnabled,
- long preadMaxBytes, boolean newVersionBehavior) {
+ long preadMaxBytes, boolean newVersionBehavior, boolean adaptiveParallelSeekEnabled) {
this.family = family;
this.minVersions = minVersions;
this.maxVersions = maxVersions;
@@ -112,6 +114,7 @@ private ScanInfo(byte[] family, int minVersions, int maxVersions, long ttl,
this.parallelSeekEnabled = parallelSeekEnabled;
this.preadMaxBytes = preadMaxBytes;
this.newVersionBehavior = newVersionBehavior;
+ this.adaptiveParallelSeekEnabled = adaptiveParallelSeekEnabled;
}
long getTableMaxRowSize() {
@@ -130,6 +133,10 @@ boolean isParallelSeekEnabled() {
return this.parallelSeekEnabled;
}
+ boolean isAdaptiveParallelSeekEnabled() {
+ return this.adaptiveParallelSeekEnabled;
+ }
+
public byte[] getFamily() {
return family;
}
@@ -181,7 +188,7 @@ ScanInfo customize(int maxVersions, long ttl, KeepDeletedCells keepDeletedCells,
long timeToPurgeDeletes) {
return new ScanInfo(family, minVersions, maxVersions, ttl, keepDeletedCells, timeToPurgeDeletes,
comparator, tableMaxRowSize, usePread, cellsPerTimeoutCheck, parallelSeekEnabled,
- preadMaxBytes, newVersionBehavior);
+ preadMaxBytes, newVersionBehavior, adaptiveParallelSeekEnabled);
}
@Override
@@ -192,6 +199,7 @@ public String toString() {
.append("tableMaxRowSize", tableMaxRowSize).append("usePread", usePread)
.append("cellsPerTimeoutCheck", cellsPerTimeoutCheck)
.append("parallelSeekEnabled", parallelSeekEnabled).append("preadMaxBytes", preadMaxBytes)
- .append("newVersionBehavior", newVersionBehavior).toString();
+ .append("newVersionBehavior", newVersionBehavior)
+ .append("adaptiveParallelSeekEnabled", adaptiveParallelSeekEnabled).toString();
}
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
index 86752f27a0f6..e289db162e4f 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
@@ -27,6 +27,7 @@
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.IntConsumer;
@@ -46,6 +47,7 @@
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.conf.ConfigKey;
import org.apache.hadoop.hbase.executor.ExecutorService;
+import org.apache.hadoop.hbase.executor.ExecutorType;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.ipc.RpcCall;
import org.apache.hadoop.hbase.ipc.RpcServer;
@@ -95,6 +97,7 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
* A flag that enables StoreFileScanner parallel-seeking
*/
private boolean parallelSeekEnabled = false;
+ private boolean adaptiveParallelSeekEnabled = false;
private ExecutorService executor;
private final Scan scan;
private final long oldestUnexpiredTS;
@@ -129,6 +132,8 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
static final boolean LAZY_SEEK_ENABLED_BY_DEFAULT = true;
public static final String STORESCANNER_PARALLEL_SEEK_ENABLE =
"hbase.storescanner.parallel.seek.enable";
+ public static final String STORESCANNER_ADAPTIVE_PARALLEL_SEEK_ENABLE =
+ "hbase.storescanner.adaptive.parallel.seek.enable";
/** Used during unit testing to ensure that lazy seek does save seek ops */
private static boolean lazySeekEnabledGlobally = LAZY_SEEK_ENABLED_BY_DEFAULT;
@@ -232,6 +237,8 @@ private StoreScanner(HStore store, Scan scan, ScanInfo scanInfo, int numColumns,
RegionServerServices rsService = store.getHRegion().getRegionServerServices();
if (rsService != null && scanInfo.isParallelSeekEnabled()) {
this.parallelSeekEnabled = true;
+ this.adaptiveParallelSeekEnabled =
+ this.parallelSeekEnabled && scanInfo.isAdaptiveParallelSeekEnabled();
this.executor = rsService.getExecutorService();
}
}
@@ -441,6 +448,8 @@ protected void seekScanners(List scanners, ExtendedCe
totalScannersSoughtBytes += PrivateCellUtil.estimatedSerializedSizeOf(c);
}
}
+ } else if (adaptiveParallelSeekEnabled) {
+ adaptiveParallelSeek(scanners, seekKey);
} else {
parallelSeek(scanners, seekKey);
}
@@ -1268,6 +1277,135 @@ private void parallelSeek(final List scanners, final
}
}
+ /**
+ * Returns the number of threads available for immediate execution in the parallel seek thread
+ * pool. Uses a conservative approach: only reports capacity when the task queue is empty AND
+ * active threads < pool size.
+ * @return number of threads available for immediate execution, or 0 if saturated
+ */
+ private int getAvailableParallelSeekCapacity() {
+ ThreadPoolExecutor pool = executor.getExecutorThreadPool(ExecutorType.RS_PARALLEL_SEEK);
+ if (!pool.getQueue().isEmpty()) {
+ return 0; // Conservative: any queued work means saturated
+ }
+ return Math.max(0, pool.getCorePoolSize() - pool.getActiveCount());
+ }
+
+ /**
+ * Seeks scanners using an adaptive strategy that switches between parallel and sequential
+ * execution based on thread pool availability.
+ *
+ * When the parallel seek thread pool is saturated, falls back to sequential seeking. After each
+ * sequential seek, re-checks capacity and opportunistically submits remaining scanners for
+ * parallel execution when slots become available.
+ *
+ * If an IOException occurs during an inline seek, we must wait for any already-submitted handlers
+ * to complete before propagating the error. This prevents the caller from closing scanners that
+ * are still being used by worker threads.
+ *
+ * @param scanners list of KeyValueScanners to seek
+ * @param kv the key to seek to
+ * @throws IOException if any seek operation fails
+ */
+ private void adaptiveParallelSeek(final List scanners,
+ final ExtendedCell kv) throws IOException {
+ if (scanners.isEmpty()) return;
+
+ int scannerCount = scanners.size();
+ // Pre-count StoreFileScanners to size the latch correctly
+ int storeFileScannerCount = 0;
+ for (KeyValueScanner scanner : scanners) {
+ if (scanner instanceof StoreFileScanner) {
+ storeFileScannerCount++;
+ }
+ }
+ CountDownLatch latch = new CountDownLatch(storeFileScannerCount);
+ List handlers = new ArrayList<>(storeFileScannerCount);
+ int index = 0;
+ IOException inlineSeekError = null;
+
+ while (index < scannerCount) {
+ int capacity = getAvailableParallelSeekCapacity();
+
+ if (capacity == 0) {
+ // Sequential fallback: process one scanner on calling thread
+ KeyValueScanner scanner = scanners.get(index);
+ try {
+ scanner.seek(kv);
+ } catch (IOException e) {
+ // Must wait for already-submitted handlers before propagating error
+ inlineSeekError = e;
+ if (scanner instanceof StoreFileScanner) {
+ latch.countDown();
+ }
+ index++;
+ break;
+ }
+ if (scanner instanceof StoreFileScanner) {
+ latch.countDown();
+ }
+ index++;
+ } else {
+ // Opportunistic parallel: submit batch up to available capacity
+ int batchEnd = Math.min(index + capacity, scannerCount);
+ for (int i = index; i < batchEnd; i++) {
+ KeyValueScanner scanner = scanners.get(i);
+ if (scanner instanceof StoreFileScanner) {
+ ParallelSeekHandler seekHandler =
+ new ParallelSeekHandler(scanner, kv, this.readPt, latch);
+ executor.submit(seekHandler);
+ handlers.add(seekHandler);
+ } else {
+ try {
+ scanner.seek(kv);
+ } catch (IOException e) {
+ // Must wait for already-submitted handlers before propagating error
+ inlineSeekError = e;
+ // Count down latch for remaining StoreFileScanners in this batch that won't be
+ // processed
+ for (int j = i + 1; j < batchEnd; j++) {
+ if (scanners.get(j) instanceof StoreFileScanner) {
+ latch.countDown();
+ }
+ }
+ index = batchEnd;
+ break;
+ }
+ }
+ }
+ if (inlineSeekError != null) {
+ break;
+ }
+ index = batchEnd;
+ }
+ }
+
+ // Count down latch for any remaining unprocessed StoreFileScanners
+ for (int i = index; i < scannerCount; i++) {
+ if (scanners.get(i) instanceof StoreFileScanner) {
+ latch.countDown();
+ }
+ }
+
+ try {
+ latch.await();
+ } catch (InterruptedException ie) {
+ throw (InterruptedIOException) new InterruptedIOException().initCause(ie);
+ }
+
+ // Check for errors from parallel handlers first
+ for (ParallelSeekHandler handler : handlers) {
+ if (handler.getErr() != null) {
+ throw new IOException(handler.getErr());
+ }
+ }
+
+ // Propagate inline seek error after all handlers have completed
+ if (inlineSeekError != null) {
+ throw inlineSeekError;
+ }
+ }
+
/**
* Used in testing.
* @return all scanners in no particular order
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreScanner.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreScanner.java
index f661e17e6ac7..edce01770a0d 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreScanner.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreScanner.java
@@ -21,12 +21,15 @@
import static org.apache.hadoop.hbase.regionserver.KeyValueScanFixture.scanFixture;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -34,6 +37,8 @@
import java.util.NavigableSet;
import java.util.Set;
import java.util.TreeSet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
@@ -51,12 +56,18 @@
import org.apache.hadoop.hbase.KeepDeletedCells;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.PrivateCellUtil;
+import org.apache.hadoop.hbase.Server;
import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.Waiter;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.RegionInfo;
import org.apache.hadoop.hbase.client.RegionInfoBuilder;
import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.executor.EventHandler;
+import org.apache.hadoop.hbase.executor.EventType;
+import org.apache.hadoop.hbase.executor.ExecutorService;
+import org.apache.hadoop.hbase.executor.ExecutorType;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.ColumnCountGetFilter;
import org.apache.hadoop.hbase.filter.Filter;
@@ -1303,4 +1314,543 @@ public void testGetFilesReadOnInitializationFailure() throws Exception {
Mockito.verify(mockScanner2, Mockito.never()).getFilesRead();
}
+ // =========================================================================
+ // Helper infrastructure for adaptive parallel seek tests
+ // =========================================================================
+
+ /**
+ * Creates a ScanInfo with both parallelSeekEnabled and adaptiveParallelSeekEnabled set to true.
+ */
+ private static ScanInfo adaptiveScanInfo() {
+ Configuration conf = HBaseConfiguration.create();
+ conf.setBoolean(StoreScanner.STORESCANNER_PARALLEL_SEEK_ENABLE, true);
+ conf.setBoolean(StoreScanner.STORESCANNER_ADAPTIVE_PARALLEL_SEEK_ENABLE, true);
+ return new ScanInfo(conf, CF, 0, Integer.MAX_VALUE, Long.MAX_VALUE, KeepDeletedCells.FALSE,
+ HConstants.DEFAULT_BLOCKSIZE, 0, CellComparator.getInstance(), false);
+ }
+
+ /**
+ * Creates a ScanInfo with parallelSeekEnabled=true but adaptiveParallelSeekEnabled=false.
+ */
+ private static ScanInfo parallelOnlyScanInfo() {
+ Configuration conf = HBaseConfiguration.create();
+ conf.setBoolean(StoreScanner.STORESCANNER_PARALLEL_SEEK_ENABLE, true);
+ conf.setBoolean(StoreScanner.STORESCANNER_ADAPTIVE_PARALLEL_SEEK_ENABLE, false);
+ return new ScanInfo(conf, CF, 0, Integer.MAX_VALUE, Long.MAX_VALUE, KeepDeletedCells.FALSE,
+ HConstants.DEFAULT_BLOCKSIZE, 0, CellComparator.getInstance(), false);
+ }
+
+ /**
+ * Starts an ExecutorService with an RS_PARALLEL_SEEK pool of the given size.
+ */
+ private static ExecutorService startParallelSeekExecutor(int poolSize) {
+ ExecutorService exec = new ExecutorService("test_adaptive_" + System.nanoTime());
+ exec.startExecutorService(exec.new ExecutorConfig()
+ .setExecutorType(ExecutorType.RS_PARALLEL_SEEK).setCorePoolSize(poolSize));
+ return exec;
+ }
+
+ /**
+ * Uses reflection to inject adaptiveParallelSeekEnabled=true, parallelSeekEnabled=true, and the
+ * given executor into a StoreScanner instance.
+ */
+ private static void injectAdaptiveFields(StoreScanner scanner, ExecutorService executor)
+ throws Exception {
+ Field adaptiveField =
+ StoreScanner.class.getDeclaredField("adaptiveParallelSeekEnabled");
+ adaptiveField.setAccessible(true);
+ adaptiveField.set(scanner, true);
+
+ Field parallelField =
+ StoreScanner.class.getDeclaredField("parallelSeekEnabled");
+ parallelField.setAccessible(true);
+ parallelField.set(scanner, true);
+
+ Field executorField =
+ StoreScanner.class.getDeclaredField("executor");
+ executorField.setAccessible(true);
+ executorField.set(scanner, executor);
+ }
+
+ /**
+ * A simple KeyValueScanner stub that is NOT a StoreFileScanner. Records seek calls and can be
+ * configured to throw on seek.
+ */
+ private static class MockMemStoreScanner implements KeyValueScanner {
+ private final AtomicInteger seekCount = new AtomicInteger(0);
+ private final IOException seekError;
+
+ MockMemStoreScanner() {
+ this.seekError = null;
+ }
+
+ MockMemStoreScanner(IOException seekError) {
+ this.seekError = seekError;
+ }
+
+ @Override
+ public boolean seek(ExtendedCell key) throws IOException {
+ if (seekError != null) throw seekError;
+ seekCount.incrementAndGet();
+ return true;
+ }
+
+ int getSeekCount() {
+ return seekCount.get();
+ }
+
+ @Override
+ public ExtendedCell peek() {
+ return null;
+ }
+
+ @Override
+ public ExtendedCell next() {
+ return null;
+ }
+
+ @Override
+ public boolean requestSeek(ExtendedCell kv, boolean forward, boolean useBloom) {
+ return false;
+ }
+
+ @Override
+ public boolean isFileScanner() {
+ return false;
+ }
+
+ @Override
+ public boolean backwardSeek(ExtendedCell key) {
+ return false;
+ }
+
+ @Override
+ public boolean seekToPreviousRow(ExtendedCell key) {
+ return false;
+ }
+
+ @Override
+ public boolean seekToLastRow() {
+ return false;
+ }
+
+ @Override
+ public ExtendedCell getNextIndexedKey() {
+ return null;
+ }
+
+ @Override
+ public void close() {
+ }
+
+ @Override
+ public boolean shouldUseScanner(Scan scan, HStore store, long oldestUnexpiredTS) {
+ return true;
+ }
+
+ @Override
+ public boolean reseek(ExtendedCell key) throws IOException {
+ return seek(key);
+ }
+
+ @Override
+ public boolean realSeekDone() {
+ return true;
+ }
+
+ @Override
+ public void enforceSeek() {
+ }
+
+ @Override
+ public void recordBlockSize(java.util.function.IntConsumer blockSizeConsumer) {
+ }
+
+ @Override
+ public org.apache.hadoop.fs.Path getFilePath() {
+ return null;
+ }
+
+ @Override
+ public java.util.Set getFilesRead() {
+ return java.util.Collections.emptySet();
+ }
+
+ @Override
+ public void shipped() throws IOException {
+ }
+ }
+
+ /**
+ * Saturates the RS_PARALLEL_SEEK thread pool in the given ExecutorService by submitting
+ * poolSize blocking tasks. Returns a CountDownLatch that can be counted down to release all
+ * blocked tasks.
+ */
+ private static CountDownLatch saturatePool(ExecutorService exec, int poolSize,
+ Server server) throws Exception {
+ CountDownLatch blockLatch = new CountDownLatch(1);
+ CountDownLatch startedLatch = new CountDownLatch(poolSize);
+ for (int i = 0; i < poolSize; i++) {
+ exec.submit(new EventHandler(server, EventType.RS_PARALLEL_SEEK) {
+ @Override
+ public void process() throws IOException {
+ startedLatch.countDown();
+ try {
+ blockLatch.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ });
+ }
+ // Wait until all blocking tasks are actually running
+ assertTrue(startedLatch.await(10, TimeUnit.SECONDS),
+ "Pool should be saturated within 10s");
+ return blockLatch;
+ }
+
+ /** Build a simple non-null ExtendedCell to use as a seek key. */
+ private static ExtendedCell makeSeekKey() {
+ return ExtendedCellBuilderFactory.create(CellBuilderType.DEEP_COPY)
+ .setRow(Bytes.toBytes("row1")).setFamily(CF).setQualifier(Bytes.toBytes("col"))
+ .setTimestamp(1L).setType(org.apache.hadoop.hbase.Cell.Type.Put)
+ .setValue(new byte[0]).build();
+ }
+
+ // =========================================================================
+ // Adaptive parallel seek unit tests
+ // =========================================================================
+
+ /**
+ * Test 1: adaptive=true, parallel=false => sequential path is used (adaptive ignored).
+ */
+ @Test
+ public void testAdaptiveIgnoredWhenParallelDisabled() throws Exception {
+ // ScanInfo with parallel=false → adaptive flag must be ignored
+ Configuration conf = HBaseConfiguration.create();
+ conf.setBoolean(StoreScanner.STORESCANNER_PARALLEL_SEEK_ENABLE, false);
+ conf.setBoolean(StoreScanner.STORESCANNER_ADAPTIVE_PARALLEL_SEEK_ENABLE, true);
+ ScanInfo si = new ScanInfo(conf, CF, 0, Integer.MAX_VALUE, Long.MAX_VALUE,
+ KeepDeletedCells.FALSE, HConstants.DEFAULT_BLOCKSIZE, 0, CellComparator.getInstance(), false);
+
+ // parallel disabled in ScanInfo → adaptiveParallelSeekEnabled stays false in StoreScanner
+ assertFalse(si.isParallelSeekEnabled(),
+ "parallelSeekEnabled should be false");
+ // adaptive flag from ScanInfo is there but StoreScanner ctor requires store != null
+ // with store file count > 1 to set parallelSeekEnabled. So just verify via ScanInfo:
+ assertTrue(si.isAdaptiveParallelSeekEnabled(),
+ "ScanInfo adaptive flag should be true when configured");
+
+ // Build a StoreScanner with test constructor: store=null so parallelSeekEnabled=false
+ MockMemStoreScanner ms1 = new MockMemStoreScanner();
+ MockMemStoreScanner ms2 = new MockMemStoreScanner();
+ List scanners = Arrays.asList(ms1, ms2);
+
+ Scan scan = new Scan();
+ StoreScanner storeScanner = new StoreScanner(scan, si, (NavigableSet) null, scanners);
+
+ // Verify adaptiveParallelSeekEnabled is false (because store=null → parallelSeekEnabled=false)
+ java.lang.reflect.Field adaptiveField =
+ StoreScanner.class.getDeclaredField("adaptiveParallelSeekEnabled");
+ adaptiveField.setAccessible(true);
+ assertFalse((Boolean) adaptiveField.get(storeScanner),
+ "adaptiveParallelSeekEnabled must be false when parallel seek is disabled");
+
+ storeScanner.close();
+ }
+
+ /**
+ * Test 2: empty scanner list returns without error.
+ * Requirements: implied by empty list handling in adaptiveParallelSeek
+ */
+ @Test
+ public void testAdaptiveParallelSeekEmptyList() throws Exception {
+ ExecutorService exec = startParallelSeekExecutor(4);
+ try {
+ Scan scan = new Scan();
+ List empty = new ArrayList<>();
+ StoreScanner storeScanner =
+ new StoreScanner(scan, adaptiveScanInfo(), (NavigableSet) null, empty);
+ injectAdaptiveFields(storeScanner, exec);
+
+ // Call seekScanners with an empty list — should return without error
+ storeScanner.seekScanners(empty, makeSeekKey(), false, true);
+
+ storeScanner.close();
+ } finally {
+ exec.shutdown();
+ }
+ }
+
+ /**
+ * Test 3: all sequential when pool saturated (capacity always 0).
+ */
+ @Test
+ public void testAllSequentialWhenPoolSaturated() throws Exception {
+ int poolSize = 2;
+ ExecutorService exec = startParallelSeekExecutor(poolSize);
+ Server server = Mockito.mock(Server.class);
+ Mockito.when(server.getConfiguration()).thenReturn(HBaseConfiguration.create());
+
+ CountDownLatch blockLatch = saturatePool(exec, poolSize, server);
+ try {
+ // Pool is saturated. Create StoreFileScanner mocks — they must be sought sequentially.
+ StoreFileScanner sfs1 = Mockito.mock(StoreFileScanner.class);
+ StoreFileScanner sfs2 = Mockito.mock(StoreFileScanner.class);
+ Mockito.when(sfs1.seek(Mockito.any())).thenReturn(true);
+ Mockito.when(sfs2.seek(Mockito.any())).thenReturn(true);
+
+ List scanners = Arrays.asList(sfs1, sfs2);
+ Scan scan = new Scan();
+ StoreScanner storeScanner =
+ new StoreScanner(scan, adaptiveScanInfo(), (NavigableSet) null, new ArrayList<>());
+ injectAdaptiveFields(storeScanner, exec);
+
+ ExtendedCell seekKey = makeSeekKey();
+ storeScanner.seekScanners(scanners, seekKey, false, true);
+
+ // Both scanners were sought (sequentially, since pool was full)
+ Mockito.verify(sfs1, Mockito.times(1)).seek(seekKey);
+ Mockito.verify(sfs2, Mockito.times(1)).seek(seekKey);
+
+ storeScanner.close();
+ } finally {
+ blockLatch.countDown();
+ exec.shutdown();
+ }
+ }
+
+ /**
+ * Test 4: all parallel when pool fully available (capacity >= scanner count).
+ */
+ @Test
+ public void testAllParallelWhenPoolFullyAvailable() throws Exception {
+ int poolSize = 4;
+ ExecutorService exec = startParallelSeekExecutor(poolSize);
+ try {
+ // Pool is completely idle. All StoreFileScanners should be sought in parallel.
+ StoreFileScanner sfs1 = Mockito.mock(StoreFileScanner.class);
+ StoreFileScanner sfs2 = Mockito.mock(StoreFileScanner.class);
+ Mockito.when(sfs1.seek(Mockito.any())).thenReturn(true);
+ Mockito.when(sfs2.seek(Mockito.any())).thenReturn(true);
+
+ List scanners = Arrays.asList(sfs1, sfs2);
+ Scan scan = new Scan();
+ StoreScanner storeScanner =
+ new StoreScanner(scan, adaptiveScanInfo(), (NavigableSet) null, new ArrayList<>());
+ injectAdaptiveFields(storeScanner, exec);
+
+ ExtendedCell seekKey = makeSeekKey();
+ storeScanner.seekScanners(scanners, seekKey, false, true);
+
+ // Both scanners must be sought
+ Mockito.verify(sfs1, Mockito.times(1)).seek(seekKey);
+ Mockito.verify(sfs2, Mockito.times(1)).seek(seekKey);
+
+ storeScanner.close();
+ } finally {
+ exec.shutdown();
+ }
+ }
+
+ /**
+ * Test 5: all memstore scanners (no StoreFileScanner) — all seeked inline.
+ */
+ @Test
+ public void testAllMemstoreScannersSeekInline() throws Exception {
+ ExecutorService exec = startParallelSeekExecutor(4);
+ try {
+ MockMemStoreScanner ms1 = new MockMemStoreScanner();
+ MockMemStoreScanner ms2 = new MockMemStoreScanner();
+ MockMemStoreScanner ms3 = new MockMemStoreScanner();
+
+ List scanners = Arrays.asList(ms1, ms2, ms3);
+ Scan scan = new Scan();
+ StoreScanner storeScanner =
+ new StoreScanner(scan, adaptiveScanInfo(), (NavigableSet) null, new ArrayList<>());
+ injectAdaptiveFields(storeScanner, exec);
+
+ storeScanner.seekScanners(scanners, makeSeekKey(), false, true);
+
+ assertEquals(1, ms1.getSeekCount(), "ms1 should be sought once");
+ assertEquals(1, ms2.getSeekCount(), "ms2 should be sought once");
+ assertEquals(1, ms3.getSeekCount(), "ms3 should be sought once");
+
+ storeScanner.close();
+ } finally {
+ exec.shutdown();
+ }
+ }
+
+ /**
+ * Test 6: IOException from sequential seek propagates immediately.
+ */
+ @Test
+ public void testIOExceptionFromSequentialSeekPropagates() throws Exception {
+ int poolSize = 2;
+ ExecutorService exec = startParallelSeekExecutor(poolSize);
+ Server server = Mockito.mock(Server.class);
+ Mockito.when(server.getConfiguration()).thenReturn(HBaseConfiguration.create());
+
+ // Saturate the pool so we go sequential
+ CountDownLatch blockLatch = saturatePool(exec, poolSize, server);
+ try {
+ IOException expectedError = new IOException("sequential seek error");
+ MockMemStoreScanner failingScanner = new MockMemStoreScanner(expectedError);
+ MockMemStoreScanner normalScanner = new MockMemStoreScanner();
+
+ List scanners =
+ Arrays.asList(failingScanner, normalScanner);
+ Scan scan = new Scan();
+ StoreScanner storeScanner =
+ new StoreScanner(scan, adaptiveScanInfo(), (NavigableSet) null, new ArrayList<>());
+ injectAdaptiveFields(storeScanner, exec);
+
+ // Should throw the IOException from the failing scanner
+ IOException thrown = null;
+ try {
+ storeScanner.seekScanners(scanners, makeSeekKey(), false, true);
+ } catch (IOException e) {
+ thrown = e;
+ }
+
+ assertNotNull(thrown, "Should have thrown IOException");
+ // The second scanner must NOT be sought since the first threw
+ assertEquals(0, normalScanner.getSeekCount(), "normalScanner should not be sought after error");
+
+ storeScanner.close();
+ } finally {
+ blockLatch.countDown();
+ exec.shutdown();
+ }
+ }
+
+ /**
+ * Test 7: IOException from parallel seek handler propagates after await.
+ */
+ @Test
+ public void testIOExceptionFromParallelHandlerPropagates() throws Exception {
+ int poolSize = 4;
+ ExecutorService exec = startParallelSeekExecutor(poolSize);
+ try {
+ // Mock StoreFileScanner that throws on seek
+ StoreFileScanner failingSfs = Mockito.mock(StoreFileScanner.class);
+ Mockito.doThrow(new IOException("parallel seek error")).when(failingSfs)
+ .seek(Mockito.any());
+
+ List scanners = Collections.singletonList(failingSfs);
+ Scan scan = new Scan();
+ StoreScanner storeScanner =
+ new StoreScanner(scan, adaptiveScanInfo(), (NavigableSet) null, new ArrayList<>());
+ injectAdaptiveFields(storeScanner, exec);
+
+ IOException thrown = null;
+ try {
+ storeScanner.seekScanners(scanners, makeSeekKey(), false, true);
+ } catch (IOException e) {
+ thrown = e;
+ }
+
+ assertNotNull(thrown, "Should have thrown IOException from parallel handler");
+
+ storeScanner.close();
+ } finally {
+ exec.shutdown();
+ }
+ }
+
+ /**
+ * Test 8: InterruptedException during await is wrapped as InterruptedIOException.
+ */
+ @Test
+ public void testInterruptedExceptionWrappedAsInterruptedIOException() throws Exception {
+ int poolSize = 2;
+ ExecutorService exec = startParallelSeekExecutor(poolSize);
+
+ // Use a scanner that blocks to give us a chance to interrupt
+ CountDownLatch scannerBlockLatch = new CountDownLatch(1);
+ StoreFileScanner blockingSfs = Mockito.mock(StoreFileScanner.class);
+ Mockito.doAnswer(invocation -> {
+ // Block until the test interrupts the thread
+ scannerBlockLatch.await();
+ return true;
+ }).when(blockingSfs).seek(Mockito.any());
+
+ List scanners = Arrays.asList(blockingSfs);
+ Scan scan = new Scan();
+ StoreScanner storeScanner =
+ new StoreScanner(scan, adaptiveScanInfo(), (NavigableSet) null, new ArrayList<>());
+ injectAdaptiveFields(storeScanner, exec);
+
+ // Run seekScanners in a separate thread so we can interrupt it
+ final IOException[] caughtError = { null };
+ Thread seekThread = new Thread(() -> {
+ try {
+ storeScanner.seekScanners(scanners, makeSeekKey(), false, true);
+ } catch (IOException e) {
+ caughtError[0] = e;
+ }
+ });
+ seekThread.start();
+
+ // Give the thread time to submit the handler and start awaiting the latch
+ Waiter.waitFor(HBaseConfiguration.create(), 5000, () -> seekThread.getState() == Thread.State.WAITING
+ || seekThread.getState() == Thread.State.TIMED_WAITING);
+
+ // Interrupt the seeking thread
+ seekThread.interrupt();
+ seekThread.join(5000);
+
+ // Release the blocking scanner so background thread can finish
+ scannerBlockLatch.countDown();
+
+ assertNotNull(caughtError[0], "Should have caught an IOException");
+ assertInstanceOf(InterruptedIOException.class, caughtError[0],
+ "IOException should be InterruptedIOException, got: " + caughtError[0].getClass());
+
+ storeScanner.close();
+ exec.shutdown();
+ }
+
+ /**
+ * Test 9: mixed capacity — capacity fluctuates between iterations. Some scanners are parallel,
+ * some sequential, all must be sought exactly once.
+ */
+ @Test
+ public void testMixedCapacityFluctuation() throws Exception {
+ // Use a pool of size 1: capacity alternates between 0 (when active) and 1 (when idle)
+ // We'll use memstore scanners (inline) mixed with StoreFileScanners to ensure all are sought
+ int poolSize = 2;
+ ExecutorService exec = startParallelSeekExecutor(poolSize);
+ try {
+ // 2 StoreFileScanners + 2 memstore scanners
+ StoreFileScanner sfs1 = Mockito.mock(StoreFileScanner.class);
+ StoreFileScanner sfs2 = Mockito.mock(StoreFileScanner.class);
+ MockMemStoreScanner ms1 = new MockMemStoreScanner();
+ MockMemStoreScanner ms2 = new MockMemStoreScanner();
+ Mockito.when(sfs1.seek(Mockito.any())).thenReturn(true);
+ Mockito.when(sfs2.seek(Mockito.any())).thenReturn(true);
+
+ List scanners =
+ Arrays.asList(ms1, sfs1, ms2, sfs2);
+
+ Scan scan = new Scan();
+ StoreScanner storeScanner =
+ new StoreScanner(scan, adaptiveScanInfo(), (NavigableSet) null, new ArrayList<>());
+ injectAdaptiveFields(storeScanner, exec);
+
+ ExtendedCell seekKey = makeSeekKey();
+ storeScanner.seekScanners(scanners, seekKey, false, true);
+
+ // All four scanners must be sought exactly once regardless of parallel/sequential path
+ assertEquals(1, ms1.getSeekCount(), "ms1 should be sought once");
+ assertEquals(1, ms2.getSeekCount(), "ms2 should be sought once");
+ Mockito.verify(sfs1, Mockito.times(1)).seek(seekKey);
+ Mockito.verify(sfs2, Mockito.times(1)).seek(seekKey);
+
+ storeScanner.close();
+ } finally {
+ exec.shutdown();
+ }
+ }
}