HBASE-19473 Miscellaneous changes to ClientScanner

- Remove superfluous logging code guard
- Simplify some of the code
- Use ArrayDeque instead of LinkedList for queue implementation
This commit is contained in:
BELUGA BEHR 2018-01-03 19:59:45 -08:00 committed by Apekshit Sharma
parent 4c0fba6a85
commit 72631a08c1
1 changed files with 50 additions and 51 deletions

View File

@ -25,29 +25,30 @@ import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesti
import java.io.IOException; import java.io.IOException;
import java.io.InterruptedIOException; import java.io.InterruptedIOException;
import java.util.LinkedList; import java.util.ArrayDeque;
import java.util.Queue; import java.util.Queue;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import org.apache.commons.lang3.mutable.MutableBoolean; import org.apache.commons.lang3.mutable.MutableBoolean;
import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.ScannerCallable.MoreResults;
import org.apache.hadoop.hbase.DoNotRetryIOException; import org.apache.hadoop.hbase.DoNotRetryIOException;
import org.apache.hadoop.hbase.exceptions.OutOfOrderScannerNextException;
import org.apache.hadoop.hbase.exceptions.ScannerResetException;
import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HRegionInfo; import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
import org.apache.hadoop.hbase.NotServingRegionException; import org.apache.hadoop.hbase.NotServingRegionException;
import org.apache.hadoop.hbase.regionserver.RegionServerStoppedException;
import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.UnknownScannerException; import org.apache.hadoop.hbase.UnknownScannerException;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.yetus.audience.InterfaceAudience; import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.apache.hadoop.hbase.client.ScannerCallable.MoreResults;
import org.apache.hadoop.hbase.exceptions.OutOfOrderScannerNextException;
import org.apache.hadoop.hbase.exceptions.ScannerResetException;
import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
import org.apache.hadoop.hbase.regionserver.RegionServerStoppedException;
import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil; import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
import org.apache.hadoop.hbase.util.Bytes;
/** /**
* Implements the scanner interface for the HBase client. If there are multiple regions in a table, * Implements the scanner interface for the HBase client. If there are multiple regions in a table,
@ -294,25 +295,30 @@ public abstract class ClientScanner extends AbstractClientScanner {
} }
protected void initSyncCache() { protected void initSyncCache() {
cache = new LinkedList<>(); cache = new ArrayDeque<>();
} }
protected Result nextWithSyncCache() throws IOException { protected Result nextWithSyncCache() throws IOException {
// If the scanner is closed and there's nothing left in the cache, next is a no-op. Result result = cache.poll();
if (cache.isEmpty() && this.closed) { if (result != null) {
return result;
}
// If there is nothing left in the cache and the scanner is closed,
// return a no-op
if (this.closed) {
return null; return null;
} }
if (cache.isEmpty()) {
loadCache();
}
if (cache.size() > 0) { loadCache();
return cache.poll();
} // try again to load from cache
result = cache.poll();
// if we exhausted this scanner before calling close, write out the scan metrics // if we exhausted this scanner before calling close, write out the scan metrics
if (result == null) {
writeScanMetrics(); writeScanMetrics();
return null; }
return result;
} }
@VisibleForTesting @VisibleForTesting
@ -410,12 +416,10 @@ public abstract class ClientScanner extends AbstractClientScanner {
long remainingResultSize = maxScannerResultSize; long remainingResultSize = maxScannerResultSize;
int countdown = this.caching; int countdown = this.caching;
// This is possible if we just stopped at the boundary of a region in the previous call. // This is possible if we just stopped at the boundary of a region in the previous call.
if (callable == null) { if (callable == null && !moveToNextRegion()) {
if (!moveToNextRegion()) {
closed = true; closed = true;
return; return;
} }
}
// This flag is set when we want to skip the result returned. We do // This flag is set when we want to skip the result returned. We do
// this when we reset scanner because it split under us. // this when we reset scanner because it split under us.
MutableBoolean retryAfterOutOfOrderException = new MutableBoolean(true); MutableBoolean retryAfterOutOfOrderException = new MutableBoolean(true);
@ -463,7 +467,6 @@ public abstract class ClientScanner extends AbstractClientScanner {
scanResultCache.addAndGet(values, callable.isHeartbeatMessage()); scanResultCache.addAndGet(values, callable.isHeartbeatMessage());
int numberOfCompleteRows = int numberOfCompleteRows =
scanResultCache.numberOfCompleteRows() - numberOfCompleteRowsBefore; scanResultCache.numberOfCompleteRows() - numberOfCompleteRowsBefore;
if (resultsToAddToCache.length > 0) {
for (Result rs : resultsToAddToCache) { for (Result rs : resultsToAddToCache) {
cache.add(rs); cache.add(rs);
long estimatedHeapSizeOfResult = calcEstimatedSize(rs); long estimatedHeapSizeOfResult = calcEstimatedSize(rs);
@ -472,7 +475,6 @@ public abstract class ClientScanner extends AbstractClientScanner {
addEstimatedSize(estimatedHeapSizeOfResult); addEstimatedSize(estimatedHeapSizeOfResult);
this.lastResult = rs; this.lastResult = rs;
} }
}
if (scan.getLimit() > 0) { if (scan.getLimit() > 0) {
int newLimit = scan.getLimit() - numberOfCompleteRows; int newLimit = scan.getLimit() - numberOfCompleteRows;
@ -491,10 +493,8 @@ public abstract class ClientScanner extends AbstractClientScanner {
// processing of the scan is taking a long time server side. Rather than continue to // processing of the scan is taking a long time server side. Rather than continue to
// loop until a limit (e.g. size or caching) is reached, break out early to avoid causing // loop until a limit (e.g. size or caching) is reached, break out early to avoid causing
// unnecesary delays to the caller // unnecesary delays to the caller
if (LOG.isTraceEnabled()) { LOG.trace("Heartbeat message received and cache contains Results. " +
LOG.trace("Heartbeat message received and cache contains Results." + "Breaking out of scan loop");
" Breaking out of scan loop");
}
// we know that the region has not been exhausted yet so just break without calling // we know that the region has not been exhausted yet so just break without calling
// closeScannerIfExhausted // closeScannerIfExhausted
break; break;
@ -560,9 +560,7 @@ public abstract class ClientScanner extends AbstractClientScanner {
// We used to catch this error, interpret, and rethrow. However, we // We used to catch this error, interpret, and rethrow. However, we
// have since decided that it's not nice for a scanner's close to // have since decided that it's not nice for a scanner's close to
// throw exceptions. Chances are it was just due to lease time out. // throw exceptions. Chances are it was just due to lease time out.
if (LOG.isDebugEnabled()) {
LOG.debug("scanner failed to close", e); LOG.debug("scanner failed to close", e);
}
} catch (IOException e) { } catch (IOException e) {
/* An exception other than UnknownScanner is unexpected. */ /* An exception other than UnknownScanner is unexpected. */
LOG.warn("scanner failed to close.", e); LOG.warn("scanner failed to close.", e);
@ -574,19 +572,20 @@ public abstract class ClientScanner extends AbstractClientScanner {
@Override @Override
public boolean renewLease() { public boolean renewLease() {
if (callable != null) { if (callable == null) {
return false;
}
// do not return any rows, do not advance the scanner // do not return any rows, do not advance the scanner
callable.setRenew(true); callable.setRenew(true);
try { try {
this.caller.callWithoutRetries(callable, this.scannerTimeout); this.caller.callWithoutRetries(callable, this.scannerTimeout);
return true;
} catch (Exception e) { } catch (Exception e) {
LOG.debug("scanner failed to renew lease", e);
return false; return false;
} finally { } finally {
callable.setRenew(false); callable.setRenew(false);
} }
return true;
}
return false;
} }
protected void initCache() { protected void initCache() {