From 82282a9f95997f1c670186ad595bfd1b3d317d14 Mon Sep 17 00:00:00 2001 From: Nick Dimiduk Date: Thu, 24 Feb 2022 17:20:57 +0100 Subject: [PATCH] HBASE-26765 Minor refactor of async scanning code (#4121) Signed-off-by: Duo Zhang --- .../hbase/client/AsyncConnectionImpl.java | 2 +- .../hadoop/hbase/client/AsyncTableImpl.java | 5 ++--- .../hbase/client/AsyncTableResultScanner.java | 22 +++++++++---------- .../hbase/client/RawAsyncTableImpl.java | 12 ++++++---- 4 files changed, 22 insertions(+), 19 deletions(-) diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncConnectionImpl.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncConnectionImpl.java index d458c6cb66e..ee6b393b3d1 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncConnectionImpl.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncConnectionImpl.java @@ -340,7 +340,7 @@ public class AsyncConnectionImpl implements AsyncConnection { public AsyncTable build() { RawAsyncTableImpl rawTable = new RawAsyncTableImpl(AsyncConnectionImpl.this, RETRY_TIMER, this); - return new AsyncTableImpl(AsyncConnectionImpl.this, rawTable, pool); + return new AsyncTableImpl(rawTable, pool); } }; } diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncTableImpl.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncTableImpl.java index ba2c5600282..f2a8e18b521 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncTableImpl.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncTableImpl.java @@ -43,12 +43,11 @@ import org.apache.yetus.audience.InterfaceAudience; @InterfaceAudience.Private class AsyncTableImpl implements AsyncTable { - private final AsyncTable rawTable; + private final RawAsyncTableImpl rawTable; private final ExecutorService pool; - AsyncTableImpl(AsyncConnectionImpl conn, AsyncTable rawTable, - ExecutorService pool) { + AsyncTableImpl(RawAsyncTableImpl rawTable, ExecutorService pool) { this.rawTable = rawTable; this.pool = pool; } diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncTableResultScanner.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncTableResultScanner.java index 7d520099744..616cf6b9e60 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncTableResultScanner.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncTableResultScanner.java @@ -23,6 +23,7 @@ import java.io.IOException; import java.io.InterruptedIOException; import java.util.ArrayDeque; import java.util.Queue; +import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.client.metrics.ScanMetrics; import org.apache.hadoop.hbase.util.FutureUtils; import org.apache.yetus.audience.InterfaceAudience; @@ -30,8 +31,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** - * The {@link ResultScanner} implementation for {@link AsyncTable}. It will fetch data automatically - * in background and cache it in memory. Typically the {@link #maxCacheSize} will be + * The {@link ResultScanner} implementation for {@link RawAsyncTableImpl}. It will fetch data + * automatically in background and cache it in memory. Typically, the {@link #maxCacheSize} will be * {@code 2 * scan.getMaxResultSize()}. */ @InterfaceAudience.Private @@ -39,7 +40,7 @@ class AsyncTableResultScanner implements ResultScanner, AdvancedScanResultConsum private static final Logger LOG = LoggerFactory.getLogger(AsyncTableResultScanner.class); - private final AsyncTable rawTable; + private final TableName tableName; private final long maxCacheSize; @@ -57,12 +58,10 @@ class AsyncTableResultScanner implements ResultScanner, AdvancedScanResultConsum private ScanResumer resumer; - public AsyncTableResultScanner(AsyncTable table, Scan scan, - long maxCacheSize) { - this.rawTable = table; + public AsyncTableResultScanner(TableName tableName, Scan scan, long maxCacheSize) { + this.tableName = tableName; this.maxCacheSize = maxCacheSize; this.scan = scan; - table.scan(scan, this); } private void addToCache(Result result) { @@ -72,9 +71,10 @@ class AsyncTableResultScanner implements ResultScanner, AdvancedScanResultConsum private void stopPrefetch(ScanController controller) { if (LOG.isDebugEnabled()) { - LOG.debug(String.format("0x%x", System.identityHashCode(this)) + - " stop prefetching when scanning " + rawTable.getName() + " as the cache size " + - cacheSize + " is greater than the maxCacheSize " + maxCacheSize); + LOG.debug("{} stop prefetching when scanning {} as the cache size {}" + + " is greater than the maxCacheSize {}", + String.format("0x%x", System.identityHashCode(this)), tableName, cacheSize, + maxCacheSize); } resumer = controller.suspend(); } @@ -138,7 +138,7 @@ class AsyncTableResultScanner implements ResultScanner, AdvancedScanResultConsum return null; } if (error != null) { - FutureUtils.rethrow(error); + throw FutureUtils.rethrow(error); } try { wait(); diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RawAsyncTableImpl.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RawAsyncTableImpl.java index 20e84cc6f73..437f12f80a1 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RawAsyncTableImpl.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RawAsyncTableImpl.java @@ -666,10 +666,14 @@ class RawAsyncTableImpl implements AsyncTable { } @Override - public ResultScanner getScanner(Scan scan) { - return new AsyncTableResultScanner(this, ReflectionUtils.newInstance(scan.getClass(), scan), - resultSize2CacheSize( - scan.getMaxResultSize() > 0 ? scan.getMaxResultSize() : defaultScannerMaxResultSize)); + public AsyncTableResultScanner getScanner(Scan scan) { + final long maxCacheSize = resultSize2CacheSize( + scan.getMaxResultSize() > 0 ? scan.getMaxResultSize() : defaultScannerMaxResultSize); + final Scan scanCopy = ReflectionUtils.newInstance(scan.getClass(), scan); + final AsyncTableResultScanner scanner = + new AsyncTableResultScanner(tableName, scanCopy, maxCacheSize); + scan(scan, scanner); + return scanner; } @Override