Repository: pinpoint-apm/pinpoint
Issue: Implement parallel result scanner for scanning rows distributed with RowKeyDistributor
Labels: enhancement

Allow distributed scanners to run in parallel that would theoretically allow for better performance when scanning rows that are well distributed using `RowKeyDistributor`


Issue comments:
Reopening as scanner creations may be done in parallel as well, and this may cut some time during scanner creation.

Reopening to apply parallel scan for finding accept applications for nodes when drawing the server map.


Pull request review comments:

Patch that resolved the issue:
diff --git a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/AbstractDistributedScannerTableCallback.java b/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/AbstractDistributedScannerTableCallback.java
deleted file mode 100644
index bf5f36ddf462..000000000000
--- a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/AbstractDistributedScannerTableCallback.java
+++ /dev/null
@@ -1,109 +0,0 @@
-/*
- * Copyright 2015 NAVER Corp.
- *
- * Licensed 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 com.navercorp.pinpoint.common.hbase;
-
-import com.navercorp.pinpoint.common.util.StopWatch;
-import com.sematext.hbase.wd.AbstractRowKeyDistributor;
-import org.apache.hadoop.hbase.client.HTableInterface;
-import org.apache.hadoop.hbase.client.ResultScanner;
-import org.apache.hadoop.hbase.client.Scan;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.data.hadoop.hbase.ResultsExtractor;
-import org.springframework.data.hadoop.hbase.TableCallback;
-
-import java.io.IOException;
-
-/**
- * @author HyunGil Jeong
- */
-public abstract class AbstractDistributedScannerTableCallback<T> implements TableCallback<T> {
-
-    protected final Logger logger = LoggerFactory.getLogger(getClass());
-    protected final boolean debugEnabled = this.logger.isDebugEnabled();
-
-    protected final Scan scan;
-    protected final AbstractRowKeyDistributor rowKeyDistributor;
-    protected final ResultsExtractor<T> resultsExtractor;
-
-    protected AbstractDistributedScannerTableCallback(Scan scan, AbstractRowKeyDistributor rowKeyDistributor, ResultsExtractor<T> resultsExtractor) {
-        this.scan = scan;
-        this.rowKeyDistributor = rowKeyDistributor;
-        this.resultsExtractor = resultsExtractor;
-    }
-
-    @Override
-    public T doInTable(HTableInterface table) throws Throwable {
-        StopWatch watch = null;
-        if (debugEnabled) {
-            watch = new StopWatch();
-            watch.start();
-        }
-        final ResultScanner[] splitScanners = splitScanners(table);
-        final ResultScanner scanner = createResultScanner(splitScanners);
-        if (debugEnabled) {
-            logger.debug("DistributedScanner createTime:{}", watch.stop());
-            watch.start();
-        }
-        try {
-            return resultsExtractor.extractData(scanner);
-        } finally {
-            scanner.close();
-            if (debugEnabled) {
-                logger.debug("DistributedScanner scanTime:{}", watch.stop());
-            }
-        }
-    }
-
-    protected abstract ResultScanner createResultScanner(ResultScanner[] scanners) throws IOException;
-
-    private ResultScanner[] splitScanners(HTableInterface htable) throws IOException {
-        Scan[] scans = this.rowKeyDistributor.getDistributedScans(this.scan);
-        final int length = scans.length;
-        for(int i = 0; i < length; i++) {
-            Scan scan = scans[i];
-            // other properties are already set upon construction
-            scan.setId(this.scan.getId() + "-" + i);
-        }
-
-        ResultScanner[] scanners = new ResultScanner[length];
-        boolean success = false;
-        try {
-            for (int i = 0; i < length; i++) {
-                scanners[i] = htable.getScanner(scans[i]);
-            }
-            success = true;
-        } finally {
-            if (!success) {
-                closeScanner(scanners);
-            }
-        }
-        return scanners;
-    }
-
-    private void closeScanner(ResultScanner[] scannerList ) {
-        for (ResultScanner scanner : scannerList) {
-            if (scanner != null) {
-                try {
-                    scanner.close();
-                } catch (Exception e) {
-                    logger.warn("Scanner.close() error Caused:{}", e.getMessage(), e);
-                }
-            }
-        }
-    }
-}
diff --git a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/DistributedScannerTableCallback.java b/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/DistributedScannerTableCallback.java
deleted file mode 100644
index d0addee89c8d..000000000000
--- a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/DistributedScannerTableCallback.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * Copyright 2015 NAVER Corp.
- *
- * Licensed 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 com.navercorp.pinpoint.common.hbase;
-
-import com.sematext.hbase.wd.AbstractRowKeyDistributor;
-import com.sematext.hbase.wd.DistributedScanner;
-import org.apache.hadoop.hbase.client.ResultScanner;
-import org.apache.hadoop.hbase.client.Scan;
-import org.springframework.data.hadoop.hbase.ResultsExtractor;
-
-import java.io.IOException;
-
-/**
- * @author HyunGil Jeong
- */
-public class DistributedScannerTableCallback<T> extends AbstractDistributedScannerTableCallback<T> {
-
-    protected DistributedScannerTableCallback(Scan scan, AbstractRowKeyDistributor rowKeyDistributor, ResultsExtractor<T> resultsExtractor) {
-        super(scan, rowKeyDistributor, resultsExtractor);
-    }
-
-    @Override
-    protected ResultScanner createResultScanner(ResultScanner[] scanners) throws IOException {
-        return new DistributedScanner(super.rowKeyDistributor, scanners);
-    }
-
-}
diff --git a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/HbaseTemplate2.java b/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/HbaseTemplate2.java
index 911a1fd0a9a0..c9eb1ee64e85 100644
--- a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/HbaseTemplate2.java
+++ b/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/HbaseTemplate2.java
@@ -16,6 +16,8 @@
 
 package com.navercorp.pinpoint.common.hbase;
 
