View Javadoc

1   /**
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
17   */
18  package org.apache.hadoop.hbase.client;
19  
20  import java.io.IOException;
21  import java.io.InterruptedIOException;
22  import java.util.ArrayList;
23  import java.util.Arrays;
24  import java.util.LinkedList;
25  import java.util.List;
26  import java.util.concurrent.ExecutorService;
27  
28  import org.apache.commons.logging.Log;
29  import org.apache.commons.logging.LogFactory;
30  import org.apache.hadoop.hbase.classification.InterfaceAudience;
31  import org.apache.hadoop.conf.Configuration;
32  import org.apache.hadoop.hbase.Cell;
33  import org.apache.hadoop.hbase.CellUtil;
34  import org.apache.hadoop.hbase.DoNotRetryIOException;
35  import org.apache.hadoop.hbase.HBaseConfiguration;
36  import org.apache.hadoop.hbase.HConstants;
37  import org.apache.hadoop.hbase.HRegionInfo;
38  import org.apache.hadoop.hbase.NotServingRegionException;
39  import org.apache.hadoop.hbase.TableName;
40  import org.apache.hadoop.hbase.UnknownScannerException;
41  import org.apache.hadoop.hbase.client.RpcRetryingCallerFactory;
42  import org.apache.hadoop.hbase.exceptions.OutOfOrderScannerNextException;
43  import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
44  import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
45  import org.apache.hadoop.hbase.protobuf.generated.MapReduceProtos;
46  import org.apache.hadoop.hbase.regionserver.RegionServerStoppedException;
47  import org.apache.hadoop.hbase.util.Bytes;
48  
49  import com.google.common.annotations.VisibleForTesting;
50  
51  /**
52   * Implements the scanner interface for the HBase client.
53   * If there are multiple regions in a table, this scanner will iterate
54   * through them all.
55   */
56  @InterfaceAudience.Private
57  public class ClientScanner extends AbstractClientScanner {
58      private static final Log LOG = LogFactory.getLog(ClientScanner.class);
59      // A byte array in which all elements are the max byte, and it is used to
60      // construct closest front row
61      static byte[] MAX_BYTE_ARRAY = Bytes.createMaxByteArray(9);
62      protected Scan scan;
63      protected boolean closed = false;
64      // Current region scanner is against.  Gets cleared if current region goes
65      // wonky: e.g. if it splits on us.
66      protected HRegionInfo currentRegion = null;
67      protected ScannerCallableWithReplicas callable = null;
68      protected final LinkedList<Result> cache = new LinkedList<Result>();
69      /**
70       * A list of partial results that have been returned from the server. This list should only
71       * contain results if this scanner does not have enough partial results to form the complete
72       * result.
73       */
74      protected final LinkedList<Result> partialResults = new LinkedList<Result>();
75      /**
76       * The row for which we are accumulating partial Results (i.e. the row of the Results stored
77       * inside partialResults). Changes to partialResultsRow and partialResults are kept in sync
78       * via the methods {@link #addToPartialResults(Result)} and {@link #clearPartialResults()}
79       */
80      protected byte[] partialResultsRow = null;
81      protected final int caching;
82      protected long lastNext;
83      // Keep lastResult returned successfully in case we have to reset scanner.
84      protected Result lastResult = null;
85      protected final long maxScannerResultSize;
86      private final ClusterConnection connection;
87      private final TableName tableName;
88      protected final int scannerTimeout;
89      protected boolean scanMetricsPublished = false;
90      protected RpcRetryingCaller<Result []> caller;
91      protected RpcControllerFactory rpcControllerFactory;
92      protected Configuration conf;
93      //The timeout on the primary. Applicable if there are multiple replicas for a region
94      //In that case, we will only wait for this much timeout on the primary before going
95      //to the replicas and trying the same scan. Note that the retries will still happen
96      //on each replica and the first successful results will be taken. A timeout of 0 is
97      //disallowed.
98      protected final int primaryOperationTimeout;
99      private int retries;
100     protected final ExecutorService pool;
101 
102   /**
103    * Create a new ClientScanner for the specified table Note that the passed {@link Scan}'s start
104    * row maybe changed changed.
105    * @param conf The {@link Configuration} to use.
106    * @param scan {@link Scan} to use in this scanner
107    * @param tableName The table that we wish to scan
108    * @param connection Connection identifying the cluster
109    * @throws IOException
110    */
111   public ClientScanner(final Configuration conf, final Scan scan, final TableName tableName,
112       ClusterConnection connection, RpcRetryingCallerFactory rpcFactory,
113       RpcControllerFactory controllerFactory, ExecutorService pool, int primaryOperationTimeout)
114       throws IOException {
115       if (LOG.isTraceEnabled()) {
116         LOG.trace("Scan table=" + tableName
117             + ", startRow=" + Bytes.toStringBinary(scan.getStartRow()));
118       }
119       this.scan = scan;
120       this.tableName = tableName;
121       this.lastNext = System.currentTimeMillis();
122       this.connection = connection;
123       this.pool = pool;
124       this.primaryOperationTimeout = primaryOperationTimeout;
125       this.retries = conf.getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,
126           HConstants.DEFAULT_HBASE_CLIENT_RETRIES_NUMBER);
127       if (scan.getMaxResultSize() > 0) {
128         this.maxScannerResultSize = scan.getMaxResultSize();
129       } else {
130         this.maxScannerResultSize = conf.getLong(
131           HConstants.HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE_KEY,
132           HConstants.DEFAULT_HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE);
133       }
134       this.scannerTimeout = HBaseConfiguration.getInt(conf,
135         HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD,
136         HConstants.HBASE_REGIONSERVER_LEASE_PERIOD_KEY,
137         HConstants.DEFAULT_HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD);
138 
139       // check if application wants to collect scan metrics
140       initScanMetrics(scan);
141 
142       // Use the caching from the Scan.  If not set, use the default cache setting for this table.
143       if (this.scan.getCaching() > 0) {
144         this.caching = this.scan.getCaching();
145       } else {
146         this.caching = conf.getInt(
147             HConstants.HBASE_CLIENT_SCANNER_CACHING,
148             HConstants.DEFAULT_HBASE_CLIENT_SCANNER_CACHING);
149       }
150 
151       this.caller = rpcFactory.<Result[]> newCaller();
152       this.rpcControllerFactory = controllerFactory;
153 
154       this.conf = conf;
155       initializeScannerInConstruction();
156     }
157 
158     protected void initializeScannerInConstruction() throws IOException{
159       // initialize the scanner
160       nextScanner(this.caching, false);
161     }
162 
163     protected ClusterConnection getConnection() {
164       return this.connection;
165     }
166 
167     /**
168      * @return Table name
169      * @deprecated As of release 0.96
170      *             (<a href="https://issues.apache.org/jira/browse/HBASE-9508">HBASE-9508</a>).
171      *             This will be removed in HBase 2.0.0. Use {@link #getTable()}.
172      */
173     @Deprecated
174     protected byte [] getTableName() {
175       return this.tableName.getName();
176     }
177 
178     protected TableName getTable() {
179       return this.tableName;
180     }
181 
182     protected int getRetries() {
183       return this.retries;
184     }
185 
186     protected int getScannerTimeout() {
187       return this.scannerTimeout;
188     }
189 
190     protected Configuration getConf() {
191       return this.conf;
192     }
193 
194     protected Scan getScan() {
195       return scan;
196     }
197 
198     protected ExecutorService getPool() {
199       return pool;
200     }
201 
202     protected int getPrimaryOperationTimeout() {
203       return primaryOperationTimeout;
204     }
205 
206     protected int getCaching() {
207       return caching;
208     }
209 
210     protected long getTimestamp() {
211       return lastNext;
212     }
213 
214     @VisibleForTesting
215     protected long getMaxResultSize() {
216       return maxScannerResultSize;
217     }
218 
219     // returns true if the passed region endKey
220     protected boolean checkScanStopRow(final byte [] endKey) {
221       if (this.scan.getStopRow().length > 0) {
222         // there is a stop row, check to see if we are past it.
223         byte [] stopRow = scan.getStopRow();
224         int cmp = Bytes.compareTo(stopRow, 0, stopRow.length,
225           endKey, 0, endKey.length);
226         if (cmp <= 0) {
227           // stopRow <= endKey (endKey is equals to or larger than stopRow)
228           // This is a stop.
229           return true;
230         }
231       }
232       return false; //unlikely.
233     }
234 
235     private boolean possiblyNextScanner(int nbRows, final boolean done) throws IOException {
236       // If we have just switched replica, don't go to the next scanner yet. Rather, try
237       // the scanner operations on the new replica, from the right point in the scan
238       // Note that when we switched to a different replica we left it at a point
239       // where we just did the "openScanner" with the appropriate startrow
240       if (callable != null && callable.switchedToADifferentReplica()) return true;
241       return nextScanner(nbRows, done);
242     }
243 
244     /*
245      * Gets a scanner for the next region.  If this.currentRegion != null, then
246      * we will move to the endrow of this.currentRegion.  Else we will get
247      * scanner at the scan.getStartRow().  We will go no further, just tidy
248      * up outstanding scanners, if <code>currentRegion != null</code> and
249      * <code>done</code> is true.
250      * @param nbRows
251      * @param done Server-side says we're done scanning.
252      */
253   protected boolean nextScanner(int nbRows, final boolean done)
254     throws IOException {
255       // Close the previous scanner if it's open
256       if (this.callable != null) {
257         this.callable.setClose();
258         call(callable, caller, scannerTimeout);
259         this.callable = null;
260       }
261 
262       // Where to start the next scanner
263       byte [] localStartKey;
264 
265       // if we're at end of table, close and return false to stop iterating
266       if (this.currentRegion != null) {
267         byte [] endKey = this.currentRegion.getEndKey();
268         if (endKey == null ||
269             Bytes.equals(endKey, HConstants.EMPTY_BYTE_ARRAY) ||
270             checkScanStopRow(endKey) ||
271             done) {
272           close();
273           if (LOG.isTraceEnabled()) {
274             LOG.trace("Finished " + this.currentRegion);
275           }
276           return false;
277         }
278         localStartKey = endKey;
279         if (LOG.isTraceEnabled()) {
280           LOG.trace("Finished " + this.currentRegion);
281         }
282       } else {
283         localStartKey = this.scan.getStartRow();
284       }
285 
286       if (LOG.isDebugEnabled() && this.currentRegion != null) {
287         // Only worth logging if NOT first region in scan.
288         LOG.debug("Advancing internal scanner to startKey at '" +
289           Bytes.toStringBinary(localStartKey) + "'");
290       }
291       try {
292         callable = getScannerCallable(localStartKey, nbRows);
293         // Open a scanner on the region server starting at the
294         // beginning of the region
295         call(callable, caller, scannerTimeout);
296         this.currentRegion = callable.getHRegionInfo();
297         if (this.scanMetrics != null) {
298           this.scanMetrics.countOfRegions.incrementAndGet();
299         }
300       } catch (IOException e) {
301         close();
302         throw e;
303       }
304       return true;
305     }
306 
307   @VisibleForTesting
308   boolean isAnyRPCcancelled() {
309     return callable.isAnyRPCcancelled();
310   }
311 
312   Result[] call(ScannerCallableWithReplicas callable,
313       RpcRetryingCaller<Result[]> caller, int scannerTimeout)
314       throws IOException, RuntimeException {
315     if (Thread.interrupted()) {
316       throw new InterruptedIOException();
317     }
318     // callWithoutRetries is at this layer. Within the ScannerCallableWithReplicas,
319     // we do a callWithRetries
320     return caller.callWithoutRetries(callable, scannerTimeout);
321   }
322 
323     @InterfaceAudience.Private
324     protected ScannerCallableWithReplicas getScannerCallable(byte [] localStartKey,
325         int nbRows) {
326       scan.setStartRow(localStartKey);
327       ScannerCallable s =
328           new ScannerCallable(getConnection(), getTable(), scan, this.scanMetrics,
329               this.rpcControllerFactory);
330       s.setCaching(nbRows);
331       ScannerCallableWithReplicas sr = new ScannerCallableWithReplicas(tableName, getConnection(),
332        s, pool, primaryOperationTimeout, scan,
333        retries, scannerTimeout, caching, conf, caller);
334       return sr;
335     }
336 
337     /**
338      * Publish the scan metrics. For now, we use scan.setAttribute to pass the metrics back to the
339      * application or TableInputFormat.Later, we could push it to other systems. We don't use
340      * metrics framework because it doesn't support multi-instances of the same metrics on the same
341      * machine; for scan/map reduce scenarios, we will have multiple scans running at the same time.
342      *
343      * By default, scan metrics are disabled; if the application wants to collect them, this
344      * behavior can be turned on by calling calling {@link Scan#setScanMetricsEnabled(boolean)}
345      * 
346      * <p>This invocation clears the scan metrics. Metrics are aggregated in the Scan instance.
347      */
348     protected void writeScanMetrics() {
349       if (this.scanMetrics == null || scanMetricsPublished) {
350         return;
351       }
352       MapReduceProtos.ScanMetrics pScanMetrics = ProtobufUtil.toScanMetrics(scanMetrics);
353       scan.setAttribute(Scan.SCAN_ATTRIBUTES_METRICS_DATA, pScanMetrics.toByteArray());
354       scanMetricsPublished = true;
355     }
356 
357     @Override
358     public Result next() throws IOException {
359       // If the scanner is closed and there's nothing left in the cache, next is a no-op.
360       if (cache.size() == 0 && this.closed) {
361         return null;
362       }
363       if (cache.size() == 0) {
364         loadCache();
365       }
366 
367       if (cache.size() > 0) {
368         return cache.poll();
369       }
370 
371       // if we exhausted this scanner before calling close, write out the scan metrics
372       writeScanMetrics();
373       return null;
374     }
375 
376   @VisibleForTesting
377   public int getCacheSize() {
378     return cache != null ? cache.size() : 0;
379   }
380 
381   /**
382    * Contact the servers to load more {@link Result}s in the cache.
383    */
384   protected void loadCache() throws IOException {
385     Result[] values = null;
386     long remainingResultSize = maxScannerResultSize;
387     int countdown = this.caching;
388 
389     // We need to reset it if it's a new callable that was created
390     // with a countdown in nextScanner
391     callable.setCaching(this.caching);
392     // This flag is set when we want to skip the result returned. We do
393     // this when we reset scanner because it split under us.
394     boolean retryAfterOutOfOrderException = true;
395     // We don't expect that the server will have more results for us if
396     // it doesn't tell us otherwise. We rely on the size or count of results
397     boolean serverHasMoreResults = false;
398     do {
399       try {
400         // Server returns a null values if scanning is to stop. Else,
401         // returns an empty array if scanning is to go on and we've just
402         // exhausted current region.
403         values = call(callable, caller, scannerTimeout);
404         // When the replica switch happens, we need to do certain operations
405         // again. The callable will openScanner with the right startkey
406         // but we need to pick up from there. Bypass the rest of the loop
407         // and let the catch-up happen in the beginning of the loop as it
408         // happens for the cases where we see exceptions. Since only openScanner
409         // would have happened, values would be null
410         if (values == null && callable.switchedToADifferentReplica()) {
411           // Any accumulated partial results are no longer valid since the callable will
412           // openScanner with the correct startkey and we must pick up from there
413           clearPartialResults();
414           this.currentRegion = callable.getHRegionInfo();
415           continue;
416         }
417         retryAfterOutOfOrderException = true;
418       } catch (DoNotRetryIOException | NeedUnmanagedConnectionException e) {
419         // An exception was thrown which makes any partial results that we were collecting
420         // invalid. The scanner will need to be reset to the beginning of a row.
421         clearPartialResults();
422 
423         // DNRIOEs are thrown to make us break out of retries. Some types of DNRIOEs want us
424         // to reset the scanner and come back in again.
425         if (e instanceof UnknownScannerException) {
426           long timeout = lastNext + scannerTimeout;
427           // If we are over the timeout, throw this exception to the client wrapped in
428           // a ScannerTimeoutException. Else, it's because the region moved and we used the old
429           // id against the new region server; reset the scanner.
430           if (timeout < System.currentTimeMillis()) {
431             LOG.info("For hints related to the following exception, please try taking a look at: " +
432                 "https://hbase.apache.org/book.html#trouble.client.scantimeout");
433             long elapsed = System.currentTimeMillis() - lastNext;
434             ScannerTimeoutException ex =
435                 new ScannerTimeoutException(elapsed + "ms passed since the last invocation, "
436                     + "timeout is currently set to " + scannerTimeout);
437             ex.initCause(e);
438             throw ex;
439           }
440         } else {
441           // If exception is any but the list below throw it back to the client; else setup
442           // the scanner and retry.
443           Throwable cause = e.getCause();
444           if ((cause != null && cause instanceof NotServingRegionException) ||
445               (cause != null && cause instanceof RegionServerStoppedException) ||
446               e instanceof OutOfOrderScannerNextException) {
447             // Pass
448             // It is easier writing the if loop test as list of what is allowed rather than
449             // as a list of what is not allowed... so if in here, it means we do not throw.
450           } else {
451             throw e;
452           }
453         }
454         // Else, its signal from depths of ScannerCallable that we need to reset the scanner.
455         if (this.lastResult != null) {
456           // The region has moved. We need to open a brand new scanner at
457           // the new location.
458           // Reset the startRow to the row we've seen last so that the new
459           // scanner starts at the correct row. Otherwise we may see previously
460           // returned rows again.
461           // (ScannerCallable by now has "relocated" the correct region)
462           if (scan.isReversed()) {
463             scan.setStartRow(createClosestRowBefore(lastResult.getRow()));
464           } else {
465             scan.setStartRow(Bytes.add(lastResult.getRow(), new byte[1]));
466           }
467         }
468         if (e instanceof OutOfOrderScannerNextException) {
469           if (retryAfterOutOfOrderException) {
470             retryAfterOutOfOrderException = false;
471           } else {
472             // TODO: Why wrap this in a DNRIOE when it already is a DNRIOE?
473             throw new DoNotRetryIOException("Failed after retry of " +
474                 "OutOfOrderScannerNextException: was there a rpc timeout?", e);
475           }
476         }
477         // Clear region.
478         this.currentRegion = null;
479         // Set this to zero so we don't try and do an rpc and close on remote server when
480         // the exception we got was UnknownScanner or the Server is going down.
481         callable = null;
482 
483         // This continue will take us to while at end of loop where we will set up new scanner.
484         continue;
485       }
486       long currentTime = System.currentTimeMillis();
487       if (this.scanMetrics != null) {
488         this.scanMetrics.sumOfMillisSecBetweenNexts.addAndGet(currentTime - lastNext);
489       }
490       lastNext = currentTime;
491       // Groom the array of Results that we received back from the server before adding that
492       // Results to the scanner's cache. If partial results are not allowed to be seen by the
493       // caller, all book keeping will be performed within this method.
494       List<Result> resultsToAddToCache =
495           getResultsToAddToCache(values, callable.isHeartbeatMessage());
496       if (!resultsToAddToCache.isEmpty()) {
497         for (Result rs : resultsToAddToCache) {
498           cache.add(rs);
499           // We don't make Iterator here
500           for (Cell cell : rs.rawCells()) {
501             remainingResultSize -= CellUtil.estimatedHeapSizeOf(cell);
502           }
503           countdown--;
504           this.lastResult = rs;
505         }
506       }
507 
508       // Caller of this method just wants a Result. If we see a heartbeat message, it means
509       // processing of the scan is taking a long time server side. Rather than continue to
510       // loop until a limit (e.g. size or caching) is reached, break out early to avoid causing
511       // unnecesary delays to the caller
512       if (callable.isHeartbeatMessage() && cache.size() > 0) {
513         if (LOG.isTraceEnabled()) {
514           LOG.trace("Heartbeat message received and cache contains Results."
515               + " Breaking out of scan loop");
516         }
517         break;
518       }
519 
520       // We expect that the server won't have more results for us when we exhaust
521       // the size (bytes or count) of the results returned. If the server *does* inform us that
522       // there are more results, we want to avoid possiblyNextScanner(...). Only when we actually
523       // get results is the moreResults context valid.
524       if (null != values && values.length > 0 && callable.hasMoreResultsContext()) {
525         // Only adhere to more server results when we don't have any partialResults
526         // as it keeps the outer loop logic the same.
527         serverHasMoreResults = callable.getServerHasMoreResults() & partialResults.isEmpty();
528       }
529       // Values == null means server-side filter has determined we must STOP
530       // !partialResults.isEmpty() means that we are still accumulating partial Results for a
531       // row. We should not change scanners before we receive all the partial Results for that
532       // row.
533     } while (doneWithRegion(remainingResultSize, countdown, serverHasMoreResults)
534         && (!partialResults.isEmpty() || possiblyNextScanner(countdown, values == null)));
535   }
536 
537   /**
538    * @param remainingResultSize
539    * @param remainingRows
540    * @param regionHasMoreResults
541    * @return true when the current region has been exhausted. When the current region has been
542    *         exhausted, the region must be changed before scanning can continue
543    */
544   private boolean doneWithRegion(long remainingResultSize, int remainingRows,
545       boolean regionHasMoreResults) {
546     return remainingResultSize > 0 && remainingRows > 0 && !regionHasMoreResults;
547   }
548 
549   /**
550    * This method ensures all of our book keeping regarding partial results is kept up to date. This
551    * method should be called once we know that the results we received back from the RPC request do
552    * not contain errors. We return a list of results that should be added to the cache. In general,
553    * this list will contain all NON-partial results from the input array (unless the client has
554    * specified that they are okay with receiving partial results)
555    * @param resultsFromServer The array of {@link Result}s returned from the server
556    * @param heartbeatMessage Flag indicating whether or not the response received from the server
557    *          represented a complete response, or a heartbeat message that was sent to keep the
558    *          client-server connection alive
559    * @return the list of results that should be added to the cache.
560    * @throws IOException
561    */
562   protected List<Result>
563       getResultsToAddToCache(Result[] resultsFromServer, boolean heartbeatMessage)
564           throws IOException {
565     int resultSize = resultsFromServer != null ? resultsFromServer.length : 0;
566     List<Result> resultsToAddToCache = new ArrayList<Result>(resultSize);
567 
568     final boolean isBatchSet = scan != null && scan.getBatch() > 0;
569     final boolean allowPartials = scan != null && scan.getAllowPartialResults();
570 
571     // If the caller has indicated in their scan that they are okay with seeing partial results,
572     // then simply add all results to the list. Note that since scan batching also returns results
573     // for a row in pieces we treat batch being set as equivalent to allowing partials. The
574     // implication of treating batching as equivalent to partial results is that it is possible
575     // the caller will receive a result back where the number of cells in the result is less than
576     // the batch size even though it may not be the last group of cells for that row.
577     if (allowPartials || isBatchSet) {
578       addResultsToList(resultsToAddToCache, resultsFromServer, 0,
579           (null == resultsFromServer ? 0 : resultsFromServer.length));
580       return resultsToAddToCache;
581     }
582 
583     // If no results were returned it indicates that either we have the all the partial results
584     // necessary to construct the complete result or the server had to send a heartbeat message
585     // to the client to keep the client-server connection alive
586     if (resultsFromServer == null || resultsFromServer.length == 0) {
587       // If this response was an empty heartbeat message, then we have not exhausted the region
588       // and thus there may be more partials server side that still need to be added to the partial
589       // list before we form the complete Result
590       if (!partialResults.isEmpty() && !heartbeatMessage) {
591         resultsToAddToCache.add(Result.createCompleteResult(partialResults));
592         clearPartialResults();
593       }
594 
595       return resultsToAddToCache;
596     }
597 
598     // In every RPC response there should be at most a single partial result. Furthermore, if
599     // there is a partial result, it is guaranteed to be in the last position of the array.
600     Result last = resultsFromServer[resultsFromServer.length - 1];
601     Result partial = last.isPartial() ? last : null;
602 
603     if (LOG.isTraceEnabled()) {
604       StringBuilder sb = new StringBuilder();
605       sb.append("number results from RPC: ").append(resultsFromServer.length).append(",");
606       sb.append("partial != null: ").append(partial != null).append(",");
607       sb.append("number of partials so far: ").append(partialResults.size());
608       LOG.trace(sb.toString());
609     }
610 
611     // There are three possibilities cases that can occur while handling partial results
612     //
613     // 1. (partial != null && partialResults.isEmpty())
614     // This is the first partial result that we have received. It should be added to
615     // the list of partialResults and await the next RPC request at which point another
616     // portion of the complete result will be received
617     //
618     // 2. !partialResults.isEmpty()
619     // Since our partialResults list is not empty it means that we have been accumulating partial
620     // Results for a particular row. We cannot form the complete/whole Result for that row until
621     // all partials for the row have been received. Thus we loop through all of the Results
622     // returned from the server and determine whether or not all partial Results for the row have
623     // been received. We know that we have received all of the partial Results for the row when:
624     // i) We notice a row change in the Results
625     // ii) We see a Result for the partial row that is NOT marked as a partial Result
626     //
627     // 3. (partial == null && partialResults.isEmpty())
628     // Business as usual. We are not accumulating partial results and there wasn't a partial result
629     // in the RPC response. This means that all of the results we received from the server are
630     // complete and can be added directly to the cache
631     if (partial != null && partialResults.isEmpty()) {
632       addToPartialResults(partial);
633 
634       // Exclude the last result, it's a partial
635       addResultsToList(resultsToAddToCache, resultsFromServer, 0, resultsFromServer.length - 1);
636     } else if (!partialResults.isEmpty()) {
637       for (int i = 0; i < resultsFromServer.length; i++) {
638         Result result = resultsFromServer[i];
639 
640         // This result is from the same row as the partial Results. Add it to the list of partials
641         // and check if it was the last partial Result for that row
642         if (Bytes.equals(partialResultsRow, result.getRow())) {
643           addToPartialResults(result);
644 
645           // If the result is not a partial, it is a signal to us that it is the last Result we
646           // need to form the complete Result client-side
647           if (!result.isPartial()) {
648             resultsToAddToCache.add(Result.createCompleteResult(partialResults));
649             clearPartialResults();
650           }
651         } else {
652           // The row of this result differs from the row of the partial results we have received so
653           // far. If our list of partials isn't empty, this is a signal to form the complete Result
654           // since the row has now changed
655           if (!partialResults.isEmpty()) {
656             resultsToAddToCache.add(Result.createCompleteResult(partialResults));
657             clearPartialResults();
658           }
659 
660           // It's possible that in one response from the server we receive the final partial for
661           // one row and receive a partial for a different row. Thus, make sure that all Results
662           // are added to the proper list
663           if (result.isPartial()) {
664             addToPartialResults(result);
665           } else {
666             resultsToAddToCache.add(result);
667           }
668         }
669       }
670     } else { // partial == null && partialResults.isEmpty() -- business as usual
671       addResultsToList(resultsToAddToCache, resultsFromServer, 0, resultsFromServer.length);
672     }
673 
674     return resultsToAddToCache;
675   }
676 
677   /**
678    * A convenience method for adding a Result to our list of partials. This method ensure that only
679    * Results that belong to the same row as the other partials can be added to the list.
680    * @param result The result that we want to add to our list of partial Results
681    * @throws IOException
682    */
683   private void addToPartialResults(final Result result) throws IOException {
684     final byte[] row = result.getRow();
685     if (partialResultsRow != null && !Bytes.equals(row, partialResultsRow)) {
686       throw new IOException("Partial result row does not match. All partial results must come "
687           + "from the same row. partialResultsRow: " + Bytes.toString(partialResultsRow) + "row: "
688           + Bytes.toString(row));
689     }
690     partialResultsRow = row;
691     partialResults.add(result);
692   }
693 
694   /**
695    * Convenience method for clearing the list of partials and resetting the partialResultsRow.
696    */
697   private void clearPartialResults() {
698     partialResults.clear();
699     partialResultsRow = null;
700   }
701 
702   /**
703    * Helper method for adding results between the indices [start, end) to the outputList
704    * @param outputList the list that results will be added to
705    * @param inputArray the array that results are taken from
706    * @param start beginning index (inclusive)
707    * @param end ending index (exclusive)
708    */
709   private void addResultsToList(List<Result> outputList, Result[] inputArray, int start, int end) {
710     if (inputArray == null || start < 0 || end > inputArray.length) return;
711 
712     for (int i = start; i < end; i++) {
713       outputList.add(inputArray[i]);
714     }
715   }
716 
717     @Override
718     public void close() {
719       if (!scanMetricsPublished) writeScanMetrics();
720       if (callable != null) {
721         callable.setClose();
722         try {
723           call(callable, caller, scannerTimeout);
724         } catch (UnknownScannerException e) {
725            // We used to catch this error, interpret, and rethrow. However, we
726            // have since decided that it's not nice for a scanner's close to
727            // throw exceptions. Chances are it was just due to lease time out.
728         } catch (IOException e) {
729            /* An exception other than UnknownScanner is unexpected. */
730            LOG.warn("scanner failed to close. Exception follows: " + e);
731         }
732         callable = null;
733       }
734       closed = true;
735     }
736 
737   /**
738    * Create the closest row before the specified row
739    * @param row
740    * @return a new byte array which is the closest front row of the specified one
741    */
742   protected static byte[] createClosestRowBefore(byte[] row) {
743     if (row == null) {
744       throw new IllegalArgumentException("The passed row is empty");
745     }
746     if (Bytes.equals(row, HConstants.EMPTY_BYTE_ARRAY)) {
747       return MAX_BYTE_ARRAY;
748     }
749     if (row[row.length - 1] == 0) {
750       return Arrays.copyOf(row, row.length - 1);
751     } else {
752       byte[] closestFrontRow = Arrays.copyOf(row, row.length);
753       closestFrontRow[row.length - 1] = (byte) ((closestFrontRow[row.length - 1] & 0xff) - 1);
754       closestFrontRow = Bytes.add(closestFrontRow, MAX_BYTE_ARRAY);
755       return closestFrontRow;
756     }
757   }
758 
759   @Override
760   public boolean renewLease() {
761     if (callable != null) {
762       // do not return any rows, do not advance the scanner
763       callable.setRenew(true);
764       try {
765         this.caller.callWithoutRetries(callable, this.scannerTimeout);
766       } catch (Exception e) {
767         return false;
768       } finally {
769         callable.setRenew(false);
770       }
771       return true;
772     }
773     return false;
774   }
775 }