View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  package org.apache.hadoop.hbase.regionserver;
20  
21  import java.io.IOException;
22  import java.io.InterruptedIOException;
23  import java.net.InetSocketAddress;
24  import java.net.UnknownHostException;
25  import java.util.ArrayList;
26  import java.util.Collections;
27  import java.util.HashMap;
28  import java.util.Iterator;
29  import java.util.List;
30  import java.util.Map;
31  import java.util.NavigableMap;
32  import java.util.Set;
33  import java.util.TreeSet;
34  import java.util.concurrent.ConcurrentHashMap;
35  import java.util.concurrent.atomic.AtomicLong;
36  
37  import org.apache.commons.logging.Log;
38  import org.apache.commons.logging.LogFactory;
39  import org.apache.hadoop.conf.Configuration;
40  import org.apache.hadoop.hbase.Cell;
41  import org.apache.hadoop.hbase.CellScannable;
42  import org.apache.hadoop.hbase.CellScanner;
43  import org.apache.hadoop.hbase.CellUtil;
44  import org.apache.hadoop.hbase.DoNotRetryIOException;
45  import org.apache.hadoop.hbase.DroppedSnapshotException;
46  import org.apache.hadoop.hbase.HBaseIOException;
47  import org.apache.hadoop.hbase.HConstants;
48  import org.apache.hadoop.hbase.HRegionInfo;
49  import org.apache.hadoop.hbase.HTableDescriptor;
50  import org.apache.hadoop.hbase.MetaTableAccessor;
51  import org.apache.hadoop.hbase.NotServingRegionException;
52  import org.apache.hadoop.hbase.ServerName;
53  import org.apache.hadoop.hbase.TableName;
54  import org.apache.hadoop.hbase.UnknownScannerException;
55  import org.apache.hadoop.hbase.classification.InterfaceAudience;
56  import org.apache.hadoop.hbase.client.Append;
57  import org.apache.hadoop.hbase.client.ConnectionUtils;
58  import org.apache.hadoop.hbase.client.Delete;
59  import org.apache.hadoop.hbase.client.Durability;
60  import org.apache.hadoop.hbase.client.Get;
61  import org.apache.hadoop.hbase.client.Increment;
62  import org.apache.hadoop.hbase.client.Mutation;
63  import org.apache.hadoop.hbase.client.Put;
64  import org.apache.hadoop.hbase.client.RegionReplicaUtil;
65  import org.apache.hadoop.hbase.client.Result;
66  import org.apache.hadoop.hbase.client.RowMutations;
67  import org.apache.hadoop.hbase.client.Scan;
68  import org.apache.hadoop.hbase.coordination.CloseRegionCoordination;
69  import org.apache.hadoop.hbase.coordination.OpenRegionCoordination;
70  import org.apache.hadoop.hbase.exceptions.FailedSanityCheckException;
71  import org.apache.hadoop.hbase.exceptions.MergeRegionException;
72  import org.apache.hadoop.hbase.exceptions.OperationConflictException;
73  import org.apache.hadoop.hbase.exceptions.OutOfOrderScannerNextException;
74  import org.apache.hadoop.hbase.filter.ByteArrayComparable;
75  import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
76  import org.apache.hadoop.hbase.ipc.HBaseRPCErrorHandler;
77  import org.apache.hadoop.hbase.ipc.PayloadCarryingRpcController;
78  import org.apache.hadoop.hbase.ipc.PriorityFunction;
79  import org.apache.hadoop.hbase.ipc.QosPriority;
80  import org.apache.hadoop.hbase.ipc.RpcCallContext;
81  import org.apache.hadoop.hbase.ipc.RpcServer;
82  import org.apache.hadoop.hbase.ipc.RpcServer.BlockingServiceAndInterface;
83  import org.apache.hadoop.hbase.ipc.RpcServerInterface;
84  import org.apache.hadoop.hbase.ipc.ServerNotRunningYetException;
85  import org.apache.hadoop.hbase.ipc.ServerRpcController;
86  import org.apache.hadoop.hbase.master.MasterRpcServices;
87  import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
88  import org.apache.hadoop.hbase.protobuf.RequestConverter;
89  import org.apache.hadoop.hbase.protobuf.ResponseConverter;
90  import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.AdminService;
91  import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.CloseRegionRequest;
92  import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.CloseRegionResponse;
93  import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.CompactRegionRequest;
94  import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.CompactRegionResponse;
95  import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.FlushRegionRequest;
96  import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.FlushRegionResponse;
97  import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.GetOnlineRegionRequest;
98  import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.GetOnlineRegionResponse;
99  import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.GetRegionInfoRequest;
100 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.GetRegionInfoResponse;
101 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.GetServerInfoRequest;
102 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.GetServerInfoResponse;
103 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.GetStoreFileRequest;
104 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.GetStoreFileResponse;
105 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.MergeRegionsRequest;
106 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.MergeRegionsResponse;
107 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.OpenRegionRequest;
108 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.OpenRegionRequest.RegionOpenInfo;
109 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.OpenRegionResponse;
110 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.OpenRegionResponse.RegionOpeningState;
111 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.ReplicateWALEntryRequest;
112 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.ReplicateWALEntryResponse;
113 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.RollWALWriterRequest;
114 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.RollWALWriterResponse;
115 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.SplitRegionRequest;
116 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.SplitRegionResponse;
117 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.StopServerRequest;
118 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.StopServerResponse;
119 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.UpdateConfigurationRequest;
120 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.UpdateConfigurationResponse;
121 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.UpdateFavoredNodesRequest;
122 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.UpdateFavoredNodesResponse;
123 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.WALEntry;
124 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.WarmupRegionRequest;
125 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.WarmupRegionResponse;
126 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
127 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.BulkLoadHFileRequest;
128 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.BulkLoadHFileRequest.FamilyPath;
129 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.BulkLoadHFileResponse;
130 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService;
131 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.Condition;
132 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.CoprocessorServiceRequest;
133 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.CoprocessorServiceResponse;
134 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.GetRequest;
135 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.GetResponse;
136 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MultiRequest;
137 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MultiResponse;
138 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutateRequest;
139 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutateResponse;
140 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProto;
141 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProto.MutationType;
142 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.RegionAction;
143 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.RegionActionResult;
144 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ResultOrException;
145 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ScanRequest;
146 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ScanResponse;
147 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.RegionInfo;
148 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.RegionSpecifier;
149 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.RegionSpecifier.RegionSpecifierType;
150 import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader;
151 import org.apache.hadoop.hbase.protobuf.generated.WALProtos.BulkLoadDescriptor;
152 import org.apache.hadoop.hbase.protobuf.generated.WALProtos.CompactionDescriptor;
153 import org.apache.hadoop.hbase.protobuf.generated.WALProtos.FlushDescriptor;
154 import org.apache.hadoop.hbase.protobuf.generated.WALProtos.RegionEventDescriptor;
155 import org.apache.hadoop.hbase.quotas.OperationQuota;
156 import org.apache.hadoop.hbase.quotas.RegionServerQuotaManager;
157 import org.apache.hadoop.hbase.regionserver.Leases.LeaseStillHeldException;
158 import org.apache.hadoop.hbase.regionserver.Region.FlushResult;
159 import org.apache.hadoop.hbase.regionserver.Region.Operation;
160 import org.apache.hadoop.hbase.regionserver.ScannerContext.LimitScope;
161 import org.apache.hadoop.hbase.regionserver.handler.OpenMetaHandler;
162 import org.apache.hadoop.hbase.regionserver.handler.OpenRegionHandler;
163 import org.apache.hadoop.hbase.wal.WAL;
164 import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
165 import org.apache.hadoop.hbase.util.Bytes;
166 import org.apache.hadoop.hbase.util.Counter;
167 import org.apache.hadoop.hbase.util.DNS;
168 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
169 import org.apache.hadoop.hbase.util.Pair;
170 import org.apache.hadoop.hbase.util.ServerRegionReplicaUtil;
171 import org.apache.hadoop.hbase.util.Strings;
172 import org.apache.hadoop.hbase.wal.WALKey;
173 import org.apache.hadoop.hbase.wal.WALSplitter;
174 import org.apache.hadoop.hbase.zookeeper.ZKSplitLog;
175 import org.apache.zookeeper.KeeperException;
176 
177 import com.google.common.annotations.VisibleForTesting;
178 import com.google.protobuf.ByteString;
179 import com.google.protobuf.Message;
180 import com.google.protobuf.RpcController;
181 import com.google.protobuf.ServiceException;
182 import com.google.protobuf.TextFormat;
183 
184 /**
185  * Implements the regionserver RPC services.
186  */
187 @InterfaceAudience.Private
188 @SuppressWarnings("deprecation")
189 public class RSRpcServices implements HBaseRPCErrorHandler,
190     AdminService.BlockingInterface, ClientService.BlockingInterface, PriorityFunction {
191   protected static final Log LOG = LogFactory.getLog(RSRpcServices.class);
192 
193   /** RPC scheduler to use for the region server. */
194   public static final String REGION_SERVER_RPC_SCHEDULER_FACTORY_CLASS =
195     "hbase.region.server.rpc.scheduler.factory.class";
196 
197   /**
198    * Minimum allowable time limit delta (in milliseconds) that can be enforced during scans. This
199    * configuration exists to prevent the scenario where a time limit is specified to be so
200    * restrictive that the time limit is reached immediately (before any cells are scanned).
201    */
202   private static final String REGION_SERVER_RPC_MINIMUM_SCAN_TIME_LIMIT_DELTA =
203       "hbase.region.server.rpc.minimum.scan.time.limit.delta";
204   /**
205    * Default value of {@link RSRpcServices#REGION_SERVER_RPC_MINIMUM_SCAN_TIME_LIMIT_DELTA}
206    */
207   private static final long DEFAULT_REGION_SERVER_RPC_MINIMUM_SCAN_TIME_LIMIT_DELTA = 10;
208 
209   // Request counter. (Includes requests that are not serviced by regions.)
210   final Counter requestCount = new Counter();
211   // Server to handle client requests.
212   final RpcServerInterface rpcServer;
213   final InetSocketAddress isa;
214 
215   private final HRegionServer regionServer;
216   private final long maxScannerResultSize;
217 
218   // The reference to the priority extraction function
219   private final PriorityFunction priority;
220 
221   private final AtomicLong scannerIdGen = new AtomicLong(0L);
222   private final ConcurrentHashMap<String, RegionScannerHolder> scanners =
223     new ConcurrentHashMap<String, RegionScannerHolder>();
224 
225   /**
226    * The lease timeout period for client scanners (milliseconds).
227    */
228   private final int scannerLeaseTimeoutPeriod;
229 
230   /**
231    * The RPC timeout period (milliseconds)
232    */
233   private final int rpcTimeout;
234 
235   /**
236    * The minimum allowable delta to use for the scan limit
237    */
238   private final long minimumScanTimeLimitDelta;
239 
240   /**
241    * Holder class which holds the RegionScanner and nextCallSeq together.
242    */
243   private static class RegionScannerHolder {
244     private AtomicLong nextCallSeq = new AtomicLong(0);
245     private RegionScanner s;
246     private Region r;
247 
248     public RegionScannerHolder(RegionScanner s, Region r) {
249       this.s = s;
250       this.r = r;
251     }
252 
253     private long getNextCallSeq() {
254       return nextCallSeq.get();
255     }
256 
257     private void incNextCallSeq() {
258       nextCallSeq.incrementAndGet();
259     }
260 
261     private void rollbackNextCallSeq() {
262       nextCallSeq.decrementAndGet();
263     }
264   }
265 
266   /**
267    * Instantiated as a scanner lease. If the lease times out, the scanner is
268    * closed
269    */
270   private class ScannerListener implements LeaseListener {
271     private final String scannerName;
272 
273     ScannerListener(final String n) {
274       this.scannerName = n;
275     }
276 
277     @Override
278     public void leaseExpired() {
279       RegionScannerHolder rsh = scanners.remove(this.scannerName);
280       if (rsh != null) {
281         RegionScanner s = rsh.s;
282         LOG.info("Scanner " + this.scannerName + " lease expired on region "
283           + s.getRegionInfo().getRegionNameAsString());
284         try {
285           Region region = regionServer.getRegion(s.getRegionInfo().getRegionName());
286           if (region != null && region.getCoprocessorHost() != null) {
287             region.getCoprocessorHost().preScannerClose(s);
288           }
289 
290           s.close();
291           if (region != null && region.getCoprocessorHost() != null) {
292             region.getCoprocessorHost().postScannerClose(s);
293           }
294         } catch (IOException e) {
295           LOG.error("Closing scanner for "
296             + s.getRegionInfo().getRegionNameAsString(), e);
297         }
298       } else {
299         LOG.warn("Scanner " + this.scannerName + " lease expired, but no related" +
300           " scanner found, hence no chance to close that related scanner!");
301       }
302     }
303   }
304 
305   private static ResultOrException getResultOrException(
306       final ClientProtos.Result r, final int index, final ClientProtos.RegionLoadStats stats) {
307     return getResultOrException(ResponseConverter.buildActionResult(r, stats), index);
308   }
309 
310   private static ResultOrException getResultOrException(final Exception e, final int index) {
311     return getResultOrException(ResponseConverter.buildActionResult(e), index);
312   }
313 
314   private static ResultOrException getResultOrException(
315       final ResultOrException.Builder builder, final int index) {
316     return builder.setIndex(index).build();
317   }
318 
319   /**
320    * Starts the nonce operation for a mutation, if needed.
321    * @param mutation Mutation.
322    * @param nonceGroup Nonce group from the request.
323    * @returns Nonce used (can be NO_NONCE).
324    */
325   private long startNonceOperation(final MutationProto mutation, long nonceGroup)
326       throws IOException, OperationConflictException {
327     if (regionServer.nonceManager == null || !mutation.hasNonce()) return HConstants.NO_NONCE;
328     boolean canProceed = false;
329     try {
330       canProceed = regionServer.nonceManager.startOperation(
331         nonceGroup, mutation.getNonce(), regionServer);
332     } catch (InterruptedException ex) {
333       throw new InterruptedIOException("Nonce start operation interrupted");
334     }
335     if (!canProceed) {
336       // TODO: instead, we could convert append/increment to get w/mvcc
337       String message = "The operation with nonce {" + nonceGroup + ", " + mutation.getNonce()
338         + "} on row [" + Bytes.toString(mutation.getRow().toByteArray())
339         + "] may have already completed";
340       throw new OperationConflictException(message);
341     }
342     return mutation.getNonce();
343   }
344 
345   /**
346    * Ends nonce operation for a mutation, if needed.
347    * @param mutation Mutation.
348    * @param nonceGroup Nonce group from the request. Always 0 in initial implementation.
349    * @param success Whether the operation for this nonce has succeeded.
350    */
351   private void endNonceOperation(final MutationProto mutation,
352       long nonceGroup, boolean success) {
353     if (regionServer.nonceManager != null && mutation.hasNonce()) {
354       regionServer.nonceManager.endOperation(nonceGroup, mutation.getNonce(), success);
355     }
356   }
357 
358   /**
359    * @return True if current call supports cellblocks
360    */
361   private boolean isClientCellBlockSupport() {
362     RpcCallContext context = RpcServer.getCurrentCall();
363     return context != null && context.isClientCellBlockSupport();
364   }
365 
366   private void addResult(final MutateResponse.Builder builder,
367       final Result result, final PayloadCarryingRpcController rpcc) {
368     if (result == null) return;
369     if (isClientCellBlockSupport()) {
370       builder.setResult(ProtobufUtil.toResultNoData(result));
371       rpcc.setCellScanner(result.cellScanner());
372     } else {
373       ClientProtos.Result pbr = ProtobufUtil.toResult(result);
374       builder.setResult(pbr);
375     }
376   }
377 
378   private void addResults(final ScanResponse.Builder builder, final List<Result> results,
379       final RpcController controller, boolean isDefaultRegion) {
380     builder.setStale(!isDefaultRegion);
381     if (results == null || results.isEmpty()) return;
382     if (isClientCellBlockSupport()) {
383       for (Result res : results) {
384         builder.addCellsPerResult(res.size());
385         builder.addPartialFlagPerResult(res.isPartial());
386       }
387       ((PayloadCarryingRpcController)controller).
388         setCellScanner(CellUtil.createCellScanner(results));
389     } else {
390       for (Result res: results) {
391         ClientProtos.Result pbr = ProtobufUtil.toResult(res);
392         builder.addResults(pbr);
393       }
394     }
395   }
396 
397   /**
398    * Mutate a list of rows atomically.
399    *
400    * @param region
401    * @param actions
402    * @param cellScanner if non-null, the mutation data -- the Cell content.
403    * @throws IOException
404    */
405   private ClientProtos.RegionLoadStats mutateRows(final Region region,
406       final List<ClientProtos.Action> actions,
407       final CellScanner cellScanner) throws IOException {
408     if (!region.getRegionInfo().isMetaTable()) {
409       regionServer.cacheFlusher.reclaimMemStoreMemory();
410     }
411     RowMutations rm = null;
412     for (ClientProtos.Action action: actions) {
413       if (action.hasGet()) {
414         throw new DoNotRetryIOException("Atomic put and/or delete only, not a Get=" +
415           action.getGet());
416       }
417       MutationType type = action.getMutation().getMutateType();
418       if (rm == null) {
419         rm = new RowMutations(action.getMutation().getRow().toByteArray());
420       }
421       switch (type) {
422       case PUT:
423         rm.add(ProtobufUtil.toPut(action.getMutation(), cellScanner));
424         break;
425       case DELETE:
426         rm.add(ProtobufUtil.toDelete(action.getMutation(), cellScanner));
427         break;
428       default:
429           throw new DoNotRetryIOException("Atomic put and/or delete only, not " + type.name());
430       }
431     }
432     region.mutateRow(rm);
433     return ((HRegion)region).getRegionStats();
434   }
435 
436   /**
437    * Mutate a list of rows atomically.
438    *
439    * @param region
440    * @param actions
441    * @param cellScanner if non-null, the mutation data -- the Cell content.
442    * @param row
443    * @param family
444    * @param qualifier
445    * @param compareOp
446    * @param comparator @throws IOException
447    */
448   private boolean checkAndRowMutate(final Region region, final List<ClientProtos.Action> actions,
449       final CellScanner cellScanner, byte[] row, byte[] family, byte[] qualifier,
450       CompareOp compareOp, ByteArrayComparable comparator) throws IOException {
451     if (!region.getRegionInfo().isMetaTable()) {
452       regionServer.cacheFlusher.reclaimMemStoreMemory();
453     }
454     RowMutations rm = null;
455     for (ClientProtos.Action action: actions) {
456       if (action.hasGet()) {
457         throw new DoNotRetryIOException("Atomic put and/or delete only, not a Get=" +
458             action.getGet());
459       }
460       MutationType type = action.getMutation().getMutateType();
461       if (rm == null) {
462         rm = new RowMutations(action.getMutation().getRow().toByteArray());
463       }
464       switch (type) {
465       case PUT:
466         rm.add(ProtobufUtil.toPut(action.getMutation(), cellScanner));
467         break;
468       case DELETE:
469         rm.add(ProtobufUtil.toDelete(action.getMutation(), cellScanner));
470         break;
471       default:
472         throw new DoNotRetryIOException("Atomic put and/or delete only, not " + type.name());
473       }
474     }
475     return region.checkAndRowMutate(row, family, qualifier, compareOp, comparator, rm, Boolean.TRUE);
476   }
477 
478   /**
479    * Execute an append mutation.
480    *
481    * @param region
482    * @param m
483    * @param cellScanner
484    * @return result to return to client if default operation should be
485    * bypassed as indicated by RegionObserver, null otherwise
486    * @throws IOException
487    */
488   private Result append(final Region region, final OperationQuota quota, final MutationProto m,
489       final CellScanner cellScanner, long nonceGroup) throws IOException {
490     long before = EnvironmentEdgeManager.currentTime();
491     Append append = ProtobufUtil.toAppend(m, cellScanner);
492     quota.addMutation(append);
493     Result r = null;
494     if (region.getCoprocessorHost() != null) {
495       r = region.getCoprocessorHost().preAppend(append);
496     }
497     if (r == null) {
498       long nonce = startNonceOperation(m, nonceGroup);
499       boolean success = false;
500       try {
501         r = region.append(append, nonceGroup, nonce);
502         success = true;
503       } finally {
504         endNonceOperation(m, nonceGroup, success);
505       }
506       if (region.getCoprocessorHost() != null) {
507         region.getCoprocessorHost().postAppend(append, r);
508       }
509     }
510     if (regionServer.metricsRegionServer != null) {
511       regionServer.metricsRegionServer.updateAppend(
512         EnvironmentEdgeManager.currentTime() - before);
513     }
514     return r;
515   }
516 
517   /**
518    * Execute an increment mutation.
519    *
520    * @param region
521    * @param mutation
522    * @return the Result
523    * @throws IOException
524    */
525   private Result increment(final Region region, final OperationQuota quota,
526       final MutationProto mutation, final CellScanner cells, long nonceGroup) throws IOException {
527     long before = EnvironmentEdgeManager.currentTime();
528     Increment increment = ProtobufUtil.toIncrement(mutation, cells);
529     quota.addMutation(increment);
530     Result r = null;
531     if (region.getCoprocessorHost() != null) {
532       r = region.getCoprocessorHost().preIncrement(increment);
533     }
534     if (r == null) {
535       long nonce = startNonceOperation(mutation, nonceGroup);
536       boolean success = false;
537       try {
538         r = region.increment(increment, nonceGroup, nonce);
539         success = true;
540       } finally {
541         endNonceOperation(mutation, nonceGroup, success);
542       }
543       if (region.getCoprocessorHost() != null) {
544         r = region.getCoprocessorHost().postIncrement(increment, r);
545       }
546     }
547     if (regionServer.metricsRegionServer != null) {
548       regionServer.metricsRegionServer.updateIncrement(
549         EnvironmentEdgeManager.currentTime() - before);
550     }
551     return r;
552   }
553 
554   /**
555    * Run through the regionMutation <code>rm</code> and per Mutation, do the work, and then when
556    * done, add an instance of a {@link ResultOrException} that corresponds to each Mutation.
557    * @param region
558    * @param actions
559    * @param cellScanner
560    * @param builder
561    * @param cellsToReturn  Could be null. May be allocated in this method.  This is what this
562    * method returns as a 'result'.
563    * @return Return the <code>cellScanner</code> passed
564    */
565   private List<CellScannable> doNonAtomicRegionMutation(final Region region,
566       final OperationQuota quota, final RegionAction actions, final CellScanner cellScanner,
567       final RegionActionResult.Builder builder, List<CellScannable> cellsToReturn, long nonceGroup) {
568     // Gather up CONTIGUOUS Puts and Deletes in this mutations List.  Idea is that rather than do
569     // one at a time, we instead pass them in batch.  Be aware that the corresponding
570     // ResultOrException instance that matches each Put or Delete is then added down in the
571     // doBatchOp call.  We should be staying aligned though the Put and Delete are deferred/batched
572     List<ClientProtos.Action> mutations = null;
573     for (ClientProtos.Action action: actions.getActionList()) {
574       ClientProtos.ResultOrException.Builder resultOrExceptionBuilder = null;
575       try {
576         Result r = null;
577         if (action.hasGet()) {
578           long before = EnvironmentEdgeManager.currentTime();
579           try {
580             Get get = ProtobufUtil.toGet(action.getGet());
581             r = region.get(get);
582           } finally {
583             if (regionServer.metricsRegionServer != null) {
584               regionServer.metricsRegionServer.updateGet(
585                 EnvironmentEdgeManager.currentTime() - before);
586             }
587           }
588         } else if (action.hasServiceCall()) {
589           resultOrExceptionBuilder = ResultOrException.newBuilder();
590           try {
591             Message result = execServiceOnRegion(region, action.getServiceCall());
592             ClientProtos.CoprocessorServiceResult.Builder serviceResultBuilder =
593                 ClientProtos.CoprocessorServiceResult.newBuilder();
594             resultOrExceptionBuilder.setServiceResult(
595                 serviceResultBuilder.setValue(
596                   serviceResultBuilder.getValueBuilder()
597                     .setName(result.getClass().getName())
598                     .setValue(result.toByteString())));
599           } catch (IOException ioe) {
600             rpcServer.getMetrics().exception(ioe);
601             resultOrExceptionBuilder.setException(ResponseConverter.buildException(ioe));
602           }
603         } else if (action.hasMutation()) {
604           MutationType type = action.getMutation().getMutateType();
605           if (type != MutationType.PUT && type != MutationType.DELETE && mutations != null &&
606               !mutations.isEmpty()) {
607             // Flush out any Puts or Deletes already collected.
608             doBatchOp(builder, region, quota, mutations, cellScanner);
609             mutations.clear();
610           }
611           switch (type) {
612           case APPEND:
613             r = append(region, quota, action.getMutation(), cellScanner, nonceGroup);
614             break;
615           case INCREMENT:
616             r = increment(region, quota, action.getMutation(), cellScanner,  nonceGroup);
617             break;
618           case PUT:
619           case DELETE:
620             // Collect the individual mutations and apply in a batch
621             if (mutations == null) {
622               mutations = new ArrayList<ClientProtos.Action>(actions.getActionCount());
623             }
624             mutations.add(action);
625             break;
626           default:
627             throw new DoNotRetryIOException("Unsupported mutate type: " + type.name());
628           }
629         } else {
630           throw new HBaseIOException("Unexpected Action type");
631         }
632         if (r != null) {
633           ClientProtos.Result pbResult = null;
634           if (isClientCellBlockSupport()) {
635             pbResult = ProtobufUtil.toResultNoData(r);
636             //  Hard to guess the size here.  Just make a rough guess.
637             if (cellsToReturn == null) cellsToReturn = new ArrayList<CellScannable>();
638             cellsToReturn.add(r);
639           } else {
640             pbResult = ProtobufUtil.toResult(r);
641           }
642           resultOrExceptionBuilder =
643             ClientProtos.ResultOrException.newBuilder().setResult(pbResult);
644         }
645         // Could get to here and there was no result and no exception.  Presumes we added
646         // a Put or Delete to the collecting Mutations List for adding later.  In this
647         // case the corresponding ResultOrException instance for the Put or Delete will be added
648         // down in the doBatchOp method call rather than up here.
649       } catch (IOException ie) {
650         rpcServer.getMetrics().exception(ie);
651         resultOrExceptionBuilder = ResultOrException.newBuilder().
652           setException(ResponseConverter.buildException(ie));
653       }
654       if (resultOrExceptionBuilder != null) {
655         // Propagate index.
656         resultOrExceptionBuilder.setIndex(action.getIndex());
657         builder.addResultOrException(resultOrExceptionBuilder.build());
658       }
659     }
660     // Finish up any outstanding mutations
661     if (mutations != null && !mutations.isEmpty()) {
662       doBatchOp(builder, region, quota, mutations, cellScanner);
663     }
664     return cellsToReturn;
665   }
666 
667   /**
668    * Execute a list of Put/Delete mutations.
669    *
670    * @param builder
671    * @param region
672    * @param mutations
673    */
674   private void doBatchOp(final RegionActionResult.Builder builder, final Region region,
675       final OperationQuota quota,
676       final List<ClientProtos.Action> mutations, final CellScanner cells) {
677     Mutation[] mArray = new Mutation[mutations.size()];
678     long before = EnvironmentEdgeManager.currentTime();
679     boolean batchContainsPuts = false, batchContainsDelete = false;
680     try {
681       int i = 0;
682       for (ClientProtos.Action action: mutations) {
683         MutationProto m = action.getMutation();
684         Mutation mutation;
685         if (m.getMutateType() == MutationType.PUT) {
686           mutation = ProtobufUtil.toPut(m, cells);
687           batchContainsPuts = true;
688         } else {
689           mutation = ProtobufUtil.toDelete(m, cells);
690           batchContainsDelete = true;
691         }
692         mArray[i++] = mutation;
693         quota.addMutation(mutation);
694       }
695 
696       if (!region.getRegionInfo().isMetaTable()) {
697         regionServer.cacheFlusher.reclaimMemStoreMemory();
698       }
699 
700       OperationStatus codes[] = region.batchMutate(mArray, HConstants.NO_NONCE,
701         HConstants.NO_NONCE);
702       for (i = 0; i < codes.length; i++) {
703         int index = mutations.get(i).getIndex();
704         Exception e = null;
705         switch (codes[i].getOperationStatusCode()) {
706           case BAD_FAMILY:
707             e = new NoSuchColumnFamilyException(codes[i].getExceptionMsg());
708             builder.addResultOrException(getResultOrException(e, index));
709             break;
710 
711           case SANITY_CHECK_FAILURE:
712             e = new FailedSanityCheckException(codes[i].getExceptionMsg());
713             builder.addResultOrException(getResultOrException(e, index));
714             break;
715 
716           default:
717             e = new DoNotRetryIOException(codes[i].getExceptionMsg());
718             builder.addResultOrException(getResultOrException(e, index));
719             break;
720 
721           case SUCCESS:
722             builder.addResultOrException(getResultOrException(
723               ClientProtos.Result.getDefaultInstance(), index,
724                 ((HRegion)region).getRegionStats()));
725             break;
726         }
727       }
728     } catch (IOException ie) {
729       for (int i = 0; i < mutations.size(); i++) {
730         builder.addResultOrException(getResultOrException(ie, mutations.get(i).getIndex()));
731       }
732     }
733     if (regionServer.metricsRegionServer != null) {
734       long after = EnvironmentEdgeManager.currentTime();
735       if (batchContainsPuts) {
736         regionServer.metricsRegionServer.updatePut(after - before);
737       }
738       if (batchContainsDelete) {
739         regionServer.metricsRegionServer.updateDelete(after - before);
740       }
741     }
742   }
743 
744   /**
745    * Execute a list of Put/Delete mutations. The function returns OperationStatus instead of
746    * constructing MultiResponse to save a possible loop if caller doesn't need MultiResponse.
747    * @param region
748    * @param mutations
749    * @param replaySeqId
750    * @return an array of OperationStatus which internally contains the OperationStatusCode and the
751    *         exceptionMessage if any
752    * @throws IOException
753    */
754   private OperationStatus [] doReplayBatchOp(final Region region,
755       final List<WALSplitter.MutationReplay> mutations, long replaySeqId) throws IOException {
756     long before = EnvironmentEdgeManager.currentTime();
757     boolean batchContainsPuts = false, batchContainsDelete = false;
758     try {
759       for (Iterator<WALSplitter.MutationReplay> it = mutations.iterator(); it.hasNext();) {
760         WALSplitter.MutationReplay m = it.next();
761 
762         if (m.type == MutationType.PUT) {
763           batchContainsPuts = true;
764         } else {
765           batchContainsDelete = true;
766         }
767 
768         NavigableMap<byte[], List<Cell>> map = m.mutation.getFamilyCellMap();
769         List<Cell> metaCells = map.get(WALEdit.METAFAMILY);
770         if (metaCells != null && !metaCells.isEmpty()) {
771           for (Cell metaCell : metaCells) {
772             CompactionDescriptor compactionDesc = WALEdit.getCompaction(metaCell);
773             boolean isDefaultReplica = RegionReplicaUtil.isDefaultReplica(region.getRegionInfo());
774             HRegion hRegion = (HRegion)region;
775             if (compactionDesc != null) {
776               // replay the compaction. Remove the files from stores only if we are the primary
777               // region replica (thus own the files)
778               hRegion.replayWALCompactionMarker(compactionDesc, !isDefaultReplica, isDefaultReplica,
779                 replaySeqId);
780               continue;
781             }
782             FlushDescriptor flushDesc = WALEdit.getFlushDescriptor(metaCell);
783             if (flushDesc != null && !isDefaultReplica) {
784               hRegion.replayWALFlushMarker(flushDesc, replaySeqId);
785               continue;
786             }
787             RegionEventDescriptor regionEvent = WALEdit.getRegionEventDescriptor(metaCell);
788             if (regionEvent != null && !isDefaultReplica) {
789               hRegion.replayWALRegionEventMarker(regionEvent);
790               continue;
791             }
792             BulkLoadDescriptor bulkLoadEvent = WALEdit.getBulkLoadDescriptor(metaCell);
793             if (bulkLoadEvent != null) {
794               hRegion.replayWALBulkLoadEventMarker(bulkLoadEvent);
795               continue;
796             }
797           }
798           it.remove();
799         }
800       }
801       requestCount.add(mutations.size());
802       if (!region.getRegionInfo().isMetaTable()) {
803         regionServer.cacheFlusher.reclaimMemStoreMemory();
804       }
805       return region.batchReplay(mutations.toArray(
806         new WALSplitter.MutationReplay[mutations.size()]), replaySeqId);
807     } finally {
808       if (regionServer.metricsRegionServer != null) {
809         long after = EnvironmentEdgeManager.currentTime();
810           if (batchContainsPuts) {
811           regionServer.metricsRegionServer.updatePut(after - before);
812         }
813         if (batchContainsDelete) {
814           regionServer.metricsRegionServer.updateDelete(after - before);
815         }
816       }
817     }
818   }
819 
820   private void closeAllScanners() {
821     // Close any outstanding scanners. Means they'll get an UnknownScanner
822     // exception next time they come in.
823     for (Map.Entry<String, RegionScannerHolder> e : scanners.entrySet()) {
824       try {
825         e.getValue().s.close();
826       } catch (IOException ioe) {
827         LOG.warn("Closing scanner " + e.getKey(), ioe);
828       }
829     }
830   }
831 
832   public RSRpcServices(HRegionServer rs) throws IOException {
833     regionServer = rs;
834 
835     RpcSchedulerFactory rpcSchedulerFactory;
836     try {
837       Class<?> rpcSchedulerFactoryClass = rs.conf.getClass(
838           REGION_SERVER_RPC_SCHEDULER_FACTORY_CLASS,
839           SimpleRpcSchedulerFactory.class);
840       rpcSchedulerFactory = ((RpcSchedulerFactory) rpcSchedulerFactoryClass.newInstance());
841     } catch (InstantiationException e) {
842       throw new IllegalArgumentException(e);
843     } catch (IllegalAccessException e) {
844       throw new IllegalArgumentException(e);
845     }
846     // Server to handle client requests.
847     InetSocketAddress initialIsa;
848     InetSocketAddress bindAddress;
849     if(this instanceof MasterRpcServices) {
850       String hostname = getHostname(rs.conf, true);
851       int port = rs.conf.getInt(HConstants.MASTER_PORT, HConstants.DEFAULT_MASTER_PORT);
852       // Creation of a HSA will force a resolve.
853       initialIsa = new InetSocketAddress(hostname, port);
854       bindAddress = new InetSocketAddress(rs.conf.get("hbase.master.ipc.address", hostname), port);
855     } else {
856       String hostname = getHostname(rs.conf, false);
857       int port = rs.conf.getInt(HConstants.REGIONSERVER_PORT,
858         HConstants.DEFAULT_REGIONSERVER_PORT);
859       // Creation of a HSA will force a resolve.
860       initialIsa = new InetSocketAddress(hostname, port);
861       bindAddress = new InetSocketAddress(
862         rs.conf.get("hbase.regionserver.ipc.address", hostname), port);
863     }
864     if (initialIsa.getAddress() == null) {
865       throw new IllegalArgumentException("Failed resolve of " + initialIsa);
866     }
867     priority = new AnnotationReadingPriorityFunction(this);
868     String name = rs.getProcessName() + "/" + initialIsa.toString();
869     // Set how many times to retry talking to another server over HConnection.
870     ConnectionUtils.setServerSideHConnectionRetriesConfig(rs.conf, name, LOG);
871     rpcServer = new RpcServer(rs, name, getServices(),
872       bindAddress, // use final bindAddress for this server.
873       rs.conf,
874       rpcSchedulerFactory.create(rs.conf, this, rs));
875 
876     scannerLeaseTimeoutPeriod = rs.conf.getInt(
877       HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD,
878       HConstants.DEFAULT_HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD);
879     maxScannerResultSize = rs.conf.getLong(
880       HConstants.HBASE_SERVER_SCANNER_MAX_RESULT_SIZE_KEY,
881       HConstants.DEFAULT_HBASE_SERVER_SCANNER_MAX_RESULT_SIZE);
882     rpcTimeout = rs.conf.getInt(
883       HConstants.HBASE_RPC_TIMEOUT_KEY,
884       HConstants.DEFAULT_HBASE_RPC_TIMEOUT);
885     minimumScanTimeLimitDelta = rs.conf.getLong(
886       REGION_SERVER_RPC_MINIMUM_SCAN_TIME_LIMIT_DELTA,
887       DEFAULT_REGION_SERVER_RPC_MINIMUM_SCAN_TIME_LIMIT_DELTA);
888 
889     InetSocketAddress address = rpcServer.getListenerAddress();
890     if (address == null) {
891       throw new IOException("Listener channel is closed");
892     }
893     // Set our address, however we need the final port that was given to rpcServer
894     isa = new InetSocketAddress(initialIsa.getHostName(), address.getPort());
895     rpcServer.setErrorHandler(this);
896     rs.setName(name);
897   }
898 
899   public static String getHostname(Configuration conf, boolean isMaster)
900       throws UnknownHostException {
901     String hostname = conf.get(isMaster? HRegionServer.MASTER_HOSTNAME_KEY :
902       HRegionServer.RS_HOSTNAME_KEY);
903     if (hostname == null || hostname.isEmpty()) {
904       String masterOrRS = isMaster ? "master" : "regionserver";
905       return Strings.domainNamePointerToHostName(DNS.getDefaultHost(
906         conf.get("hbase." + masterOrRS + ".dns.interface", "default"),
907         conf.get("hbase." + masterOrRS + ".dns.nameserver", "default")));
908     } else {
909       LOG.info("hostname is configured to be " + hostname);
910       return hostname;
911     }
912   }
913 
914   RegionScanner getScanner(long scannerId) {
915     String scannerIdString = Long.toString(scannerId);
916     RegionScannerHolder scannerHolder = scanners.get(scannerIdString);
917     if (scannerHolder != null) {
918       return scannerHolder.s;
919     }
920     return null;
921   }
922 
923   /**
924    * Get the vtime associated with the scanner.
925    * Currently the vtime is the number of "next" calls.
926    */
927   long getScannerVirtualTime(long scannerId) {
928     String scannerIdString = Long.toString(scannerId);
929     RegionScannerHolder scannerHolder = scanners.get(scannerIdString);
930     if (scannerHolder != null) {
931       return scannerHolder.getNextCallSeq();
932     }
933     return 0L;
934   }
935 
936   long addScanner(RegionScanner s, Region r) throws LeaseStillHeldException {
937     long scannerId = this.scannerIdGen.incrementAndGet();
938     String scannerName = String.valueOf(scannerId);
939 
940     RegionScannerHolder existing =
941       scanners.putIfAbsent(scannerName, new RegionScannerHolder(s, r));
942     assert existing == null : "scannerId must be unique within regionserver's whole lifecycle!";
943 
944     regionServer.leases.createLease(scannerName, this.scannerLeaseTimeoutPeriod,
945         new ScannerListener(scannerName));
946     return scannerId;
947   }
948 
949   /**
950    * Find the HRegion based on a region specifier
951    *
952    * @param regionSpecifier the region specifier
953    * @return the corresponding region
954    * @throws IOException if the specifier is not null,
955    *    but failed to find the region
956    */
957   Region getRegion(
958       final RegionSpecifier regionSpecifier) throws IOException {
959     return regionServer.getRegionByEncodedName(regionSpecifier.getValue().toByteArray(),
960         ProtobufUtil.getRegionEncodedName(regionSpecifier));
961   }
962 
963   @VisibleForTesting
964   public PriorityFunction getPriority() {
965     return priority;
966   }
967 
968   Configuration getConfiguration() {
969     return regionServer.getConfiguration();
970   }
971 
972   private RegionServerQuotaManager getQuotaManager() {
973     return regionServer.getRegionServerQuotaManager();
974   }
975 
976   void start() {
977     rpcServer.start();
978   }
979 
980   void stop() {
981     closeAllScanners();
982     rpcServer.stop();
983   }
984 
985   /**
986    * Called to verify that this server is up and running.
987    *
988    * @throws IOException
989    */
990   protected void checkOpen() throws IOException {
991     if (regionServer.isAborted()) {
992       throw new RegionServerAbortedException("Server " + regionServer.serverName + " aborting");
993     }
994     if (regionServer.isStopped()) {
995       throw new RegionServerStoppedException("Server " + regionServer.serverName + " stopping");
996     }
997     if (!regionServer.fsOk) {
998       throw new RegionServerStoppedException("File system not available");
999     }
1000     if (!regionServer.isOnline()) {
1001       throw new ServerNotRunningYetException("Server is not running yet");
1002     }
1003   }
1004 
1005   /**
1006    * @return list of blocking services and their security info classes that this server supports
1007    */
1008   protected List<BlockingServiceAndInterface> getServices() {
1009     List<BlockingServiceAndInterface> bssi = new ArrayList<BlockingServiceAndInterface>(2);
1010     bssi.add(new BlockingServiceAndInterface(
1011       ClientService.newReflectiveBlockingService(this),
1012       ClientService.BlockingInterface.class));
1013     bssi.add(new BlockingServiceAndInterface(
1014       AdminService.newReflectiveBlockingService(this),
1015       AdminService.BlockingInterface.class));
1016     return bssi;
1017   }
1018 
1019   public InetSocketAddress getSocketAddress() {
1020     return isa;
1021   }
1022 
1023   @Override
1024   public int getPriority(RequestHeader header, Message param) {
1025     return priority.getPriority(header, param);
1026   }
1027 
1028   @Override
1029   public long getDeadline(RequestHeader header, Message param) {
1030     return priority.getDeadline(header, param);
1031   }
1032 
1033   /*
1034    * Check if an OOME and, if so, abort immediately to avoid creating more objects.
1035    *
1036    * @param e
1037    *
1038    * @return True if we OOME'd and are aborting.
1039    */
1040   @Override
1041   public boolean checkOOME(final Throwable e) {
1042     boolean stop = false;
1043     try {
1044       if (e instanceof OutOfMemoryError
1045           || (e.getCause() != null && e.getCause() instanceof OutOfMemoryError)
1046           || (e.getMessage() != null && e.getMessage().contains(
1047               "java.lang.OutOfMemoryError"))) {
1048         stop = true;
1049         LOG.fatal("Run out of memory; " + getClass().getSimpleName()
1050           + " will abort itself immediately", e);
1051       }
1052     } finally {
1053       if (stop) {
1054         Runtime.getRuntime().halt(1);
1055       }
1056     }
1057     return stop;
1058   }
1059 
1060   /**
1061    * Close a region on the region server.
1062    *
1063    * @param controller the RPC controller
1064    * @param request the request
1065    * @throws ServiceException
1066    */
1067   @Override
1068   @QosPriority(priority=HConstants.ADMIN_QOS)
1069   public CloseRegionResponse closeRegion(final RpcController controller,
1070       final CloseRegionRequest request) throws ServiceException {
1071     final ServerName sn = (request.hasDestinationServer() ?
1072       ProtobufUtil.toServerName(request.getDestinationServer()) : null);
1073 
1074     try {
1075       checkOpen();
1076       if (request.hasServerStartCode()) {
1077         // check that we are the same server that this RPC is intended for.
1078         long serverStartCode = request.getServerStartCode();
1079         if (regionServer.serverName.getStartcode() !=  serverStartCode) {
1080           throw new ServiceException(new DoNotRetryIOException("This RPC was intended for a " +
1081               "different server with startCode: " + serverStartCode + ", this server is: "
1082               + regionServer.serverName));
1083         }
1084       }
1085       final String encodedRegionName = ProtobufUtil.getRegionEncodedName(request.getRegion());
1086 
1087       // Can be null if we're calling close on a region that's not online
1088       final Region region = regionServer.getFromOnlineRegions(encodedRegionName);
1089       if ((region  != null) && (region .getCoprocessorHost() != null)) {
1090         region.getCoprocessorHost().preClose(false);
1091       }
1092 
1093       requestCount.increment();
1094       LOG.info("Close " + encodedRegionName + ", moving to " + sn);
1095       CloseRegionCoordination.CloseRegionDetails crd = regionServer.getCoordinatedStateManager()
1096         .getCloseRegionCoordination().parseFromProtoRequest(request);
1097 
1098       boolean closed = regionServer.closeRegion(encodedRegionName, false, crd, sn);
1099       CloseRegionResponse.Builder builder = CloseRegionResponse.newBuilder().setClosed(closed);
1100       return builder.build();
1101     } catch (IOException ie) {
1102       throw new ServiceException(ie);
1103     }
1104   }
1105 
1106   /**
1107    * Compact a region on the region server.
1108    *
1109    * @param controller the RPC controller
1110    * @param request the request
1111    * @throws ServiceException
1112    */
1113   @Override
1114   @QosPriority(priority=HConstants.ADMIN_QOS)
1115   public CompactRegionResponse compactRegion(final RpcController controller,
1116       final CompactRegionRequest request) throws ServiceException {
1117     try {
1118       checkOpen();
1119       requestCount.increment();
1120       Region region = getRegion(request.getRegion());
1121       region.startRegionOperation(Operation.COMPACT_REGION);
1122       LOG.info("Compacting " + region.getRegionInfo().getRegionNameAsString());
1123       boolean major = false;
1124       byte [] family = null;
1125       Store store = null;
1126       if (request.hasFamily()) {
1127         family = request.getFamily().toByteArray();
1128         store = region.getStore(family);
1129         if (store == null) {
1130           throw new ServiceException(new IOException("column family " + Bytes.toString(family)
1131             + " does not exist in region " + region.getRegionInfo().getRegionNameAsString()));
1132         }
1133       }
1134       if (request.hasMajor()) {
1135         major = request.getMajor();
1136       }
1137       if (major) {
1138         if (family != null) {
1139           store.triggerMajorCompaction();
1140         } else {
1141           region.triggerMajorCompaction();
1142         }
1143       }
1144 
1145       String familyLogMsg = (family != null)?" for column family: " + Bytes.toString(family):"";
1146       if (LOG.isTraceEnabled()) {
1147         LOG.trace("User-triggered compaction requested for region "
1148           + region.getRegionInfo().getRegionNameAsString() + familyLogMsg);
1149       }
1150       String log = "User-triggered " + (major ? "major " : "") + "compaction" + familyLogMsg;
1151       if(family != null) {
1152         regionServer.compactSplitThread.requestCompaction(region, store, log,
1153           Store.PRIORITY_USER, null, RpcServer.getRequestUser());
1154       } else {
1155         regionServer.compactSplitThread.requestCompaction(region, log,
1156           Store.PRIORITY_USER, null, RpcServer.getRequestUser());
1157       }
1158       return CompactRegionResponse.newBuilder().build();
1159     } catch (IOException ie) {
1160       throw new ServiceException(ie);
1161     }
1162   }
1163 
1164   /**
1165    * Flush a region on the region server.
1166    *
1167    * @param controller the RPC controller
1168    * @param request the request
1169    * @throws ServiceException
1170    */
1171   @Override
1172   @QosPriority(priority=HConstants.ADMIN_QOS)
1173   public FlushRegionResponse flushRegion(final RpcController controller,
1174       final FlushRegionRequest request) throws ServiceException {
1175     try {
1176       checkOpen();
1177       requestCount.increment();
1178       Region region = getRegion(request.getRegion());
1179       LOG.info("Flushing " + region.getRegionInfo().getRegionNameAsString());
1180       boolean shouldFlush = true;
1181       if (request.hasIfOlderThanTs()) {
1182         shouldFlush = region.getEarliestFlushTimeForAllStores() < request.getIfOlderThanTs();
1183       }
1184       FlushRegionResponse.Builder builder = FlushRegionResponse.newBuilder();
1185       if (shouldFlush) {
1186         boolean writeFlushWalMarker =  request.hasWriteFlushWalMarker() ?
1187             request.getWriteFlushWalMarker() : false;
1188         long startTime = EnvironmentEdgeManager.currentTime();
1189         // Go behind the curtain so we can manage writing of the flush WAL marker
1190         HRegion.FlushResultImpl flushResult = (HRegion.FlushResultImpl)
1191             ((HRegion)region).flushcache(true, writeFlushWalMarker);
1192         if (flushResult.isFlushSucceeded()) {
1193           long endTime = EnvironmentEdgeManager.currentTime();
1194           regionServer.metricsRegionServer.updateFlushTime(endTime - startTime);
1195         }
1196         boolean compactionNeeded = flushResult.isCompactionNeeded();
1197         if (compactionNeeded) {
1198           regionServer.compactSplitThread.requestSystemCompaction(region,
1199             "Compaction through user triggered flush");
1200         }
1201         builder.setFlushed(flushResult.isFlushSucceeded());
1202         builder.setWroteFlushWalMarker(flushResult.wroteFlushWalMarker);
1203       }
1204       builder.setLastFlushTime(region.getEarliestFlushTimeForAllStores());
1205       return builder.build();
1206     } catch (DroppedSnapshotException ex) {
1207       // Cache flush can fail in a few places. If it fails in a critical
1208       // section, we get a DroppedSnapshotException and a replay of wal
1209       // is required. Currently the only way to do this is a restart of
1210       // the server.
1211       regionServer.abort("Replay of WAL required. Forcing server shutdown", ex);
1212       throw new ServiceException(ex);
1213     } catch (IOException ie) {
1214       throw new ServiceException(ie);
1215     }
1216   }
1217 
1218   @Override
1219   @QosPriority(priority=HConstants.ADMIN_QOS)
1220   public GetOnlineRegionResponse getOnlineRegion(final RpcController controller,
1221       final GetOnlineRegionRequest request) throws ServiceException {
1222     try {
1223       checkOpen();
1224       requestCount.increment();
1225       Map<String, Region> onlineRegions = regionServer.onlineRegions;
1226       List<HRegionInfo> list = new ArrayList<HRegionInfo>(onlineRegions.size());
1227       for (Region region: onlineRegions.values()) {
1228         list.add(region.getRegionInfo());
1229       }
1230       Collections.sort(list);
1231       return ResponseConverter.buildGetOnlineRegionResponse(list);
1232     } catch (IOException ie) {
1233       throw new ServiceException(ie);
1234     }
1235   }
1236 
1237   @Override
1238   @QosPriority(priority=HConstants.ADMIN_QOS)
1239   public GetRegionInfoResponse getRegionInfo(final RpcController controller,
1240       final GetRegionInfoRequest request) throws ServiceException {
1241     try {
1242       checkOpen();
1243       requestCount.increment();
1244       Region region = getRegion(request.getRegion());
1245       HRegionInfo info = region.getRegionInfo();
1246       GetRegionInfoResponse.Builder builder = GetRegionInfoResponse.newBuilder();
1247       builder.setRegionInfo(HRegionInfo.convert(info));
1248       if (request.hasCompactionState() && request.getCompactionState()) {
1249         builder.setCompactionState(region.getCompactionState());
1250       }
1251       builder.setIsRecovering(region.isRecovering());
1252       return builder.build();
1253     } catch (IOException ie) {
1254       throw new ServiceException(ie);
1255     }
1256   }
1257 
1258   /**
1259    * Get some information of the region server.
1260    *
1261    * @param controller the RPC controller
1262    * @param request the request
1263    * @throws ServiceException
1264    */
1265   @Override
1266   @QosPriority(priority=HConstants.ADMIN_QOS)
1267   public GetServerInfoResponse getServerInfo(final RpcController controller,
1268       final GetServerInfoRequest request) throws ServiceException {
1269     try {
1270       checkOpen();
1271     } catch (IOException ie) {
1272       throw new ServiceException(ie);
1273     }
1274     requestCount.increment();
1275     int infoPort = regionServer.infoServer != null ? regionServer.infoServer.getPort() : -1;
1276     return ResponseConverter.buildGetServerInfoResponse(regionServer.serverName, infoPort);
1277   }
1278 
1279   @Override
1280   @QosPriority(priority=HConstants.ADMIN_QOS)
1281   public GetStoreFileResponse getStoreFile(final RpcController controller,
1282       final GetStoreFileRequest request) throws ServiceException {
1283     try {
1284       checkOpen();
1285       Region region = getRegion(request.getRegion());
1286       requestCount.increment();
1287       Set<byte[]> columnFamilies;
1288       if (request.getFamilyCount() == 0) {
1289         columnFamilies = region.getTableDesc().getFamiliesKeys();
1290       } else {
1291         columnFamilies = new TreeSet<byte[]>(Bytes.BYTES_RAWCOMPARATOR);
1292         for (ByteString cf: request.getFamilyList()) {
1293           columnFamilies.add(cf.toByteArray());
1294         }
1295       }
1296       int nCF = columnFamilies.size();
1297       List<String>  fileList = region.getStoreFileList(
1298         columnFamilies.toArray(new byte[nCF][]));
1299       GetStoreFileResponse.Builder builder = GetStoreFileResponse.newBuilder();
1300       builder.addAllStoreFile(fileList);
1301       return builder.build();
1302     } catch (IOException ie) {
1303       throw new ServiceException(ie);
1304     }
1305   }
1306 
1307   /**
1308    * Merge regions on the region server.
1309    *
1310    * @param controller the RPC controller
1311    * @param request the request
1312    * @return merge regions response
1313    * @throws ServiceException
1314    */
1315   @Override
1316   @QosPriority(priority = HConstants.ADMIN_QOS)
1317   public MergeRegionsResponse mergeRegions(final RpcController controller,
1318       final MergeRegionsRequest request) throws ServiceException {
1319     try {
1320       checkOpen();
1321       requestCount.increment();
1322       Region regionA = getRegion(request.getRegionA());
1323       Region regionB = getRegion(request.getRegionB());
1324       boolean forcible = request.getForcible();
1325       long masterSystemTime = request.hasMasterSystemTime() ? request.getMasterSystemTime() : -1;
1326       regionA.startRegionOperation(Operation.MERGE_REGION);
1327       regionB.startRegionOperation(Operation.MERGE_REGION);
1328       if (regionA.getRegionInfo().getReplicaId() != HRegionInfo.DEFAULT_REPLICA_ID ||
1329           regionB.getRegionInfo().getReplicaId() != HRegionInfo.DEFAULT_REPLICA_ID) {
1330         throw new ServiceException(new MergeRegionException("Can't merge non-default replicas"));
1331       }
1332       LOG.info("Receiving merging request for  " + regionA + ", " + regionB
1333           + ",forcible=" + forcible);
1334       long startTime = EnvironmentEdgeManager.currentTime();
1335       FlushResult flushResult = regionA.flush(true);
1336       if (flushResult.isFlushSucceeded()) {
1337         long endTime = EnvironmentEdgeManager.currentTime();
1338         regionServer.metricsRegionServer.updateFlushTime(endTime - startTime);
1339       }
1340       startTime = EnvironmentEdgeManager.currentTime();
1341       flushResult = regionB.flush(true);
1342       if (flushResult.isFlushSucceeded()) {
1343         long endTime = EnvironmentEdgeManager.currentTime();
1344         regionServer.metricsRegionServer.updateFlushTime(endTime - startTime);
1345       }
1346       regionServer.compactSplitThread.requestRegionsMerge(regionA, regionB, forcible,
1347           masterSystemTime, RpcServer.getRequestUser());
1348       return MergeRegionsResponse.newBuilder().build();
1349     } catch (DroppedSnapshotException ex) {
1350       regionServer.abort("Replay of WAL required. Forcing server shutdown", ex);
1351       throw new ServiceException(ex);
1352     } catch (IOException ie) {
1353       throw new ServiceException(ie);
1354     }
1355   }
1356 
1357   /**
1358    * Open asynchronously a region or a set of regions on the region server.
1359    *
1360    * The opening is coordinated by ZooKeeper, and this method requires the znode to be created
1361    *  before being called. As a consequence, this method should be called only from the master.
1362    * <p>
1363    * Different manages states for the region are:<ul>
1364    *  <li>region not opened: the region opening will start asynchronously.</li>
1365    *  <li>a close is already in progress: this is considered as an error.</li>
1366    *  <li>an open is already in progress: this new open request will be ignored. This is important
1367    *  because the Master can do multiple requests if it crashes.</li>
1368    *  <li>the region is already opened:  this new open request will be ignored./li>
1369    *  </ul>
1370    * </p>
1371    * <p>
1372    * Bulk assign: If there are more than 1 region to open, it will be considered as a bulk assign.
1373    * For a single region opening, errors are sent through a ServiceException. For bulk assign,
1374    * errors are put in the response as FAILED_OPENING.
1375    * </p>
1376    * @param controller the RPC controller
1377    * @param request the request
1378    * @throws ServiceException
1379    */
1380   @Override
1381   @QosPriority(priority=HConstants.ADMIN_QOS)
1382   public OpenRegionResponse openRegion(final RpcController controller,
1383       final OpenRegionRequest request) throws ServiceException {
1384     requestCount.increment();
1385     if (request.hasServerStartCode()) {
1386       // check that we are the same server that this RPC is intended for.
1387       long serverStartCode = request.getServerStartCode();
1388       if (regionServer.serverName.getStartcode() !=  serverStartCode) {
1389         throw new ServiceException(new DoNotRetryIOException("This RPC was intended for a " +
1390             "different server with startCode: " + serverStartCode + ", this server is: "
1391             + regionServer.serverName));
1392       }
1393     }
1394 
1395     OpenRegionResponse.Builder builder = OpenRegionResponse.newBuilder();
1396     final int regionCount = request.getOpenInfoCount();
1397     final Map<TableName, HTableDescriptor> htds =
1398         new HashMap<TableName, HTableDescriptor>(regionCount);
1399     final boolean isBulkAssign = regionCount > 1;
1400     try {
1401       checkOpen();
1402     } catch (IOException ie) {
1403       TableName tableName = null;
1404       if (regionCount == 1) {
1405         RegionInfo ri = request.getOpenInfo(0).getRegion();
1406         if (ri != null) {
1407           tableName = ProtobufUtil.toTableName(ri.getTableName());
1408         }
1409       }
1410       if (!TableName.META_TABLE_NAME.equals(tableName)) {
1411         throw new ServiceException(ie);
1412       }
1413       // We are assigning meta, wait a little for regionserver to finish initialization.
1414       int timeout = regionServer.conf.getInt(HConstants.HBASE_RPC_TIMEOUT_KEY,
1415         HConstants.DEFAULT_HBASE_RPC_TIMEOUT) >> 2; // Quarter of RPC timeout
1416       long endTime = System.currentTimeMillis() + timeout;
1417       synchronized (regionServer.online) {
1418         try {
1419           while (System.currentTimeMillis() <= endTime
1420               && !regionServer.isStopped() && !regionServer.isOnline()) {
1421             regionServer.online.wait(regionServer.msgInterval);
1422           }
1423           checkOpen();
1424         } catch (InterruptedException t) {
1425           Thread.currentThread().interrupt();
1426           throw new ServiceException(t);
1427         } catch (IOException e) {
1428           throw new ServiceException(e);
1429         }
1430       }
1431     }
1432 
1433     long masterSystemTime = request.hasMasterSystemTime() ? request.getMasterSystemTime() : -1;
1434 
1435     for (RegionOpenInfo regionOpenInfo : request.getOpenInfoList()) {
1436       final HRegionInfo region = HRegionInfo.convert(regionOpenInfo.getRegion());
1437       OpenRegionCoordination coordination = regionServer.getCoordinatedStateManager().
1438         getOpenRegionCoordination();
1439       OpenRegionCoordination.OpenRegionDetails ord =
1440         coordination.parseFromProtoRequest(regionOpenInfo);
1441 
1442       HTableDescriptor htd;
1443       try {
1444         final Region onlineRegion = regionServer.getFromOnlineRegions(region.getEncodedName());
1445         if (onlineRegion != null) {
1446           //Check if the region can actually be opened.
1447           if (onlineRegion.getCoprocessorHost() != null) {
1448             onlineRegion.getCoprocessorHost().preOpen();
1449           }
1450           // See HBASE-5094. Cross check with hbase:meta if still this RS is owning
1451           // the region.
1452           Pair<HRegionInfo, ServerName> p = MetaTableAccessor.getRegion(
1453             regionServer.getConnection(), region.getRegionName());
1454           if (regionServer.serverName.equals(p.getSecond())) {
1455             Boolean closing = regionServer.regionsInTransitionInRS.get(region.getEncodedNameAsBytes());
1456             // Map regionsInTransitionInRSOnly has an entry for a region only if the region
1457             // is in transition on this RS, so here closing can be null. If not null, it can
1458             // be true or false. True means the region is opening on this RS; while false
1459             // means the region is closing. Only return ALREADY_OPENED if not closing (i.e.
1460             // not in transition any more, or still transition to open.
1461             if (!Boolean.FALSE.equals(closing)
1462                 && regionServer.getFromOnlineRegions(region.getEncodedName()) != null) {
1463               LOG.warn("Attempted open of " + region.getEncodedName()
1464                 + " but already online on this server");
1465               builder.addOpeningState(RegionOpeningState.ALREADY_OPENED);
1466               continue;
1467             }
1468           } else {
1469             LOG.warn("The region " + region.getEncodedName() + " is online on this server"
1470               + " but hbase:meta does not have this server - continue opening.");
1471             regionServer.removeFromOnlineRegions(onlineRegion, null);
1472           }
1473         }
1474         LOG.info("Open " + region.getRegionNameAsString());
1475         htd = htds.get(region.getTable());
1476         if (htd == null) {
1477           htd = regionServer.tableDescriptors.get(region.getTable());
1478           htds.put(region.getTable(), htd);
1479         }
1480 
1481         final Boolean previous = regionServer.regionsInTransitionInRS.putIfAbsent(
1482           region.getEncodedNameAsBytes(), Boolean.TRUE);
1483 
1484         if (Boolean.FALSE.equals(previous)) {
1485           // There is a close in progress. We need to mark this open as failed in ZK.
1486 
1487           coordination.tryTransitionFromOfflineToFailedOpen(regionServer, region, ord);
1488 
1489           throw new RegionAlreadyInTransitionException("Received OPEN for the region:"
1490             + region.getRegionNameAsString() + " , which we are already trying to CLOSE ");
1491         }
1492 
1493         if (Boolean.TRUE.equals(previous)) {
1494           // An open is in progress. This is supported, but let's log this.
1495           LOG.info("Receiving OPEN for the region:" +
1496             region.getRegionNameAsString() + " , which we are already trying to OPEN"
1497               + " - ignoring this new request for this region.");
1498         }
1499 
1500         // We are opening this region. If it moves back and forth for whatever reason, we don't
1501         // want to keep returning the stale moved record while we are opening/if we close again.
1502         regionServer.removeFromMovedRegions(region.getEncodedName());
1503 
1504         if (previous == null) {
1505           // check if the region to be opened is marked in recovering state in ZK
1506           if (ZKSplitLog.isRegionMarkedRecoveringInZK(regionServer.getZooKeeper(),
1507               region.getEncodedName())) {
1508             // Check if current region open is for distributedLogReplay. This check is to support
1509             // rolling restart/upgrade where we want to Master/RS see same configuration
1510             if (!regionOpenInfo.hasOpenForDistributedLogReplay()
1511                   || regionOpenInfo.getOpenForDistributedLogReplay()) {
1512               regionServer.recoveringRegions.put(region.getEncodedName(), null);
1513             } else {
1514               // Remove stale recovery region from ZK when we open region not for recovering which
1515               // could happen when turn distributedLogReplay off from on.
1516               List<String> tmpRegions = new ArrayList<String>();
1517               tmpRegions.add(region.getEncodedName());
1518               ZKSplitLog.deleteRecoveringRegionZNodes(regionServer.getZooKeeper(),
1519                 tmpRegions);
1520             }
1521           }
1522           // If there is no action in progress, we can submit a specific handler.
1523           // Need to pass the expected version in the constructor.
1524           if (region.isMetaRegion()) {
1525             regionServer.service.submit(new OpenMetaHandler(
1526               regionServer, regionServer, region, htd, masterSystemTime, coordination, ord));
1527           } else {
1528             regionServer.updateRegionFavoredNodesMapping(region.getEncodedName(),
1529               regionOpenInfo.getFavoredNodesList());
1530             regionServer.service.submit(new OpenRegionHandler(
1531               regionServer, regionServer, region, htd, masterSystemTime, coordination, ord));
1532           }
1533         }
1534 
1535         builder.addOpeningState(RegionOpeningState.OPENED);
1536 
1537       } catch (KeeperException zooKeeperEx) {
1538         LOG.error("Can't retrieve recovering state from zookeeper", zooKeeperEx);
1539         throw new ServiceException(zooKeeperEx);
1540       } catch (IOException ie) {
1541         LOG.warn("Failed opening region " + region.getRegionNameAsString(), ie);
1542         if (isBulkAssign) {
1543           builder.addOpeningState(RegionOpeningState.FAILED_OPENING);
1544         } else {
1545           throw new ServiceException(ie);
1546         }
1547       }
1548     }
1549     return builder.build();
1550   }
1551 
1552   /**
1553    *  Wamrmup a region on this server.
1554    *
1555    * This method should only be called by Master. It synchrnously opens the region and
1556    * closes the region bringing the most important pages in cache.
1557    * <p>
1558    *
1559    * @param controller the RPC controller
1560    * @param request the request
1561    * @throws ServiceException
1562    */
1563   @Override
1564   public WarmupRegionResponse warmupRegion(final RpcController controller,
1565       final WarmupRegionRequest request) throws ServiceException {
1566 
1567     RegionInfo regionInfo = request.getRegionInfo();
1568     final HRegionInfo region = HRegionInfo.convert(regionInfo);
1569     HTableDescriptor htd;
1570     WarmupRegionResponse response = WarmupRegionResponse.getDefaultInstance();
1571 
1572     try {
1573       checkOpen();
1574       String encodedName = region.getEncodedName();
1575       byte[] encodedNameBytes = region.getEncodedNameAsBytes();
1576       final Region onlineRegion = regionServer.getFromOnlineRegions(encodedName);
1577 
1578       if (onlineRegion != null) {
1579         LOG.info("Region already online. Skipping warming up " + region);
1580         return response;
1581       }
1582 
1583       if (LOG.isDebugEnabled()) {
1584         LOG.debug("Warming up Region " + region.getRegionNameAsString());
1585       }
1586 
1587       htd = regionServer.tableDescriptors.get(region.getTable());
1588 
1589       if (regionServer.getRegionsInTransitionInRS().containsKey(encodedNameBytes)) {
1590         LOG.info("Region is in transition. Skipping warmup " + region);
1591         return response;
1592       }
1593 
1594       HRegion.warmupHRegion(region, htd, regionServer.getWAL(region),
1595           regionServer.getConfiguration(), regionServer, null);
1596 
1597     } catch (IOException ie) {
1598       LOG.error("Failed warming up region " + region.getRegionNameAsString(), ie);
1599       throw new ServiceException(ie);
1600     }
1601 
1602     return response;
1603   }
1604 
1605   /**
1606    * Replay the given changes when distributedLogReplay WAL edits from a failed RS. The guarantee is
1607    * that the given mutations will be durable on the receiving RS if this method returns without any
1608    * exception.
1609    * @param controller the RPC controller
1610    * @param request the request
1611    * @throws ServiceException
1612    */
1613   @Override
1614   @QosPriority(priority = HConstants.REPLAY_QOS)
1615   public ReplicateWALEntryResponse replay(final RpcController controller,
1616       final ReplicateWALEntryRequest request) throws ServiceException {
1617     long before = EnvironmentEdgeManager.currentTime();
1618     CellScanner cells = ((PayloadCarryingRpcController) controller).cellScanner();
1619     try {
1620       checkOpen();
1621       List<WALEntry> entries = request.getEntryList();
1622       if (entries == null || entries.isEmpty()) {
1623         // empty input
1624         return ReplicateWALEntryResponse.newBuilder().build();
1625       }
1626       ByteString regionName = entries.get(0).getKey().getEncodedRegionName();
1627       Region region = regionServer.getRegionByEncodedName(regionName.toStringUtf8());
1628       RegionCoprocessorHost coprocessorHost =
1629           ServerRegionReplicaUtil.isDefaultReplica(region.getRegionInfo())
1630             ? region.getCoprocessorHost()
1631             : null; // do not invoke coprocessors if this is a secondary region replica
1632       List<Pair<WALKey, WALEdit>> walEntries = new ArrayList<Pair<WALKey, WALEdit>>();
1633 
1634       // Skip adding the edits to WAL if this is a secondary region replica
1635       boolean isPrimary = RegionReplicaUtil.isDefaultReplica(region.getRegionInfo());
1636       Durability durability = isPrimary ? Durability.USE_DEFAULT : Durability.SKIP_WAL;
1637 
1638       for (WALEntry entry : entries) {
1639         if (!regionName.equals(entry.getKey().getEncodedRegionName())) {
1640           throw new NotServingRegionException("Replay request contains entries from multiple " +
1641               "regions. First region:" + regionName.toStringUtf8() + " , other region:"
1642               + entry.getKey().getEncodedRegionName());
1643         }
1644         if (regionServer.nonceManager != null && isPrimary) {
1645           long nonceGroup = entry.getKey().hasNonceGroup()
1646             ? entry.getKey().getNonceGroup() : HConstants.NO_NONCE;
1647           long nonce = entry.getKey().hasNonce() ? entry.getKey().getNonce() : HConstants.NO_NONCE;
1648           regionServer.nonceManager.reportOperationFromWal(nonceGroup, nonce, entry.getKey().getWriteTime());
1649         }
1650         Pair<WALKey, WALEdit> walEntry = (coprocessorHost == null) ? null :
1651           new Pair<WALKey, WALEdit>();
1652         List<WALSplitter.MutationReplay> edits = WALSplitter.getMutationsFromWALEntry(entry,
1653           cells, walEntry, durability);
1654         if (coprocessorHost != null) {
1655           // Start coprocessor replay here. The coprocessor is for each WALEdit instead of a
1656           // KeyValue.
1657           if (coprocessorHost.preWALRestore(region.getRegionInfo(), walEntry.getFirst(),
1658             walEntry.getSecond())) {
1659             // if bypass this log entry, ignore it ...
1660             continue;
1661           }
1662           walEntries.add(walEntry);
1663         }
1664         if(edits!=null && !edits.isEmpty()) {
1665           long replaySeqId = (entry.getKey().hasOrigSequenceNumber()) ?
1666             entry.getKey().getOrigSequenceNumber() : entry.getKey().getLogSequenceNumber();
1667           OperationStatus[] result = doReplayBatchOp(region, edits, replaySeqId);
1668           // check if it's a partial success
1669           for (int i = 0; result != null && i < result.length; i++) {
1670             if (result[i] != OperationStatus.SUCCESS) {
1671               throw new IOException(result[i].getExceptionMsg());
1672             }
1673           }
1674         }
1675       }
1676 
1677       //sync wal at the end because ASYNC_WAL is used above
1678       WAL wal = getWAL(region);
1679       if (wal != null) {
1680         wal.sync();
1681       }
1682 
1683       if (coprocessorHost != null) {
1684         for (Pair<WALKey, WALEdit> entry : walEntries) {
1685           coprocessorHost.postWALRestore(region.getRegionInfo(), entry.getFirst(),
1686             entry.getSecond());
1687         }
1688       }
1689       return ReplicateWALEntryResponse.newBuilder().build();
1690     } catch (IOException ie) {
1691       throw new ServiceException(ie);
1692     } finally {
1693       if (regionServer.metricsRegionServer != null) {
1694         regionServer.metricsRegionServer.updateReplay(
1695           EnvironmentEdgeManager.currentTime() - before);
1696       }
1697     }
1698   }
1699 
1700   WAL getWAL(Region region) {
1701     return ((HRegion)region).getWAL();
1702   }
1703 
1704   /**
1705    * Replicate WAL entries on the region server.
1706    *
1707    * @param controller the RPC controller
1708    * @param request the request
1709    * @throws ServiceException
1710    */
1711   @Override
1712   @QosPriority(priority=HConstants.REPLICATION_QOS)
1713   public ReplicateWALEntryResponse replicateWALEntry(final RpcController controller,
1714       final ReplicateWALEntryRequest request) throws ServiceException {
1715     try {
1716       checkOpen();
1717       if (regionServer.replicationSinkHandler != null) {
1718         requestCount.increment();
1719         List<WALEntry> entries = request.getEntryList();
1720         CellScanner cellScanner = ((PayloadCarryingRpcController)controller).cellScanner();
1721         regionServer.getRegionServerCoprocessorHost().preReplicateLogEntries(entries, cellScanner);
1722         regionServer.replicationSinkHandler.replicateLogEntries(entries, cellScanner);
1723         regionServer.getRegionServerCoprocessorHost().postReplicateLogEntries(entries, cellScanner);
1724         return ReplicateWALEntryResponse.newBuilder().build();
1725       } else {
1726         throw new ServiceException("Replication services are not initialized yet");
1727       }
1728     } catch (IOException ie) {
1729       throw new ServiceException(ie);
1730     }
1731   }
1732 
1733   /**
1734    * Roll the WAL writer of the region server.
1735    * @param controller the RPC controller
1736    * @param request the request
1737    * @throws ServiceException
1738    */
1739   @Override
1740   public RollWALWriterResponse rollWALWriter(final RpcController controller,
1741       final RollWALWriterRequest request) throws ServiceException {
1742     try {
1743       checkOpen();
1744       requestCount.increment();
1745       regionServer.getRegionServerCoprocessorHost().preRollWALWriterRequest();
1746       regionServer.walRoller.requestRollAll();
1747       regionServer.getRegionServerCoprocessorHost().postRollWALWriterRequest();
1748       RollWALWriterResponse.Builder builder = RollWALWriterResponse.newBuilder();
1749       return builder.build();
1750     } catch (IOException ie) {
1751       throw new ServiceException(ie);
1752     }
1753   }
1754 
1755   /**
1756    * Split a region on the region server.
1757    *
1758    * @param controller the RPC controller
1759    * @param request the request
1760    * @throws ServiceException
1761    */
1762   @Override
1763   @QosPriority(priority=HConstants.ADMIN_QOS)
1764   public SplitRegionResponse splitRegion(final RpcController controller,
1765       final SplitRegionRequest request) throws ServiceException {
1766     try {
1767       checkOpen();
1768       requestCount.increment();
1769       Region region = getRegion(request.getRegion());
1770       region.startRegionOperation(Operation.SPLIT_REGION);
1771       if (region.getRegionInfo().getReplicaId() != HRegionInfo.DEFAULT_REPLICA_ID) {
1772         throw new IOException("Can't split replicas directly. "
1773             + "Replicas are auto-split when their primary is split.");
1774       }
1775       LOG.info("Splitting " + region.getRegionInfo().getRegionNameAsString());
1776       long startTime = EnvironmentEdgeManager.currentTime();
1777       FlushResult flushResult = region.flush(true);
1778       if (flushResult.isFlushSucceeded()) {
1779         long endTime = EnvironmentEdgeManager.currentTime();
1780         regionServer.metricsRegionServer.updateFlushTime(endTime - startTime);
1781       }
1782       byte[] splitPoint = null;
1783       if (request.hasSplitPoint()) {
1784         splitPoint = request.getSplitPoint().toByteArray();
1785       }
1786       ((HRegion)region).forceSplit(splitPoint);
1787       regionServer.compactSplitThread.requestSplit(region, ((HRegion)region).checkSplit(),
1788         RpcServer.getRequestUser());
1789       return SplitRegionResponse.newBuilder().build();
1790     } catch (DroppedSnapshotException ex) {
1791       regionServer.abort("Replay of WAL required. Forcing server shutdown", ex);
1792       throw new ServiceException(ex);
1793     } catch (IOException ie) {
1794       throw new ServiceException(ie);
1795     }
1796   }
1797 
1798   /**
1799    * Stop the region server.
1800    *
1801    * @param controller the RPC controller
1802    * @param request the request
1803    * @throws ServiceException
1804    */
1805   @Override
1806   @QosPriority(priority=HConstants.ADMIN_QOS)
1807   public StopServerResponse stopServer(final RpcController controller,
1808       final StopServerRequest request) throws ServiceException {
1809     requestCount.increment();
1810     String reason = request.getReason();
1811     regionServer.stop(reason);
1812     return StopServerResponse.newBuilder().build();
1813   }
1814 
1815   @Override
1816   public UpdateFavoredNodesResponse updateFavoredNodes(RpcController controller,
1817       UpdateFavoredNodesRequest request) throws ServiceException {
1818     List<UpdateFavoredNodesRequest.RegionUpdateInfo> openInfoList = request.getUpdateInfoList();
1819     UpdateFavoredNodesResponse.Builder respBuilder = UpdateFavoredNodesResponse.newBuilder();
1820     for (UpdateFavoredNodesRequest.RegionUpdateInfo regionUpdateInfo : openInfoList) {
1821       HRegionInfo hri = HRegionInfo.convert(regionUpdateInfo.getRegion());
1822       regionServer.updateRegionFavoredNodesMapping(hri.getEncodedName(),
1823         regionUpdateInfo.getFavoredNodesList());
1824     }
1825     respBuilder.setResponse(openInfoList.size());
1826     return respBuilder.build();
1827   }
1828 
1829   /**
1830    * Atomically bulk load several HFiles into an open region
1831    * @return true if successful, false is failed but recoverably (no action)
1832    * @throws IOException if failed unrecoverably
1833    */
1834   @Override
1835   public BulkLoadHFileResponse bulkLoadHFile(final RpcController controller,
1836       final BulkLoadHFileRequest request) throws ServiceException {
1837     try {
1838       checkOpen();
1839       requestCount.increment();
1840       Region region = getRegion(request.getRegion());
1841       List<Pair<byte[], String>> familyPaths = new ArrayList<Pair<byte[], String>>();
1842       for (FamilyPath familyPath: request.getFamilyPathList()) {
1843         familyPaths.add(new Pair<byte[], String>(familyPath.getFamily().toByteArray(),
1844           familyPath.getPath()));
1845       }
1846       boolean bypass = false;
1847       if (region.getCoprocessorHost() != null) {
1848         bypass = region.getCoprocessorHost().preBulkLoadHFile(familyPaths);
1849       }
1850       boolean loaded = false;
1851       if (!bypass) {
1852         loaded = region.bulkLoadHFiles(familyPaths, request.getAssignSeqNum(), null);
1853       }
1854       if (region.getCoprocessorHost() != null) {
1855         loaded = region.getCoprocessorHost().postBulkLoadHFile(familyPaths, loaded);
1856       }
1857       BulkLoadHFileResponse.Builder builder = BulkLoadHFileResponse.newBuilder();
1858       builder.setLoaded(loaded);
1859       return builder.build();
1860     } catch (IOException ie) {
1861       throw new ServiceException(ie);
1862     }
1863   }
1864 
1865   @Override
1866   public CoprocessorServiceResponse execService(final RpcController controller,
1867       final CoprocessorServiceRequest request) throws ServiceException {
1868     try {
1869       checkOpen();
1870       requestCount.increment();
1871       Region region = getRegion(request.getRegion());
1872       Message result = execServiceOnRegion(region, request.getCall());
1873       CoprocessorServiceResponse.Builder builder =
1874         CoprocessorServiceResponse.newBuilder();
1875       builder.setRegion(RequestConverter.buildRegionSpecifier(
1876         RegionSpecifierType.REGION_NAME, region.getRegionInfo().getRegionName()));
1877       builder.setValue(
1878         builder.getValueBuilder().setName(result.getClass().getName())
1879           .setValue(result.toByteString()));
1880       return builder.build();
1881     } catch (IOException ie) {
1882       throw new ServiceException(ie);
1883     }
1884   }
1885 
1886   private Message execServiceOnRegion(Region region,
1887       final ClientProtos.CoprocessorServiceCall serviceCall) throws IOException {
1888     // ignore the passed in controller (from the serialized call)
1889     ServerRpcController execController = new ServerRpcController();
1890     Message result = region.execService(execController, serviceCall);
1891     if (execController.getFailedOn() != null) {
1892       throw execController.getFailedOn();
1893     }
1894     return result;
1895   }
1896 
1897   /**
1898    * Get data from a table.
1899    *
1900    * @param controller the RPC controller
1901    * @param request the get request
1902    * @throws ServiceException
1903    */
1904   @Override
1905   public GetResponse get(final RpcController controller,
1906       final GetRequest request) throws ServiceException {
1907     long before = EnvironmentEdgeManager.currentTime();
1908     OperationQuota quota = null;
1909     try {
1910       checkOpen();
1911       requestCount.increment();
1912       Region region = getRegion(request.getRegion());
1913 
1914       GetResponse.Builder builder = GetResponse.newBuilder();
1915       ClientProtos.Get get = request.getGet();
1916       Boolean existence = null;
1917       Result r = null;
1918       quota = getQuotaManager().checkQuota(region, OperationQuota.OperationType.GET);
1919 
1920       if (get.hasClosestRowBefore() && get.getClosestRowBefore()) {
1921         if (get.getColumnCount() != 1) {
1922           throw new DoNotRetryIOException(
1923             "get ClosestRowBefore supports one and only one family now, not "
1924               + get.getColumnCount() + " families");
1925         }
1926         byte[] row = get.getRow().toByteArray();
1927         byte[] family = get.getColumn(0).getFamily().toByteArray();
1928         r = region.getClosestRowBefore(row, family);
1929       } else {
1930         Get clientGet = ProtobufUtil.toGet(get);
1931         if (get.getExistenceOnly() && region.getCoprocessorHost() != null) {
1932           existence = region.getCoprocessorHost().preExists(clientGet);
1933         }
1934         if (existence == null) {
1935           r = region.get(clientGet);
1936           if (get.getExistenceOnly()) {
1937             boolean exists = r.getExists();
1938             if (region.getCoprocessorHost() != null) {
1939               exists = region.getCoprocessorHost().postExists(clientGet, exists);
1940             }
1941             existence = exists;
1942           }
1943         }
1944       }
1945       if (existence != null){
1946         ClientProtos.Result pbr =
1947             ProtobufUtil.toResult(existence, region.getRegionInfo().getReplicaId() != 0);
1948         builder.setResult(pbr);
1949       } else  if (r != null) {
1950         ClientProtos.Result pbr = ProtobufUtil.toResult(r);
1951         builder.setResult(pbr);
1952       }
1953       if (r != null) {
1954         quota.addGetResult(r);
1955       }
1956       return builder.build();
1957     } catch (IOException ie) {
1958       throw new ServiceException(ie);
1959     } finally {
1960       if (regionServer.metricsRegionServer != null) {
1961         regionServer.metricsRegionServer.updateGet(
1962           EnvironmentEdgeManager.currentTime() - before);
1963       }
1964       if (quota != null) {
1965         quota.close();
1966       }
1967     }
1968   }
1969 
1970   /**
1971    * Execute multiple actions on a table: get, mutate, and/or execCoprocessor
1972    *
1973    * @param rpcc the RPC controller
1974    * @param request the multi request
1975    * @throws ServiceException
1976    */
1977   @Override
1978   public MultiResponse multi(final RpcController rpcc, final MultiRequest request)
1979   throws ServiceException {
1980     try {
1981       checkOpen();
1982     } catch (IOException ie) {
1983       throw new ServiceException(ie);
1984     }
1985 
1986     // rpc controller is how we bring in data via the back door;  it is unprotobuf'ed data.
1987     // It is also the conduit via which we pass back data.
1988     PayloadCarryingRpcController controller = (PayloadCarryingRpcController)rpcc;
1989     CellScanner cellScanner = controller != null ? controller.cellScanner(): null;
1990     if (controller != null) controller.setCellScanner(null);
1991 
1992     long nonceGroup = request.hasNonceGroup() ? request.getNonceGroup() : HConstants.NO_NONCE;
1993 
1994     // this will contain all the cells that we need to return. It's created later, if needed.
1995     List<CellScannable> cellsToReturn = null;
1996     MultiResponse.Builder responseBuilder = MultiResponse.newBuilder();
1997     RegionActionResult.Builder regionActionResultBuilder = RegionActionResult.newBuilder();
1998     Boolean processed = null;
1999 
2000     for (RegionAction regionAction : request.getRegionActionList()) {
2001       this.requestCount.add(regionAction.getActionCount());
2002       OperationQuota quota;
2003       Region region;
2004       regionActionResultBuilder.clear();
2005       try {
2006         region = getRegion(regionAction.getRegion());
2007         quota = getQuotaManager().checkQuota(region, regionAction.getActionList());
2008       } catch (IOException e) {
2009         rpcServer.getMetrics().exception(e);
2010         regionActionResultBuilder.setException(ResponseConverter.buildException(e));
2011         responseBuilder.addRegionActionResult(regionActionResultBuilder.build());
2012         continue;  // For this region it's a failure.
2013       }
2014 
2015       if (regionAction.hasAtomic() && regionAction.getAtomic()) {
2016         // How does this call happen?  It may need some work to play well w/ the surroundings.
2017         // Need to return an item per Action along w/ Action index.  TODO.
2018         try {
2019           if (request.hasCondition()) {
2020             Condition condition = request.getCondition();
2021             byte[] row = condition.getRow().toByteArray();
2022             byte[] family = condition.getFamily().toByteArray();
2023             byte[] qualifier = condition.getQualifier().toByteArray();
2024             CompareOp compareOp = CompareOp.valueOf(condition.getCompareType().name());
2025             ByteArrayComparable comparator =
2026                 ProtobufUtil.toComparator(condition.getComparator());
2027             processed = checkAndRowMutate(region, regionAction.getActionList(),
2028                   cellScanner, row, family, qualifier, compareOp, comparator);
2029           } else {
2030             ClientProtos.RegionLoadStats stats = mutateRows(region, regionAction.getActionList(),
2031                 cellScanner);
2032             // add the stats to the request
2033             if(stats != null) {
2034               responseBuilder.addRegionActionResult(RegionActionResult.newBuilder()
2035                   .addResultOrException(ResultOrException.newBuilder().setLoadStats(stats)));
2036             }
2037             processed = Boolean.TRUE;
2038           }
2039         } catch (IOException e) {
2040           rpcServer.getMetrics().exception(e);
2041           // As it's atomic, we may expect it's a global failure.
2042           regionActionResultBuilder.setException(ResponseConverter.buildException(e));
2043         }
2044       } else {
2045         // doNonAtomicRegionMutation manages the exception internally
2046         cellsToReturn = doNonAtomicRegionMutation(region, quota, regionAction, cellScanner,
2047             regionActionResultBuilder, cellsToReturn, nonceGroup);
2048       }
2049       responseBuilder.addRegionActionResult(regionActionResultBuilder.build());
2050       quota.close();
2051     }
2052     // Load the controller with the Cells to return.
2053     if (cellsToReturn != null && !cellsToReturn.isEmpty() && controller != null) {
2054       controller.setCellScanner(CellUtil.createCellScanner(cellsToReturn));
2055     }
2056     if (processed != null) responseBuilder.setProcessed(processed);
2057     return responseBuilder.build();
2058   }
2059 
2060   /**
2061    * Mutate data in a table.
2062    *
2063    * @param rpcc the RPC controller
2064    * @param request the mutate request
2065    * @throws ServiceException
2066    */
2067   @Override
2068   public MutateResponse mutate(final RpcController rpcc,
2069       final MutateRequest request) throws ServiceException {
2070     // rpc controller is how we bring in data via the back door;  it is unprotobuf'ed data.
2071     // It is also the conduit via which we pass back data.
2072     PayloadCarryingRpcController controller = (PayloadCarryingRpcController)rpcc;
2073     CellScanner cellScanner = controller != null? controller.cellScanner(): null;
2074     OperationQuota quota = null;
2075     // Clear scanner so we are not holding on to reference across call.
2076     if (controller != null) controller.setCellScanner(null);
2077     try {
2078       checkOpen();
2079       requestCount.increment();
2080       Region region = getRegion(request.getRegion());
2081       MutateResponse.Builder builder = MutateResponse.newBuilder();
2082       MutationProto mutation = request.getMutation();
2083       if (!region.getRegionInfo().isMetaTable()) {
2084         regionServer.cacheFlusher.reclaimMemStoreMemory();
2085       }
2086       long nonceGroup = request.hasNonceGroup() ? request.getNonceGroup() : HConstants.NO_NONCE;
2087       Result r = null;
2088       Boolean processed = null;
2089       MutationType type = mutation.getMutateType();
2090       long mutationSize = 0;
2091       quota = getQuotaManager().checkQuota(region, OperationQuota.OperationType.MUTATE);
2092       switch (type) {
2093       case APPEND:
2094         // TODO: this doesn't actually check anything.
2095         r = append(region, quota, mutation, cellScanner, nonceGroup);
2096         break;
2097       case INCREMENT:
2098         // TODO: this doesn't actually check anything.
2099         r = increment(region, quota, mutation, cellScanner, nonceGroup);
2100         break;
2101       case PUT:
2102         Put put = ProtobufUtil.toPut(mutation, cellScanner);
2103         quota.addMutation(put);
2104         if (request.hasCondition()) {
2105           Condition condition = request.getCondition();
2106           byte[] row = condition.getRow().toByteArray();
2107           byte[] family = condition.getFamily().toByteArray();
2108           byte[] qualifier = condition.getQualifier().toByteArray();
2109           CompareOp compareOp = CompareOp.valueOf(condition.getCompareType().name());
2110           ByteArrayComparable comparator =
2111             ProtobufUtil.toComparator(condition.getComparator());
2112           if (region.getCoprocessorHost() != null) {
2113             processed = region.getCoprocessorHost().preCheckAndPut(
2114               row, family, qualifier, compareOp, comparator, put);
2115           }
2116           if (processed == null) {
2117             boolean result = region.checkAndMutate(row, family,
2118               qualifier, compareOp, comparator, put, true);
2119             if (region.getCoprocessorHost() != null) {
2120               result = region.getCoprocessorHost().postCheckAndPut(row, family,
2121                 qualifier, compareOp, comparator, put, result);
2122             }
2123             processed = result;
2124           }
2125         } else {
2126           region.put(put);
2127           processed = Boolean.TRUE;
2128         }
2129         break;
2130       case DELETE:
2131         Delete delete = ProtobufUtil.toDelete(mutation, cellScanner);
2132         quota.addMutation(delete);
2133         if (request.hasCondition()) {
2134           Condition condition = request.getCondition();
2135           byte[] row = condition.getRow().toByteArray();
2136           byte[] family = condition.getFamily().toByteArray();
2137           byte[] qualifier = condition.getQualifier().toByteArray();
2138           CompareOp compareOp = CompareOp.valueOf(condition.getCompareType().name());
2139           ByteArrayComparable comparator =
2140             ProtobufUtil.toComparator(condition.getComparator());
2141           if (region.getCoprocessorHost() != null) {
2142             processed = region.getCoprocessorHost().preCheckAndDelete(
2143               row, family, qualifier, compareOp, comparator, delete);
2144           }
2145           if (processed == null) {
2146             boolean result = region.checkAndMutate(row, family,
2147               qualifier, compareOp, comparator, delete, true);
2148             if (region.getCoprocessorHost() != null) {
2149               result = region.getCoprocessorHost().postCheckAndDelete(row, family,
2150                 qualifier, compareOp, comparator, delete, result);
2151             }
2152             processed = result;
2153           }
2154         } else {
2155           region.delete(delete);
2156           processed = Boolean.TRUE;
2157         }
2158         break;
2159       default:
2160           throw new DoNotRetryIOException(
2161             "Unsupported mutate type: " + type.name());
2162       }
2163       if (processed != null) builder.setProcessed(processed.booleanValue());
2164       addResult(builder, r, controller);
2165       return builder.build();
2166     } catch (IOException ie) {
2167       regionServer.checkFileSystem();
2168       throw new ServiceException(ie);
2169     } finally {
2170       if (quota != null) {
2171         quota.close();
2172       }
2173     }
2174   }
2175 
2176   /**
2177    * Scan data in a table.
2178    *
2179    * @param controller the RPC controller
2180    * @param request the scan request
2181    * @throws ServiceException
2182    */
2183   @Override
2184   public ScanResponse scan(final RpcController controller, final ScanRequest request)
2185   throws ServiceException {
2186     OperationQuota quota = null;
2187     Leases.Lease lease = null;
2188     String scannerName = null;
2189     try {
2190       if (!request.hasScannerId() && !request.hasScan()) {
2191         throw new DoNotRetryIOException(
2192           "Missing required input: scannerId or scan");
2193       }
2194       long scannerId = -1;
2195       if (request.hasScannerId()) {
2196         scannerId = request.getScannerId();
2197         scannerName = String.valueOf(scannerId);
2198       }
2199       try {
2200         checkOpen();
2201       } catch (IOException e) {
2202         // If checkOpen failed, server not running or filesystem gone,
2203         // cancel this lease; filesystem is gone or we're closing or something.
2204         if (scannerName != null) {
2205           LOG.debug("Server shutting down and client tried to access missing scanner "
2206             + scannerName);
2207           if (regionServer.leases != null) {
2208             try {
2209               regionServer.leases.cancelLease(scannerName);
2210             } catch (LeaseException le) {
2211               // No problem, ignore
2212             }
2213           }
2214         }
2215         throw e;
2216       }
2217       requestCount.increment();
2218 
2219       int ttl = 0;
2220       Region region = null;
2221       RegionScanner scanner = null;
2222       RegionScannerHolder rsh = null;
2223       boolean moreResults = true;
2224       boolean closeScanner = false;
2225       boolean isSmallScan = false;
2226       ScanResponse.Builder builder = ScanResponse.newBuilder();
2227       if (request.hasCloseScanner()) {
2228         closeScanner = request.getCloseScanner();
2229       }
2230       int rows = closeScanner ? 0 : 1;
2231       if (request.hasNumberOfRows()) {
2232         rows = request.getNumberOfRows();
2233       }
2234       if (request.hasScannerId()) {
2235         rsh = scanners.get(scannerName);
2236         if (rsh == null) {
2237           LOG.info("Client tried to access missing scanner " + scannerName);
2238           throw new UnknownScannerException(
2239             "Name: " + scannerName + ", already closed?");
2240         }
2241         scanner = rsh.s;
2242         HRegionInfo hri = scanner.getRegionInfo();
2243         region = regionServer.getRegion(hri.getRegionName());
2244         if (region != rsh.r) { // Yes, should be the same instance
2245           throw new NotServingRegionException("Region was re-opened after the scanner"
2246             + scannerName + " was created: " + hri.getRegionNameAsString());
2247         }
2248       } else {
2249         region = getRegion(request.getRegion());
2250         ClientProtos.Scan protoScan = request.getScan();
2251         boolean isLoadingCfsOnDemandSet = protoScan.hasLoadColumnFamiliesOnDemand();
2252         Scan scan = ProtobufUtil.toScan(protoScan);
2253         // if the request doesn't set this, get the default region setting.
2254         if (!isLoadingCfsOnDemandSet) {
2255           scan.setLoadColumnFamiliesOnDemand(region.isLoadingCfsOnDemandDefault());
2256         }
2257 
2258         isSmallScan = scan.isSmall();
2259         if (!scan.hasFamilies()) {
2260           // Adding all families to scanner
2261           for (byte[] family: region.getTableDesc().getFamiliesKeys()) {
2262             scan.addFamily(family);
2263           }
2264         }
2265 
2266         if (region.getCoprocessorHost() != null) {
2267           scanner = region.getCoprocessorHost().preScannerOpen(scan);
2268         }
2269         if (scanner == null) {
2270           scanner = region.getScanner(scan);
2271         }
2272         if (region.getCoprocessorHost() != null) {
2273           scanner = region.getCoprocessorHost().postScannerOpen(scan, scanner);
2274         }
2275         scannerId = addScanner(scanner, region);
2276         scannerName = String.valueOf(scannerId);
2277         ttl = this.scannerLeaseTimeoutPeriod;
2278       }
2279       if (request.hasRenew() && request.getRenew()) {
2280         rsh = scanners.get(scannerName);
2281         lease = regionServer.leases.removeLease(scannerName);
2282         if (lease != null && rsh != null) {
2283           regionServer.leases.addLease(lease);
2284           // Increment the nextCallSeq value which is the next expected from client.
2285           rsh.incNextCallSeq();
2286         }
2287         return builder.build();
2288       }
2289 
2290       quota = getQuotaManager().checkQuota(region, OperationQuota.OperationType.SCAN);
2291       long maxQuotaResultSize = Math.min(maxScannerResultSize, quota.getReadAvailable());
2292       if (rows > 0) {
2293         // if nextCallSeq does not match throw Exception straight away. This needs to be
2294         // performed even before checking of Lease.
2295         // See HBASE-5974
2296         if (request.hasNextCallSeq()) {
2297           if (rsh == null) {
2298             rsh = scanners.get(scannerName);
2299           }
2300           if (rsh != null) {
2301             if (request.getNextCallSeq() != rsh.getNextCallSeq()) {
2302               throw new OutOfOrderScannerNextException(
2303                 "Expected nextCallSeq: " + rsh.getNextCallSeq()
2304                 + " But the nextCallSeq got from client: " + request.getNextCallSeq() +
2305                 "; request=" + TextFormat.shortDebugString(request));
2306             }
2307             // Increment the nextCallSeq value which is the next expected from client.
2308             rsh.incNextCallSeq();
2309           }
2310         }
2311         try {
2312           // Remove lease while its being processed in server; protects against case
2313           // where processing of request takes > lease expiration time.
2314           lease = regionServer.leases.removeLease(scannerName);
2315           List<Result> results = new ArrayList<Result>();
2316           long totalCellSize = 0;
2317           long currentScanResultSize = 0;
2318 
2319           boolean done = false;
2320           // Call coprocessor. Get region info from scanner.
2321           if (region != null && region.getCoprocessorHost() != null) {
2322             Boolean bypass = region.getCoprocessorHost().preScannerNext(
2323               scanner, results, rows);
2324             if (!results.isEmpty()) {
2325               for (Result r : results) {
2326                 for (Cell cell : r.rawCells()) {
2327                   totalCellSize += CellUtil.estimatedSerializedSizeOf(cell);
2328                   currentScanResultSize += CellUtil.estimatedHeapSizeOfWithoutTags(cell);
2329                 }
2330               }
2331             }
2332             if (bypass != null && bypass.booleanValue()) {
2333               done = true;
2334             }
2335           }
2336 
2337           if (!done) {
2338             long maxResultSize = Math.min(scanner.getMaxResultSize(), maxQuotaResultSize);
2339             if (maxResultSize <= 0) {
2340               maxResultSize = maxQuotaResultSize;
2341             }
2342             // This is cells inside a row. Default size is 10 so if many versions or many cfs,
2343             // then we'll resize. Resizings show in profiler. Set it higher than 10. For now
2344             // arbitrary 32. TODO: keep record of general size of results being returned.
2345             List<Cell> values = new ArrayList<Cell>(32);
2346             region.startRegionOperation(Operation.SCAN);
2347             try {
2348               int i = 0;
2349               synchronized(scanner) {
2350                 boolean stale = (region.getRegionInfo().getReplicaId() != 0);
2351                 boolean clientHandlesPartials =
2352                     request.hasClientHandlesPartials() && request.getClientHandlesPartials();
2353                 boolean clientHandlesHeartbeats =
2354                     request.hasClientHandlesHeartbeats() && request.getClientHandlesHeartbeats();
2355 
2356                 // On the server side we must ensure that the correct ordering of partial results is
2357                 // returned to the client to allow them to properly reconstruct the partial results.
2358                 // If the coprocessor host is adding to the result list, we cannot guarantee the
2359                 // correct ordering of partial results and so we prevent partial results from being
2360                 // formed.
2361                 boolean serverGuaranteesOrderOfPartials = currentScanResultSize == 0;
2362                 boolean allowPartialResults =
2363                     clientHandlesPartials && serverGuaranteesOrderOfPartials && !isSmallScan;
2364                 boolean moreRows = false;
2365 
2366                 // Heartbeat messages occur when the processing of the ScanRequest is exceeds a
2367                 // certain time threshold on the server. When the time threshold is exceeded, the
2368                 // server stops the scan and sends back whatever Results it has accumulated within
2369                 // that time period (may be empty). Since heartbeat messages have the potential to
2370                 // create partial Results (in the event that the timeout occurs in the middle of a
2371                 // row), we must only generate heartbeat messages when the client can handle both
2372                 // heartbeats AND partials
2373                 boolean allowHeartbeatMessages = clientHandlesHeartbeats && allowPartialResults;
2374 
2375                 // Default value of timeLimit is negative to indicate no timeLimit should be
2376                 // enforced.
2377                 long timeLimit = -1;
2378 
2379                 // Set the time limit to be half of the more restrictive timeout value (one of the
2380                 // timeout values must be positive). In the event that both values are positive, the
2381                 // more restrictive of the two is used to calculate the limit.
2382                 if (allowHeartbeatMessages && (scannerLeaseTimeoutPeriod > 0 || rpcTimeout > 0)) {
2383                   long timeLimitDelta;
2384                   if (scannerLeaseTimeoutPeriod > 0 && rpcTimeout > 0) {
2385                     timeLimitDelta = Math.min(scannerLeaseTimeoutPeriod, rpcTimeout);
2386                   } else {
2387                     timeLimitDelta =
2388                         scannerLeaseTimeoutPeriod > 0 ? scannerLeaseTimeoutPeriod : rpcTimeout;
2389                   }
2390                   // Use half of whichever timeout value was more restrictive... But don't allow
2391                   // the time limit to be less than the allowable minimum (could cause an
2392                   // immediatate timeout before scanning any data).
2393                   timeLimitDelta = Math.max(timeLimitDelta / 2, minimumScanTimeLimitDelta);
2394                   timeLimit = System.currentTimeMillis() + timeLimitDelta;
2395                 }
2396 
2397                 final LimitScope sizeScope =
2398                     allowPartialResults ? LimitScope.BETWEEN_CELLS : LimitScope.BETWEEN_ROWS;
2399                 final LimitScope timeScope =
2400                     allowHeartbeatMessages ? LimitScope.BETWEEN_CELLS : LimitScope.BETWEEN_ROWS;
2401 
2402                 // Configure with limits for this RPC. Set keep progress true since size progress
2403                 // towards size limit should be kept between calls to nextRaw
2404                 ScannerContext.Builder contextBuilder = ScannerContext.newBuilder(true);
2405                 contextBuilder.setSizeLimit(sizeScope, maxResultSize);
2406                 contextBuilder.setBatchLimit(scanner.getBatch());
2407                 contextBuilder.setTimeLimit(timeScope, timeLimit);
2408                 ScannerContext scannerContext = contextBuilder.build();
2409 
2410                 boolean limitReached = false;
2411                 while (i < rows) {
2412                   // Reset the batch progress to 0 before every call to RegionScanner#nextRaw. The
2413                   // batch limit is a limit on the number of cells per Result. Thus, if progress is
2414                   // being tracked (i.e. scannerContext.keepProgress() is true) then we need to
2415                   // reset the batch progress between nextRaw invocations since we don't want the
2416                   // batch progress from previous calls to affect future calls
2417                   scannerContext.setBatchProgress(0);
2418 
2419                   // Collect values to be returned here
2420                   moreRows = scanner.nextRaw(values, scannerContext);
2421 
2422                   if (!values.isEmpty()) {
2423                     for (Cell cell : values) {
2424                       totalCellSize += CellUtil.estimatedSerializedSizeOf(cell);
2425                     }
2426                     final boolean partial = scannerContext.partialResultFormed();
2427                     results.add(Result.create(values, null, stale, partial));
2428                     i++;
2429                   }
2430 
2431                   boolean sizeLimitReached = scannerContext.checkSizeLimit(LimitScope.BETWEEN_ROWS);
2432                   boolean timeLimitReached = scannerContext.checkTimeLimit(LimitScope.BETWEEN_ROWS);
2433                   boolean rowLimitReached = i >= rows;
2434                   limitReached = sizeLimitReached || timeLimitReached || rowLimitReached;
2435 
2436                   if (limitReached || !moreRows) {
2437                     if (LOG.isTraceEnabled()) {
2438                       LOG.trace("Done scanning. limitReached: " + limitReached + " moreRows: "
2439                           + moreRows + " scannerContext: " + scannerContext);
2440                     }
2441                     // We only want to mark a ScanResponse as a heartbeat message in the event that
2442                     // there are more values to be read server side. If there aren't more values,
2443                     // marking it as a heartbeat is wasteful because the client will need to issue
2444                     // another ScanRequest only to realize that they already have all the values
2445                     if (moreRows) {
2446                       // Heartbeat messages occur when the time limit has been reached.
2447                       builder.setHeartbeatMessage(timeLimitReached);
2448                     }
2449                     break;
2450                   }
2451                   values.clear();
2452                 }
2453 
2454                 if (limitReached || moreRows) {
2455                   // We stopped prematurely
2456                   builder.setMoreResultsInRegion(true);
2457                 } else {
2458                   // We didn't get a single batch
2459                   builder.setMoreResultsInRegion(false);
2460                 }
2461               }
2462               region.updateReadRequestsCount(i);
2463               region.getMetrics().updateScanNext(totalCellSize);
2464               if (regionServer.metricsRegionServer != null) {
2465                 regionServer.metricsRegionServer.updateScannerNext(totalCellSize);
2466               }
2467             } finally {
2468               region.closeRegionOperation();
2469             }
2470 
2471             // coprocessor postNext hook
2472             if (region != null && region.getCoprocessorHost() != null) {
2473               region.getCoprocessorHost().postScannerNext(scanner, results, rows, true);
2474             }
2475           }
2476 
2477           quota.addScanResult(results);
2478 
2479           // If the scanner's filter - if any - is done with the scan
2480           // and wants to tell the client to stop the scan. This is done by passing
2481           // a null result, and setting moreResults to false.
2482           if (scanner.isFilterDone() && results.isEmpty()) {
2483             moreResults = false;
2484             results = null;
2485           } else {
2486             addResults(builder, results, controller, RegionReplicaUtil.isDefaultReplica(region.getRegionInfo()));
2487           }
2488         } catch (IOException e) {
2489           // if we have an exception on scanner next and we are using the callSeq
2490           // we should rollback because the client will retry with the same callSeq
2491           // and get an OutOfOrderScannerNextException if we don't do so.
2492           if (rsh != null && request.hasNextCallSeq()) {
2493             rsh.rollbackNextCallSeq();
2494           }
2495           throw e;
2496         } finally {
2497           // We're done. On way out re-add the above removed lease.
2498           // Adding resets expiration time on lease.
2499           if (scanners.containsKey(scannerName)) {
2500             if (lease != null) regionServer.leases.addLease(lease);
2501             ttl = this.scannerLeaseTimeoutPeriod;
2502           }
2503         }
2504       }
2505 
2506       if (!moreResults || closeScanner) {
2507         ttl = 0;
2508         moreResults = false;
2509         if (region != null && region.getCoprocessorHost() != null) {
2510           if (region.getCoprocessorHost().preScannerClose(scanner)) {
2511             return builder.build(); // bypass
2512           }
2513         }
2514         rsh = scanners.remove(scannerName);
2515         if (rsh != null) {
2516           scanner = rsh.s;
2517           scanner.close();
2518           regionServer.leases.cancelLease(scannerName);
2519           if (region != null && region.getCoprocessorHost() != null) {
2520             region.getCoprocessorHost().postScannerClose(scanner);
2521           }
2522         }
2523       }
2524 
2525       if (ttl > 0) {
2526         builder.setTtl(ttl);
2527       }
2528       builder.setScannerId(scannerId);
2529       builder.setMoreResults(moreResults);
2530       return builder.build();
2531     } catch (IOException ie) {
2532       if (scannerName != null && ie instanceof NotServingRegionException) {
2533         RegionScannerHolder rsh = scanners.remove(scannerName);
2534         if (rsh != null) {
2535           try {
2536             RegionScanner scanner = rsh.s;
2537             LOG.warn(scannerName + " encountered " + ie.getMessage() + ", closing ...");
2538             scanner.close();
2539             regionServer.leases.cancelLease(scannerName);
2540           } catch (IOException e) {
2541            LOG.warn("Getting exception closing " + scannerName, e);
2542           }
2543         }
2544       }
2545       throw new ServiceException(ie);
2546     } finally {
2547       if (quota != null) {
2548         quota.close();
2549       }
2550     }
2551   }
2552 
2553   @Override
2554   public CoprocessorServiceResponse execRegionServerService(RpcController controller,
2555       CoprocessorServiceRequest request) throws ServiceException {
2556     return regionServer.execRegionServerService(controller, request);
2557   }
2558 
2559   @Override
2560   public UpdateConfigurationResponse updateConfiguration(
2561       RpcController controller, UpdateConfigurationRequest request)
2562       throws ServiceException {
2563     try {
2564       this.regionServer.updateConfiguration();
2565     } catch (Exception e) {
2566       throw new ServiceException(e);
2567     }
2568     return UpdateConfigurationResponse.getDefaultInstance();
2569   }
2570 }