+import com.navercorp.pinpoint.common.hbase.parallel.ParallelResultScanner;
+import com.navercorp.pinpoint.common.hbase.parallel.ScanTaskException;
 import com.navercorp.pinpoint.common.util.ExecutorFactory;
 import com.navercorp.pinpoint.common.util.PinpointThreadFactory;
 import com.navercorp.pinpoint.common.util.StopWatch;
@@ -48,6 +50,7 @@ public class HbaseTemplate2 extends HbaseTemplate implements HbaseOperations2, I
     private static final int DEFAULT_MAX_THREADS_PER_PARALLEL_SCAN = 1;
 
     private final Logger logger = LoggerFactory.getLogger(this.getClass());
+    private final boolean debugEnabled = this.logger.isDebugEnabled();
 
     private ExecutorService executor;
     private boolean enableParallelScan = false;
@@ -358,24 +361,87 @@ public <T> List<List<T>> find(String tableName, List<Scan> scanList, RowMapper<T
     @Override
     public <T> List<T> find(String tableName, final Scan scan, final AbstractRowKeyDistributor rowKeyDistributor, final RowMapper<T> action) {
         final ResultsExtractor<List<T>> resultsExtractor = new RowMapperResultsExtractor<>(action);
-        return execute(tableName, new DistributedScannerTableCallback<>(scan, rowKeyDistributor, resultsExtractor));
+        return executeDistributedScan(tableName, scan, rowKeyDistributor, resultsExtractor);
     }
 
     @Override
     public <T> List<T> find(String tableName, final Scan scan, final AbstractRowKeyDistributor rowKeyDistributor, final int limit, final RowMapper<T> action) {
         final ResultsExtractor<List<T>> resultsExtractor = new LimitRowMapperResultsExtractor<>(action, limit);
-        return execute(tableName, new DistributedScannerTableCallback<>(scan, rowKeyDistributor, resultsExtractor));
+        return executeDistributedScan(tableName, scan, rowKeyDistributor, resultsExtractor);
     }
 
     @Override
     public <T> List<T> find(String tableName, final Scan scan, final AbstractRowKeyDistributor rowKeyDistributor, int limit, final RowMapper<T> action, final LimitEventHandler limitEventHandler) {
         final LimitRowMapperResultsExtractor<T> resultsExtractor = new LimitRowMapperResultsExtractor<>(action, limit, limitEventHandler);
-        return execute(tableName, new DistributedScannerTableCallback<>(scan, rowKeyDistributor, resultsExtractor));
+        return executeDistributedScan(tableName, scan, rowKeyDistributor, resultsExtractor);
     }
 
     @Override
     public <T> T find(String tableName, final Scan scan, final AbstractRowKeyDistributor rowKeyDistributor, final ResultsExtractor<T> action) {
-        return execute(tableName, new DistributedScannerTableCallback<>(scan, rowKeyDistributor, action));
+        return executeDistributedScan(tableName, scan, rowKeyDistributor, action);
+    }
+
+    protected final <T> T executeDistributedScan(String tableName, final Scan scan, final AbstractRowKeyDistributor rowKeyDistributor, final ResultsExtractor<T> action) {
+        return execute(tableName, new TableCallback<T>() {
+            @Override
+            public T doInTable(HTableInterface table) throws Throwable {
+                StopWatch watch = null;
+                if (debugEnabled) {
+                    watch = new StopWatch();
+                    watch.start();
+                }
+                final ResultScanner[] splitScanners = splitScan(table, scan, rowKeyDistributor);
+                final ResultScanner scanner = new DistributedScanner(rowKeyDistributor, splitScanners);
+                if (debugEnabled) {
+                    logger.debug("DistributedScanner createTime: {}ms", watch.stop());
+                    watch.start();
+                }
+                try {
+                    return action.extractData(scanner);
+                } finally {
+                    scanner.close();
+                    if (debugEnabled) {
+                        logger.debug("DistributedScanner scanTime: {}ms", watch.stop());
+                    }
+                }
+            }
+        });
+    }
+
+    private ResultScanner[] splitScan(HTableInterface htable, Scan originalScan, AbstractRowKeyDistributor rowKeyDistributor) throws IOException {
+        Scan[] scans = rowKeyDistributor.getDistributedScans(originalScan);
+        final int length = scans.length;
+        for(int i = 0; i < length; i++) {
+            Scan scan = scans[i];
+            // other properties are already set upon construction
+            scan.setId(scan.getId() + "-" + i);
+        }
+
+        ResultScanner[] scanners = new ResultScanner[length];
+        boolean success = false;
+        try {
+            for (int i = 0; i < length; i++) {
+                scanners[i] = htable.getScanner(scans[i]);
+            }
+            success = true;
+        } finally {
+            if (!success) {
+                closeScanner(scanners);
+            }
+        }
+        return scanners;
+    }
+
+    private void closeScanner(ResultScanner[] scannerList ) {
+        for (ResultScanner scanner : scannerList) {
+            if (scanner != null) {
+                try {
+                    scanner.close();
+                } catch (Exception e) {
+                    logger.warn("Scanner.close() error Caused:{}", e.getMessage(), e);
+                }
+            }
+        }
     }
 
     @Override
@@ -386,7 +452,7 @@ public <T> List<T> findParallel(String tableName, Scan scan, AbstractRowKeyDistr
         } else {
             int numThreadsUsed = numParallelThreads < this.maxThreadsPerParallelScan ? numParallelThreads : this.maxThreadsPerParallelScan;
             final ResultsExtractor<List<T>> resultsExtractor = new RowMapperResultsExtractor<>(action);
-            return execute(tableName, new ParallelDistributedScannerTableCallback<>(scan, rowKeyDistributor, resultsExtractor, this.executor, numThreadsUsed));
+            return executeParallelDistributedScan(tableName, scan, rowKeyDistributor, resultsExtractor, numThreadsUsed);
         }
     }
 
@@ -398,7 +464,7 @@ public <T> List<T> findParallel(String tableName, Scan scan, AbstractRowKeyDistr
         } else {
             int numThreadsUsed = numParallelThreads < this.maxThreadsPerParallelScan ? numParallelThreads : this.maxThreadsPerParallelScan;
             final ResultsExtractor<List<T>> resultsExtractor = new LimitRowMapperResultsExtractor<>(action, limit);
-            return execute(tableName, new ParallelDistributedScannerTableCallback<>(scan, rowKeyDistributor, resultsExtractor, this.executor, numThreadsUsed));
+            return executeParallelDistributedScan(tableName, scan, rowKeyDistributor, resultsExtractor, numThreadsUsed);
         }
     }
 
@@ -410,7 +476,7 @@ public <T> List<T> findParallel(String tableName, Scan scan, AbstractRowKeyDistr
         } else {
             int numThreadsUsed = numParallelThreads < this.maxThreadsPerParallelScan ? numParallelThreads : this.maxThreadsPerParallelScan;
             final LimitRowMapperResultsExtractor<T> resultsExtractor = new LimitRowMapperResultsExtractor<>(action, limit, limitEventHandler);
-            return execute(tableName, new ParallelDistributedScannerTableCallback<>(scan, rowKeyDistributor, resultsExtractor, this.executor, numThreadsUsed));
+            return executeParallelDistributedScan(tableName, scan, rowKeyDistributor, resultsExtractor, numThreadsUsed);
         }
     }
 
@@ -421,7 +487,42 @@ public <T> T findParallel(String tableName, Scan scan, AbstractRowKeyDistributor
             return find(tableName, scan, rowKeyDistributor, action);
         } else {
             int numThreadsUsed = numParallelThreads < this.maxThreadsPerParallelScan ? numParallelThreads : this.maxThreadsPerParallelScan;
-            return execute(tableName, new ParallelDistributedScannerTableCallback<>(scan, rowKeyDistributor, action, this.executor, numThreadsUsed));
+            return executeParallelDistributedScan(tableName, scan, rowKeyDistributor, action, numThreadsUsed);
+        }
+    }
+
+    protected final <T> T executeParallelDistributedScan(String tableName, Scan scan, AbstractRowKeyDistributor rowKeyDistributor, ResultsExtractor<T> action, int numParallelThreads) {
+        try {
+            StopWatch watch = null;
+            if (debugEnabled) {
+                watch = new StopWatch();
+                watch.start();
+            }
+            ParallelResultScanner scanner = new ParallelResultScanner(tableName, this, this.executor, scan, rowKeyDistributor, numParallelThreads);
+            if (debugEnabled) {
+                logger.debug("ParallelDistributedScanner createTime: {}ms", watch.stop());
+                watch.start();
+            }
+            try {
+                return action.extractData(scanner);
+            } finally {
+                scanner.close();
+                if (debugEnabled) {
+                    logger.debug("ParallelDistributedScanner scanTime: {}ms", watch.stop());
+                }
+            }
+        } catch (Throwable th) {
+            Throwable throwable = th;
+            if (th instanceof ScanTaskException) {
+                throwable = th.getCause();
+            }
+            if (throwable instanceof Error) {
+                throw ((Error) th);
+            }
+            if (throwable instanceof RuntimeException) {
+                throw ((RuntimeException) th);
+            }
+            throw convertHbaseAccessException((Exception) throwable);
         }
     }
 
diff --git a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/ParallelDistributedScannerTableCallback.java b/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/ParallelDistributedScannerTableCallback.java
deleted file mode 100644
index f02715a63745..000000000000
--- a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/ParallelDistributedScannerTableCallback.java
+++ /dev/null
@@ -1,74 +0,0 @@
-/*
- * Copyright 2015 NAVER Corp.
- *
- * Licensed 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 com.navercorp.pinpoint.common.hbase;
-
-import com.sematext.hbase.wd.AbstractRowKeyDistributor;
-import com.sematext.hbase.wd.DistributedScanner;
-import org.apache.hadoop.hbase.client.ResultScanner;
-import org.apache.hadoop.hbase.client.Scan;
-import org.springframework.data.hadoop.hbase.ResultsExtractor;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.ExecutorService;
-
-/**
- * @author HyunGil Jeong
- */
-public class ParallelDistributedScannerTableCallback<T> extends AbstractDistributedScannerTableCallback<T> {
-
-    private final ExecutorService executor;
-    private final int numCaching;
-    private final int numParallelThreads;
-
-    protected ParallelDistributedScannerTableCallback(Scan scan, AbstractRowKeyDistributor rowKeyDistributor, ResultsExtractor<T> resultsExtractor, ExecutorService executor, int numParallelThreads) {
-        super(scan, rowKeyDistributor, resultsExtractor);
-        this.executor = executor;
-        this.numCaching = scan.getCaching();
-        this.numParallelThreads = numParallelThreads;
-    }
-
-    @Override
-    protected ResultScanner createResultScanner(ResultScanner[] scanners) throws IOException {
-        if (scanners.length < this.numParallelThreads) {
-            return new ParallelResultScanner(this.executor, this.numCaching, rowKeyDistributor, scanners);
-        } else {
-            ResultScanner[] distributedScanners = createDistributedScanners(scanners);
-            return new ParallelResultScanner(this.executor, this.numCaching, rowKeyDistributor, distributedScanners);
-        }
-    }
-
-    private ResultScanner[] createDistributedScanners(ResultScanner[] scanners) throws IOException {
-        int maxIndividualScans = (scanners.length + (this.numParallelThreads - 1)) / this.numParallelThreads;
-        List<List<ResultScanner>> scannerDistributions = new ArrayList<>(this.numParallelThreads);
-        for (int i = 0; i < this.numParallelThreads; ++i) {
-            scannerDistributions.add(new ArrayList<ResultScanner>(maxIndividualScans));
-        }
-        for (int i = 0; i < scanners.length; ++i) {
-            scannerDistributions.get(i % this.numParallelThreads).add(scanners[i]);
-        }
-        DistributedScanner[] distributedScanners = new DistributedScanner[this.numParallelThreads];
-        for (int i = 0; i < distributedScanners.length; ++i) {
-            List<ResultScanner> scannerDistribution = scannerDistributions.get(i);
-            ResultScanner[] scannersForDistributedScanner = scannerDistribution.toArray(new ResultScanner[scannerDistribution.size()]);
-            distributedScanners[i] = new DistributedScanner(rowKeyDistributor, scannersForDistributedScanner);
-        }
-        return distributedScanners;
-    }
-
-}
diff --git a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/ParallelResultScanner.java b/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/parallel/ParallelResultScanner.java
similarity index 66%
rename from commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/ParallelResultScanner.java
rename to commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/parallel/ParallelResultScanner.java
index 0e016130607f..11cf4ca36040 100644
--- a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/ParallelResultScanner.java
+++ b/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/parallel/ParallelResultScanner.java
@@ -14,19 +14,19 @@
  * limitations under the License.
  */
 
-package com.navercorp.pinpoint.common.hbase;
+package com.navercorp.pinpoint.common.hbase.parallel;
 
 import com.sematext.hbase.wd.AbstractRowKeyDistributor;
 import org.apache.hadoop.hbase.client.Result;
 import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Scan;
 import org.apache.hadoop.hbase.util.Bytes;
+import org.springframework.data.hadoop.hbase.HbaseAccessor;
 
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
 
 /**
@@ -34,34 +34,70 @@
  */
 public class ParallelResultScanner implements ResultScanner {
 
-    private static final int DEFAULT_SCAN_TASK_QUEUE_SIZE = 100;
-
     private final AbstractRowKeyDistributor keyDistributor;
     private final List<ScanTask> scanTasks;
     private final Result[] nextResults;
     private Result next = null;
 
-    public ParallelResultScanner(ExecutorService executor, int numCaching, AbstractRowKeyDistributor keyDistributor, ResultScanner[] scanners) throws IOException {
+    public ParallelResultScanner(String tableName, HbaseAccessor hbaseAccessor, ExecutorService executor, Scan originalScan, AbstractRowKeyDistributor keyDistributor, int numParallelThreads) throws IOException {
+        if (hbaseAccessor == null) {
+            throw new NullPointerException("hbaseAccessor must not be null");
+        }
         if (executor == null) {
             throw new NullPointerException("executor must not be null");
         }
         if (keyDistributor == null) {
             throw new NullPointerException("keyDistributor must not be null");
         }
-        if (scanners == null) {
-            throw new NullPointerException("scanners must not be null");
+        if (originalScan == null) {
+            throw new NullPointerException("originalScan must not be null");
         }
         this.keyDistributor = keyDistributor;
-        this.nextResults = new Result[scanners.length];
-        this.scanTasks = new ArrayList<>(scanners.length);
-        final int scanTaskQueueSize = numCaching < DEFAULT_SCAN_TASK_QUEUE_SIZE ? numCaching : DEFAULT_SCAN_TASK_QUEUE_SIZE;
-        for (ResultScanner scanner : scanners) {
-            ScanTask scanTask = new ScanTask(scanner, scanTaskQueueSize);
-            this.scanTasks.add(scanTask);
+
+        final ScanTaskConfig scanTaskConfig = new ScanTaskConfig(tableName, hbaseAccessor, keyDistributor, originalScan.getCaching());
+        final Scan[] splitScans = splitScans(originalScan);
+
+        this.scanTasks = createScanTasks(scanTaskConfig, splitScans, numParallelThreads);
+        this.nextResults = new Result[scanTasks.size()];
+        for (ScanTask scanTask : scanTasks) {
             executor.execute(scanTask);
         }
     }
 
+    private Scan[] splitScans(Scan originalScan) throws IOException {
+        Scan[] scans = this.keyDistributor.getDistributedScans(originalScan);
+        for (int i = 0; i < scans.length; ++i) {
+            Scan scan = scans[i];
+            scan.setId(originalScan.getId() + "-" + i);
+        }
+        return scans;
+    }
+
+    private List<ScanTask> createScanTasks(ScanTaskConfig scanTaskConfig, Scan[] splitScans, int numParallelThreads) {
+        if (splitScans.length <= numParallelThreads) {
+            List<ScanTask> scanTasks = new ArrayList<>(splitScans.length);
+            for (Scan scan : splitScans) {
+                scanTasks.add(new ScanTask(scanTaskConfig, scan));
+            }
+            return scanTasks;
+        } else {
+            int maxIndividualScans = (splitScans.length + (numParallelThreads - 1)) / numParallelThreads;
+            List<List<Scan>> scanDistributions = new ArrayList<>(numParallelThreads);
+            for (int i = 0; i < numParallelThreads; ++i) {
+                scanDistributions.add(new ArrayList<Scan>(maxIndividualScans));
+            }
+            for (int i = 0; i < splitScans.length; ++i) {
+                scanDistributions.get(i % numParallelThreads).add(splitScans[i]);
+            }
+            List<ScanTask> scanTasks = new ArrayList<>(numParallelThreads);
+            for (List<Scan> scanDistribution : scanDistributions) {
+                Scan[] scansForSingleTask = scanDistribution.toArray(new Scan[scanDistribution.size()]);
+                scanTasks.add(new ScanTask(scanTaskConfig, scansForSingleTask));
+            }
+            return scanTasks;
+        }
+    }
+
     private boolean hasNext() throws IOException {
         if (next != null) {
             return true;
@@ -85,6 +121,8 @@ private Result nextInternal() throws IOException {
         int indexOfResultToUse = -1;
         for (int i = 0; i < this.scanTasks.size(); ++i) {
             ScanTask scanTask = this.scanTasks.get(i);
+            // fail fast in case of errors
+            checkTask(scanTask);
             if (nextResults[i] == null) {
                 try {
                     nextResults[i] = scanTask.getResult();
@@ -108,57 +146,10 @@ private Result nextInternal() throws IOException {
         return result;
     }
 
-    private static class ScanTask implements Runnable {
-
-        private static final Result END_RESULT = new Result();
-
-        private final ResultScanner scanner;
-        private final BlockingQueue<Result> resultQueue;
-        private volatile boolean isQueueClosed = false;
-        private volatile boolean isFinished = false;
-
-        private ScanTask(ResultScanner scanner, int queueSize) {
-            this.scanner = scanner;
-            this.resultQueue = new ArrayBlockingQueue<>(queueSize);
-        }
-
-        @Override
-        public void run() {
-            try {
-                for (Result result : this.scanner) {
-                    this.resultQueue.put(result);
-                    if (this.isFinished) {
-                        break;
-                    }
-                }
-                this.resultQueue.put(END_RESULT);
-            } catch (InterruptedException e) {
-                Thread.currentThread().interrupt();
-                this.resultQueue.clear();
-                this.resultQueue.add(END_RESULT);
-            } finally {
-                this.isFinished = true;
-                this.scanner.close();
-            }
-        }
-
-        public Result getResult() throws InterruptedException {
-            if (this.isQueueClosed) {
-                return null;
-            }
-            Result take = this.resultQueue.take();
-            if (take == END_RESULT) {
-                this.isQueueClosed = true;
-                return null;
-            }
-            return take;
-        }
-
-        public void close() {
-            this.isFinished = true;
-            // signal threads blocked on resultQueue
-            this.resultQueue.clear();
-            this.resultQueue.offer(END_RESULT);
+    private void checkTask(ScanTask scanTask) {
+        Throwable th = scanTask.getThrowable();
+        if (th != null) {
+            throw new ScanTaskException(th);
         }
     }
 
@@ -229,6 +220,5 @@ public void remove() {
                 throw new UnsupportedOperationException();
             }
         };
-
     }
 }
diff --git a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/parallel/ScanTask.java b/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/parallel/ScanTask.java
new file mode 100644
index 000000000000..75fd80fd73bb
--- /dev/null
+++ b/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/parallel/ScanTask.java
@@ -0,0 +1,136 @@
+/*
+ * Copyright 2015 NAVER Corp.
+ *
+ * Licensed 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 com.navercorp.pinpoint.common.hbase.parallel;
+
+import com.sematext.hbase.wd.AbstractRowKeyDistributor;
+import com.sematext.hbase.wd.DistributedScanner;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.client.HTableInterface;
+import org.apache.hadoop.hbase.client.HTableInterfaceFactory;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Scan;
+import org.springframework.data.hadoop.hbase.HbaseUtils;
+
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+
+/**
+ * @author HyunGil Jeong
+ */
+public class ScanTask implements Runnable {
+
+    private static final Result END_RESULT = new Result();
+
+    private final String tableName;
+    private final Configuration configuration;
+    private final Charset charset;
+    private final HTableInterfaceFactory tableFactory;
+    private final AbstractRowKeyDistributor rowKeyDistributor;
+
+    private final Scan[] scans;
+    private final BlockingQueue<Result> resultQueue;
+
+    private volatile Throwable throwable;
+    private volatile boolean isQueueClosed = false;
+    private volatile boolean isDone = false;
+
+    public ScanTask(ScanTaskConfig scanTaskConfig, Scan... scans) {
+        if (scanTaskConfig == null) {
+            throw new NullPointerException("scanTaskConfig must not be null");
+        }
+        if (scans == null) {
+            throw new NullPointerException("scans must not be null");
+        }
+        if (scans.length == 0) {
+            throw new IllegalArgumentException("scans must not be empty");
+        }
+        this.tableName = scanTaskConfig.getTableName();
+        this.configuration = scanTaskConfig.getConfiguration();
+        this.charset = scanTaskConfig.getCharset();
+        this.tableFactory = scanTaskConfig.getTableFactory();
+        this.rowKeyDistributor = scanTaskConfig.getRowKeyDistributor();
+        this.scans = scans;
+        this.resultQueue = new ArrayBlockingQueue<>(scanTaskConfig.getScanTaskQueueSize());
+    }
+
+    @Override
+    public void run() {
+        HTableInterface table = null;
+        try {
+            table = HbaseUtils.getHTable(this.tableName, this.configuration, this.charset, this.tableFactory);
+            ResultScanner scanner = createResultScanner(table);
+            try {
+                for (Result result : scanner) {
+                    this.resultQueue.put(result);
+                    if (this.isDone) {
+                        break;
+                    }
+                }
+            } finally {
+                this.isDone = true;
+                this.resultQueue.put(END_RESULT);
+                scanner.close();
+            }
+        } catch (Throwable th) {
+            this.throwable = th;
+            this.resultQueue.clear();
+            this.resultQueue.offer(END_RESULT);
+        } finally {
+            HbaseUtils.releaseTable(this.tableName, table);
+        }
+    }
+
+    private ResultScanner createResultScanner(HTableInterface table) throws IOException {
+        if (scans.length == 1) {
+            Scan scan = scans[0];
+            return table.getScanner(scan);
+        } else {
+            ResultScanner[] scanners = new ResultScanner[this.scans.length];
+            for (int i = 0; i < scanners.length; ++i) {
+                scanners[i] = table.getScanner(this.scans[i]);
+            }
+            return new DistributedScanner(this.rowKeyDistributor, scanners);
+        }
+    }
+
+    public Result getResult() throws InterruptedException {
+        if (this.isQueueClosed) {
+            return null;
+        }
+        Result take = this.resultQueue.take();
+        if (take == END_RESULT) {
+            this.isQueueClosed = true;
+            return null;
+        }
+        return take;
+    }
+
+    public void close() {
+        this.isDone = true;
+        // signal threads blocked on resultQueue
+        this.resultQueue.clear();
+        this.resultQueue.add(END_RESULT);
+    }
+
+    public Throwable getThrowable() {
+        return this.throwable;
+    }
+
+}
diff --git a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/parallel/ScanTaskConfig.java b/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/parallel/ScanTaskConfig.java
new file mode 100644
index 000000000000..f17d9ec3923d
--- /dev/null
+++ b/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/parallel/ScanTaskConfig.java
@@ -0,0 +1,88 @@
+/*
+ * Copyright 2015 NAVER Corp.
+ *
+ * Licensed 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 com.navercorp.pinpoint.common.hbase.parallel;
+
+import com.sematext.hbase.wd.AbstractRowKeyDistributor;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.client.HTableInterfaceFactory;
+import org.springframework.data.hadoop.hbase.HbaseAccessor;
+
+import java.nio.charset.Charset;
+
+/**
+ * @author HyunGil Jeong
+ */
+public class ScanTaskConfig {
+
+    private final String tableName;
+    private final Configuration configuration;
+    private final Charset charset;
+    private final HTableInterfaceFactory tableFactory;
+
+    private final AbstractRowKeyDistributor rowKeyDistributor;
+    private final int scanTaskQueueSize;
+
+    public ScanTaskConfig(String tableName, HbaseAccessor hbaseAccessor, AbstractRowKeyDistributor rowKeyDistributor, int scanCaching) {
+        this(tableName, hbaseAccessor.getConfiguration(), hbaseAccessor.getCharset(), hbaseAccessor.getTableFactory(), rowKeyDistributor, scanCaching);
+    }
+
+    public ScanTaskConfig(String tableName, Configuration configuration, Charset charset, HTableInterfaceFactory tableFactory, AbstractRowKeyDistributor rowKeyDistributor, int scanCaching) {
+        if (tableName == null) {
+            throw new NullPointerException("No table specified");
+        }
+        if (rowKeyDistributor == null) {
+            throw new NullPointerException("rowKeyDistributor must not be null");
+        }
+        this.tableName = tableName;
+        this.configuration = configuration;
+        this.charset = charset;
+        this.tableFactory = tableFactory;
+        this.rowKeyDistributor = rowKeyDistributor;
+        if (scanCaching > 0) {
+            this.scanTaskQueueSize = scanCaching;
+        } else {
+            this.scanTaskQueueSize = configuration.getInt(
+                    HConstants.HBASE_CLIENT_SCANNER_CACHING,
+                    HConstants.DEFAULT_HBASE_CLIENT_SCANNER_CACHING);
+        }
+    }
+
+    public String getTableName() {
+        return tableName;
+    }
+
+    public Configuration getConfiguration() {
+        return configuration;
+    }
+
+    public Charset getCharset() {
+        return charset;
+    }
+
+    public HTableInterfaceFactory getTableFactory() {
+        return tableFactory;
+    }
+
+    public AbstractRowKeyDistributor getRowKeyDistributor() {
+        return rowKeyDistributor;
+    }
+
+    public int getScanTaskQueueSize() {
+        return scanTaskQueueSize;
+    }
+}
diff --git a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/parallel/ScanTaskException.java b/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/parallel/ScanTaskException.java
new file mode 100644
index 000000000000..1df0a24953a1
--- /dev/null
+++ b/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/parallel/ScanTaskException.java
@@ -0,0 +1,33 @@
+/*
+ * Copyright 2015 NAVER Corp.
+ *
+ * Licensed 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 com.navercorp.pinpoint.common.hbase.parallel;
+
+/**
+ * @author HyunGil Jeong
+ */
+public class ScanTaskException extends RuntimeException {
+
+    private static final long serialVersionUID = 8554224683436066023L;
+
+    public ScanTaskException(Throwable th) {
+        super(th);
+    }
+
+    public ScanTaskException(String message, Throwable th) {
+        super(message, th);
+    }
+}
diff --git a/docker/pinpoint-web/hbase.properties b/docker/pinpoint-web/hbase.properties
index 4161185c010c..b94b1893aa93 100644
--- a/docker/pinpoint-web/hbase.properties
+++ b/docker/pinpoint-web/hbase.properties
@@ -19,4 +19,10 @@ hbase.ipc.client.socket.timeout.write=60000
 hbase.client.thread.max=128
 hbase.client.threadPool.queueSize=5120
 # prestartAllCoreThreads
-hbase.client.threadPool.prestart=false
\ No newline at end of file
+hbase.client.threadPool.prestart=false
+
+#==================================================================================
+# hbase parallel scan options
+hbase.client.parallel.scan.enable=false
+hbase.client.parallel.scan.maxthreads=128
+hbase.client.parallel.scan.maxthreadsperscan=16
\ No newline at end of file
diff --git a/quickstart/web/src/main/resources/hbase.properties b/quickstart/web/src/main/resources/hbase.properties
index be6e5ea43e88..ac1e4a555505 100644
--- a/quickstart/web/src/main/resources/hbase.properties
+++ b/quickstart/web/src/main/resources/hbase.properties
@@ -20,4 +20,10 @@ hbase.ipc.client.socket.timeout.write=60000
 hbase.client.thread.max=128
 hbase.client.threadPool.queueSize=5120
 # prestartAllCoreThreads
-hbase.client.threadPool.prestart=false
\ No newline at end of file
+hbase.client.threadPool.prestart=false
+
+#==================================================================================
+# hbase parallel scan options
+hbase.client.parallel.scan.enable=false
+hbase.client.parallel.scan.maxthreads=128
+hbase.client.parallel.scan.maxthreadsperscan=16
\ No newline at end of file


Test patch: