1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.regionserver;
20
21 import com.google.common.annotations.VisibleForTesting;
22 import com.google.protobuf.ByteString;
23 import com.google.protobuf.Message;
24 import com.google.protobuf.RpcController;
25 import com.google.protobuf.ServiceException;
26 import com.google.protobuf.TextFormat;
27
28 import java.io.FileNotFoundException;
29 import java.io.IOException;
30 import java.io.InterruptedIOException;
31 import java.net.BindException;
32 import java.net.InetSocketAddress;
33 import java.net.UnknownHostException;
34 import java.util.ArrayList;
35 import java.util.Collections;
36 import java.util.HashMap;
37 import java.util.Iterator;
38 import java.util.List;
39 import java.util.Map;
40 import java.util.Map.Entry;
41 import java.util.NavigableMap;
42 import java.util.Set;
43 import java.util.TreeSet;
44 import java.util.concurrent.ConcurrentHashMap;
45 import java.util.concurrent.ConcurrentMap;
46 import java.util.concurrent.atomic.AtomicLong;
47
48 import org.apache.commons.lang.mutable.MutableObject;
49 import org.apache.commons.logging.Log;
50 import org.apache.commons.logging.LogFactory;
51 import org.apache.hadoop.conf.Configuration;
52 import org.apache.hadoop.hbase.Cell;
53 import org.apache.hadoop.hbase.CellScannable;
54 import org.apache.hadoop.hbase.CellScanner;
55 import org.apache.hadoop.hbase.CellUtil;
56 import org.apache.hadoop.hbase.DoNotRetryIOException;
57 import org.apache.hadoop.hbase.DroppedSnapshotException;
58 import org.apache.hadoop.hbase.HBaseIOException;
59 import org.apache.hadoop.hbase.HConstants;
60 import org.apache.hadoop.hbase.HRegionInfo;
61 import org.apache.hadoop.hbase.HTableDescriptor;
62 import org.apache.hadoop.hbase.MetaTableAccessor;
63 import org.apache.hadoop.hbase.MultiActionResultTooLarge;
64 import org.apache.hadoop.hbase.NotServingRegionException;
65 import org.apache.hadoop.hbase.ServerName;
66 import org.apache.hadoop.hbase.TableName;
67 import org.apache.hadoop.hbase.UnknownScannerException;
68 import org.apache.hadoop.hbase.classification.InterfaceAudience;
69 import org.apache.hadoop.hbase.client.Append;
70 import org.apache.hadoop.hbase.client.ConnectionUtils;
71 import org.apache.hadoop.hbase.client.Delete;
72 import org.apache.hadoop.hbase.client.Durability;
73 import org.apache.hadoop.hbase.client.Get;
74 import org.apache.hadoop.hbase.client.Increment;
75 import org.apache.hadoop.hbase.client.Mutation;
76 import org.apache.hadoop.hbase.client.Put;
77 import org.apache.hadoop.hbase.client.RegionReplicaUtil;
78 import org.apache.hadoop.hbase.client.Result;
79 import org.apache.hadoop.hbase.client.RowMutations;
80 import org.apache.hadoop.hbase.client.Scan;
81 import org.apache.hadoop.hbase.client.VersionInfoUtil;
82 import org.apache.hadoop.hbase.conf.ConfigurationObserver;
83 import org.apache.hadoop.hbase.coordination.CloseRegionCoordination;
84 import org.apache.hadoop.hbase.coordination.OpenRegionCoordination;
85 import org.apache.hadoop.hbase.exceptions.FailedSanityCheckException;
86 import org.apache.hadoop.hbase.exceptions.MergeRegionException;
87 import org.apache.hadoop.hbase.exceptions.OperationConflictException;
88 import org.apache.hadoop.hbase.exceptions.OutOfOrderScannerNextException;
89 import org.apache.hadoop.hbase.exceptions.ScannerResetException;
90 import org.apache.hadoop.hbase.filter.ByteArrayComparable;
91 import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
92 import org.apache.hadoop.hbase.ipc.HBaseRPCErrorHandler;
93 import org.apache.hadoop.hbase.ipc.PayloadCarryingRpcController;
94 import org.apache.hadoop.hbase.ipc.PriorityFunction;
95 import org.apache.hadoop.hbase.ipc.QosPriority;
96 import org.apache.hadoop.hbase.ipc.RpcCallContext;
97 import org.apache.hadoop.hbase.ipc.RpcServer;
98 import org.apache.hadoop.hbase.ipc.RpcServer.BlockingServiceAndInterface;
99 import org.apache.hadoop.hbase.ipc.RpcServerInterface;
100 import org.apache.hadoop.hbase.ipc.ServerNotRunningYetException;
101 import org.apache.hadoop.hbase.ipc.ServerRpcController;
102 import org.apache.hadoop.hbase.master.MasterRpcServices;
103 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
104 import org.apache.hadoop.hbase.protobuf.RequestConverter;
105 import org.apache.hadoop.hbase.protobuf.ResponseConverter;
106 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.AdminService;
107 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.CloseRegionRequest;
108 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.CloseRegionResponse;
109 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.CompactRegionRequest;
110 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.CompactRegionResponse;
111 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.FlushRegionRequest;
112 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.FlushRegionResponse;
113 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.GetOnlineRegionRequest;
114 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.GetOnlineRegionResponse;
115 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.GetRegionInfoRequest;
116 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.GetRegionInfoResponse;
117 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.GetServerInfoRequest;
118 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.GetServerInfoResponse;
119 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.GetStoreFileRequest;
120 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.GetStoreFileResponse;
121 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.MergeRegionsRequest;
122 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.MergeRegionsResponse;
123 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.OpenRegionRequest;
124 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.OpenRegionRequest.RegionOpenInfo;
125 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.OpenRegionResponse;
126 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.OpenRegionResponse.RegionOpeningState;
127 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.ReplicateWALEntryRequest;
128 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.ReplicateWALEntryResponse;
129 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.RollWALWriterRequest;
130 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.RollWALWriterResponse;
131 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.SplitRegionRequest;
132 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.SplitRegionResponse;
133 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.StopServerRequest;
134 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.StopServerResponse;
135 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.UpdateConfigurationRequest;
136 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.UpdateConfigurationResponse;
137 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.UpdateFavoredNodesRequest;
138 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.UpdateFavoredNodesResponse;
139 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.WALEntry;
140 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.WarmupRegionRequest;
141 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.WarmupRegionResponse;
142 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
143 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.Action;
144 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.BulkLoadHFileRequest;
145 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.BulkLoadHFileRequest.FamilyPath;
146 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.BulkLoadHFileResponse;
147 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService;
148 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.Condition;
149 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.CoprocessorServiceRequest;
150 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.CoprocessorServiceResponse;
151 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.GetRequest;
152 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.GetResponse;
153 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MultiRegionLoadStats;
154 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MultiRequest;
155 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MultiResponse;
156 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutateRequest;
157 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutateResponse;
158 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProto;
159 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProto.MutationType;
160 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.RegionAction;
161 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.RegionActionResult;
162 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ResultOrException;
163 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ScanRequest;
164 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ScanResponse;
165 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.NameBytesPair;
166 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.NameInt64Pair;
167 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.RegionInfo;
168 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.RegionSpecifier;
169 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.RegionSpecifier.RegionSpecifierType;
170 import org.apache.hadoop.hbase.protobuf.generated.MapReduceProtos.ScanMetrics;
171 import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader;
172 import org.apache.hadoop.hbase.protobuf.generated.WALProtos.BulkLoadDescriptor;
173 import org.apache.hadoop.hbase.protobuf.generated.WALProtos.CompactionDescriptor;
174 import org.apache.hadoop.hbase.protobuf.generated.WALProtos.FlushDescriptor;
175 import org.apache.hadoop.hbase.protobuf.generated.WALProtos.RegionEventDescriptor;
176 import org.apache.hadoop.hbase.quotas.OperationQuota;
177 import org.apache.hadoop.hbase.quotas.RegionServerQuotaManager;
178 import org.apache.hadoop.hbase.regionserver.Leases.LeaseStillHeldException;
179 import org.apache.hadoop.hbase.regionserver.Region.Operation;
180 import org.apache.hadoop.hbase.regionserver.ScannerContext.LimitScope;
181 import org.apache.hadoop.hbase.regionserver.handler.OpenMetaHandler;
182 import org.apache.hadoop.hbase.regionserver.handler.OpenPriorityRegionHandler;
183 import org.apache.hadoop.hbase.regionserver.handler.OpenRegionHandler;
184 import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
185 import org.apache.hadoop.hbase.security.User;
186 import org.apache.hadoop.hbase.util.Bytes;
187 import org.apache.hadoop.hbase.util.Counter;
188 import org.apache.hadoop.hbase.util.DNS;
189 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
190 import org.apache.hadoop.hbase.util.Pair;
191 import org.apache.hadoop.hbase.util.ServerRegionReplicaUtil;
192 import org.apache.hadoop.hbase.util.Strings;
193 import org.apache.hadoop.hbase.wal.WAL;
194 import org.apache.hadoop.hbase.wal.WALKey;
195 import org.apache.hadoop.hbase.wal.WALSplitter;
196 import org.apache.hadoop.hbase.zookeeper.ZKSplitLog;
197 import org.apache.zookeeper.KeeperException;
198
199
200
201
202 @InterfaceAudience.Private
203 @SuppressWarnings("deprecation")
204 public class RSRpcServices implements HBaseRPCErrorHandler,
205 AdminService.BlockingInterface, ClientService.BlockingInterface, PriorityFunction,
206 ConfigurationObserver {
207 protected static final Log LOG = LogFactory.getLog(RSRpcServices.class);
208
209
210 public static final String REGION_SERVER_RPC_SCHEDULER_FACTORY_CLASS =
211 "hbase.region.server.rpc.scheduler.factory.class";
212
213
214
215
216
217
218 private static final String REGION_SERVER_RPC_MINIMUM_SCAN_TIME_LIMIT_DELTA =
219 "hbase.region.server.rpc.minimum.scan.time.limit.delta";
220
221
222
223 private static final long DEFAULT_REGION_SERVER_RPC_MINIMUM_SCAN_TIME_LIMIT_DELTA = 10;
224
225
226 final Counter requestCount = new Counter();
227
228
229 final Counter rpcGetRequestCount = new Counter();
230
231
232 final Counter rpcScanRequestCount = new Counter();
233
234
235 final Counter rpcMultiRequestCount = new Counter();
236
237
238 final Counter rpcMutateRequestCount = new Counter();
239
240
241 final RpcServerInterface rpcServer;
242 final InetSocketAddress isa;
243
244 private final HRegionServer regionServer;
245 private final long maxScannerResultSize;
246
247
248 private final PriorityFunction priority;
249
250 private final AtomicLong scannerIdGen = new AtomicLong(0L);
251 private final ConcurrentMap<String, RegionScannerHolder> scanners = new ConcurrentHashMap<>();
252
253
254
255
256 private final int scannerLeaseTimeoutPeriod;
257
258
259
260
261 private final int rpcTimeout;
262
263
264
265
266 private final long minimumScanTimeLimitDelta;
267
268
269
270
271 private static final class RegionScannerHolder {
272
273 private final AtomicLong nextCallSeq = new AtomicLong(0);
274 private final String scannerName;
275 private final RegionScanner s;
276 private final Region r;
277
278 public RegionScannerHolder(String scannerName, RegionScanner s, Region r) {
279 this.scannerName = scannerName;
280 this.s = s;
281 this.r = r;
282 }
283
284 public long getNextCallSeq() {
285 return nextCallSeq.get();
286 }
287
288 public boolean incNextCallSeq(long currentSeq) {
289
290 return nextCallSeq.compareAndSet(currentSeq, currentSeq + 1);
291 }
292 }
293
294
295
296
297
298 private class ScannerListener implements LeaseListener {
299 private final String scannerName;
300
301 ScannerListener(final String n) {
302 this.scannerName = n;
303 }
304
305 @Override
306 public void leaseExpired() {
307 RegionScannerHolder rsh = scanners.remove(this.scannerName);
308 if (rsh != null) {
309 RegionScanner s = rsh.s;
310 LOG.info("Scanner " + this.scannerName + " lease expired on region "
311 + s.getRegionInfo().getRegionNameAsString());
312 Region region = null;
313 try {
314 region = regionServer.getRegion(s.getRegionInfo().getRegionName());
315 if (region != null && region.getCoprocessorHost() != null) {
316 region.getCoprocessorHost().preScannerClose(s);
317 }
318 } catch (IOException e) {
319 LOG.error("Closing scanner for " + s.getRegionInfo().getRegionNameAsString(), e);
320 } finally {
321 try {
322 s.close();
323 if (region != null && region.getCoprocessorHost() != null) {
324 region.getCoprocessorHost().postScannerClose(s);
325 }
326 } catch (IOException e) {
327 LOG.error("Closing scanner for " + s.getRegionInfo().getRegionNameAsString(), e);
328 }
329 }
330 } else {
331 LOG.warn("Scanner " + this.scannerName + " lease expired, but no related" +
332 " scanner found, hence no chance to close that related scanner!");
333 }
334 }
335 }
336
337 private static ResultOrException getResultOrException(final ClientProtos.Result r,
338 final int index){
339 return getResultOrException(ResponseConverter.buildActionResult(r), index);
340 }
341
342 private static ResultOrException getResultOrException(final Exception e, final int index) {
343 return getResultOrException(ResponseConverter.buildActionResult(e), index);
344 }
345
346 private static ResultOrException getResultOrException(
347 final ResultOrException.Builder builder, final int index) {
348 return builder.setIndex(index).build();
349 }
350
351
352
353
354
355
356
357 private boolean startNonceOperation(final MutationProto mutation, long nonceGroup)
358 throws IOException, OperationConflictException {
359 if (regionServer.nonceManager == null || !mutation.hasNonce()) return true;
360 boolean canProceed = false;
361 try {
362 canProceed = regionServer.nonceManager.startOperation(
363 nonceGroup, mutation.getNonce(), regionServer);
364 } catch (InterruptedException ex) {
365 throw new InterruptedIOException("Nonce start operation interrupted");
366 }
367 return canProceed;
368 }
369
370
371
372
373
374
375
376 private void endNonceOperation(final MutationProto mutation,
377 long nonceGroup, boolean success) {
378 if (regionServer.nonceManager != null && mutation.hasNonce()) {
379 regionServer.nonceManager.endOperation(nonceGroup, mutation.getNonce(), success);
380 }
381 }
382
383
384
385
386 private boolean isClientCellBlockSupport() {
387 return isClientCellBlockSupport(RpcServer.getCurrentCall());
388 }
389
390 private boolean isClientCellBlockSupport(RpcCallContext context) {
391 return context != null && context.isClientCellBlockSupported();
392 }
393
394 private void addResult(final MutateResponse.Builder builder,
395 final Result result, final PayloadCarryingRpcController rpcc) {
396 if (result == null) return;
397 if (isClientCellBlockSupport()) {
398 builder.setResult(ProtobufUtil.toResultNoData(result));
399 rpcc.setCellScanner(result.cellScanner());
400 } else {
401 ClientProtos.Result pbr = ProtobufUtil.toResult(result);
402 builder.setResult(pbr);
403 }
404 }
405
406 private void addResults(final ScanResponse.Builder builder, final List<Result> results,
407 final PayloadCarryingRpcController controller, boolean isDefaultRegion) {
408 builder.setStale(!isDefaultRegion);
409 if (results == null || results.isEmpty()) return;
410 if (isClientCellBlockSupport()) {
411 for (Result res : results) {
412 builder.addCellsPerResult(res.size());
413 builder.addPartialFlagPerResult(res.isPartial());
414 }
415 controller.setCellScanner(CellUtil.createCellScanner(results));
416 } else {
417 for (Result res : results) {
418 ClientProtos.Result pbr = ProtobufUtil.toResult(res);
419 builder.addResults(pbr);
420 }
421 }
422 }
423
424
425
426
427
428
429
430
431
432 private void mutateRows(final Region region,
433 final List<ClientProtos.Action> actions,
434 final CellScanner cellScanner, RegionActionResult.Builder builder) throws IOException {
435 if (!region.getRegionInfo().isMetaTable()) {
436 regionServer.cacheFlusher.reclaimMemStoreMemory();
437 }
438 RowMutations rm = null;
439 int i = 0;
440 ClientProtos.ResultOrException.Builder resultOrExceptionOrBuilder =
441 ClientProtos.ResultOrException.newBuilder();
442 for (ClientProtos.Action action: actions) {
443 if (action.hasGet()) {
444 throw new DoNotRetryIOException("Atomic put and/or delete only, not a Get=" +
445 action.getGet());
446 }
447 MutationType type = action.getMutation().getMutateType();
448 if (rm == null) {
449 rm = new RowMutations(action.getMutation().getRow().toByteArray());
450 }
451 switch (type) {
452 case PUT:
453 rm.add(ProtobufUtil.toPut(action.getMutation(), cellScanner));
454 break;
455 case DELETE:
456 rm.add(ProtobufUtil.toDelete(action.getMutation(), cellScanner));
457 break;
458 default:
459 throw new DoNotRetryIOException("Atomic put and/or delete only, not " + type.name());
460 }
461
462
463 resultOrExceptionOrBuilder.clear();
464 resultOrExceptionOrBuilder.setIndex(i++);
465 builder.addResultOrException(
466 resultOrExceptionOrBuilder.build());
467 }
468 region.mutateRow(rm);
469 }
470
471
472
473
474
475
476
477
478
479
480
481
482
483 private boolean checkAndRowMutate(final Region region, final List<ClientProtos.Action> actions,
484 final CellScanner cellScanner, byte[] row, byte[] family, byte[] qualifier,
485 CompareOp compareOp, ByteArrayComparable comparator,
486 RegionActionResult.Builder builder) throws IOException {
487 if (!region.getRegionInfo().isMetaTable()) {
488 regionServer.cacheFlusher.reclaimMemStoreMemory();
489 }
490 RowMutations rm = null;
491 int i = 0;
492 ClientProtos.ResultOrException.Builder resultOrExceptionOrBuilder =
493 ClientProtos.ResultOrException.newBuilder();
494 for (ClientProtos.Action action: actions) {
495 if (action.hasGet()) {
496 throw new DoNotRetryIOException("Atomic put and/or delete only, not a Get=" +
497 action.getGet());
498 }
499 MutationType type = action.getMutation().getMutateType();
500 if (rm == null) {
501 rm = new RowMutations(action.getMutation().getRow().toByteArray());
502 }
503 switch (type) {
504 case PUT:
505 rm.add(ProtobufUtil.toPut(action.getMutation(), cellScanner));
506 break;
507 case DELETE:
508 rm.add(ProtobufUtil.toDelete(action.getMutation(), cellScanner));
509 break;
510 default:
511 throw new DoNotRetryIOException("Atomic put and/or delete only, not " + type.name());
512 }
513
514
515 resultOrExceptionOrBuilder.clear();
516 resultOrExceptionOrBuilder.setIndex(i++);
517 builder.addResultOrException(
518 resultOrExceptionOrBuilder.build());
519 }
520 return region.checkAndRowMutate(row, family, qualifier, compareOp,
521 comparator, rm, Boolean.TRUE);
522 }
523
524
525
526
527
528
529
530
531
532
533
534
535 private Result append(final Region region, final OperationQuota quota,
536 final MutationProto mutation, final CellScanner cellScanner, long nonceGroup)
537 throws IOException {
538 long before = EnvironmentEdgeManager.currentTime();
539 Append append = ProtobufUtil.toAppend(mutation, cellScanner);
540 quota.addMutation(append);
541 Result r = null;
542 if (region.getCoprocessorHost() != null) {
543 r = region.getCoprocessorHost().preAppend(append);
544 }
545 if (r == null) {
546 boolean canProceed = startNonceOperation(mutation, nonceGroup);
547 boolean success = false;
548 try {
549 long nonce = mutation.hasNonce() ? mutation.getNonce() : HConstants.NO_NONCE;
550 if (canProceed) {
551 r = region.append(append, nonceGroup, nonce);
552 } else {
553
554 List<Cell> results = region.get(ProtobufUtil.toGet(mutation, cellScanner), false,
555 nonceGroup, nonce);
556 r = Result.create(results);
557 }
558 success = true;
559 } finally {
560 if (canProceed) {
561 endNonceOperation(mutation, nonceGroup, success);
562 }
563 }
564 if (region.getCoprocessorHost() != null) {
565 region.getCoprocessorHost().postAppend(append, r);
566 }
567 }
568 if (regionServer.metricsRegionServer != null) {
569 regionServer.metricsRegionServer.updateAppend(
570 EnvironmentEdgeManager.currentTime() - before);
571 }
572 return r;
573 }
574
575
576
577
578
579
580
581
582
583
584
585 private Result increment(final Region region, final OperationQuota quota,
586 final MutationProto mutation, final CellScanner cellScanner, long nonceGroup)
587 throws IOException {
588 long before = EnvironmentEdgeManager.currentTime();
589 Increment increment = ProtobufUtil.toIncrement(mutation, cellScanner);
590 quota.addMutation(increment);
591 Result r = null;
592 if (region.getCoprocessorHost() != null) {
593 r = region.getCoprocessorHost().preIncrement(increment);
594 }
595 if (r == null) {
596 boolean canProceed = startNonceOperation(mutation, nonceGroup);
597 boolean success = false;
598 try {
599 long nonce = mutation.hasNonce() ? mutation.getNonce() : HConstants.NO_NONCE;
600 if (canProceed) {
601 r = region.increment(increment, nonceGroup, nonce);
602 } else {
603
604 List<Cell> results = region.get(ProtobufUtil.toGet(mutation, cellScanner), false,
605 nonceGroup, nonce);
606 r = Result.create(results);
607 }
608 success = true;
609 } finally {
610 if (canProceed) {
611 endNonceOperation(mutation, nonceGroup, success);
612 }
613 }
614 if (region.getCoprocessorHost() != null) {
615 r = region.getCoprocessorHost().postIncrement(increment, r);
616 }
617 }
618 if (regionServer.metricsRegionServer != null) {
619 regionServer.metricsRegionServer.updateIncrement(
620 EnvironmentEdgeManager.currentTime() - before);
621 }
622 return r;
623 }
624
625
626
627
628
629
630
631
632
633
634
635
636 private List<CellScannable> doNonAtomicRegionMutation(final Region region,
637 final OperationQuota quota, final RegionAction actions, final CellScanner cellScanner,
638 final RegionActionResult.Builder builder, List<CellScannable> cellsToReturn, long nonceGroup) {
639
640
641
642
643 List<ClientProtos.Action> mutations = null;
644 long maxQuotaResultSize = Math.min(maxScannerResultSize, quota.getReadAvailable());
645 RpcCallContext context = RpcServer.getCurrentCall();
646 IOException sizeIOE = null;
647 Object lastBlock = null;
648 ClientProtos.ResultOrException.Builder resultOrExceptionBuilder = ResultOrException.newBuilder();
649 boolean hasResultOrException = false;
650 for (ClientProtos.Action action : actions.getActionList()) {
651 hasResultOrException = false;
652 resultOrExceptionBuilder.clear();
653 try {
654 Result r = null;
655
656 if (context != null
657 && context.isRetryImmediatelySupported()
658 && (context.getResponseCellSize() > maxQuotaResultSize
659 || context.getResponseBlockSize() + context.getResponseExceptionSize()
660 > maxQuotaResultSize)) {
661
662
663
664 if (sizeIOE == null ) {
665
666
667
668
669 sizeIOE = new MultiActionResultTooLarge("Max size exceeded"
670 + " CellSize: " + context.getResponseCellSize()
671 + " BlockSize: " + context.getResponseBlockSize());
672
673
674
675 rpcServer.getMetrics().exception(sizeIOE);
676 }
677
678
679
680
681
682 hasResultOrException = true;
683 NameBytesPair pair = ResponseConverter.buildException(sizeIOE);
684 resultOrExceptionBuilder.setException(pair);
685 context.incrementResponseExceptionSize(pair.getSerializedSize());
686 resultOrExceptionBuilder.setIndex(action.getIndex());
687 builder.addResultOrException(resultOrExceptionBuilder.build());
688 if (cellScanner != null) {
689 skipCellsForMutation(action, cellScanner);
690 }
691 continue;
692 }
693 if (action.hasGet()) {
694 long before = EnvironmentEdgeManager.currentTime();
695 try {
696 Get get = ProtobufUtil.toGet(action.getGet());
697 r = region.get(get);
698 } finally {
699 if (regionServer.metricsRegionServer != null) {
700 regionServer.metricsRegionServer.updateGet(
701 EnvironmentEdgeManager.currentTime() - before);
702 }
703 }
704 } else if (action.hasServiceCall()) {
705 hasResultOrException = true;
706 try {
707 Message result = execServiceOnRegion(region, action.getServiceCall());
708 ClientProtos.CoprocessorServiceResult.Builder serviceResultBuilder =
709 ClientProtos.CoprocessorServiceResult.newBuilder();
710 resultOrExceptionBuilder.setServiceResult(
711 serviceResultBuilder.setValue(
712 serviceResultBuilder.getValueBuilder()
713 .setName(result.getClass().getName())
714 .setValue(result.toByteString())));
715 } catch (IOException ioe) {
716 rpcServer.getMetrics().exception(ioe);
717 NameBytesPair pair = ResponseConverter.buildException(ioe);
718 resultOrExceptionBuilder.setException(pair);
719 context.incrementResponseExceptionSize(pair.getSerializedSize());
720 }
721 } else if (action.hasMutation()) {
722 MutationType type = action.getMutation().getMutateType();
723 if (type != MutationType.PUT && type != MutationType.DELETE && mutations != null &&
724 !mutations.isEmpty()) {
725
726 doBatchOp(builder, region, quota, mutations, cellScanner);
727 mutations.clear();
728 }
729 switch (type) {
730 case APPEND:
731 r = append(region, quota, action.getMutation(), cellScanner, nonceGroup);
732 break;
733 case INCREMENT:
734 r = increment(region, quota, action.getMutation(), cellScanner, nonceGroup);
735 break;
736 case PUT:
737 case DELETE:
738
739 if (mutations == null) {
740 mutations = new ArrayList<ClientProtos.Action>(actions.getActionCount());
741 }
742 mutations.add(action);
743 break;
744 default:
745 throw new DoNotRetryIOException("Unsupported mutate type: " + type.name());
746 }
747 } else {
748 throw new HBaseIOException("Unexpected Action type");
749 }
750 if (r != null) {
751 ClientProtos.Result pbResult = null;
752 if (isClientCellBlockSupport()) {
753 pbResult = ProtobufUtil.toResultNoData(r);
754
755 if (cellsToReturn == null) {
756 cellsToReturn = new ArrayList<CellScannable>();
757 }
758 cellsToReturn.add(r);
759 } else {
760 pbResult = ProtobufUtil.toResult(r);
761 }
762 lastBlock = addSize(context, r, lastBlock);
763 hasResultOrException = true;
764 resultOrExceptionBuilder.setResult(pbResult);
765 }
766
767
768
769
770 } catch (IOException ie) {
771 rpcServer.getMetrics().exception(ie);
772 hasResultOrException = true;
773 NameBytesPair pair = ResponseConverter.buildException(ie);
774 resultOrExceptionBuilder.setException(pair);
775 context.incrementResponseExceptionSize(pair.getSerializedSize());
776 }
777 if (hasResultOrException) {
778
779 resultOrExceptionBuilder.setIndex(action.getIndex());
780 builder.addResultOrException(resultOrExceptionBuilder.build());
781 }
782 }
783
784 if (mutations != null && !mutations.isEmpty()) {
785 doBatchOp(builder, region, quota, mutations, cellScanner);
786 }
787 return cellsToReturn;
788 }
789
790
791
792
793
794
795
796
797 private void doBatchOp(final RegionActionResult.Builder builder, final Region region,
798 final OperationQuota quota,
799 final List<ClientProtos.Action> mutations, final CellScanner cells) {
800 Mutation[] mArray = new Mutation[mutations.size()];
801 long before = EnvironmentEdgeManager.currentTime();
802 boolean batchContainsPuts = false, batchContainsDelete = false;
803 try {
804 int i = 0;
805 for (ClientProtos.Action action: mutations) {
806 MutationProto m = action.getMutation();
807 Mutation mutation;
808 if (m.getMutateType() == MutationType.PUT) {
809 mutation = ProtobufUtil.toPut(m, cells);
810 batchContainsPuts = true;
811 } else {
812 mutation = ProtobufUtil.toDelete(m, cells);
813 batchContainsDelete = true;
814 }
815 mArray[i++] = mutation;
816 quota.addMutation(mutation);
817 }
818
819 if (!region.getRegionInfo().isMetaTable()) {
820 regionServer.cacheFlusher.reclaimMemStoreMemory();
821 }
822
823 OperationStatus[] codes = region.batchMutate(mArray, HConstants.NO_NONCE,
824 HConstants.NO_NONCE);
825 for (i = 0; i < codes.length; i++) {
826 int index = mutations.get(i).getIndex();
827 Exception e = null;
828 switch (codes[i].getOperationStatusCode()) {
829 case BAD_FAMILY:
830 e = new NoSuchColumnFamilyException(codes[i].getExceptionMsg());
831 builder.addResultOrException(getResultOrException(e, index));
832 break;
833
834 case SANITY_CHECK_FAILURE:
835 e = new FailedSanityCheckException(codes[i].getExceptionMsg());
836 builder.addResultOrException(getResultOrException(e, index));
837 break;
838
839 default:
840 e = new DoNotRetryIOException(codes[i].getExceptionMsg());
841 builder.addResultOrException(getResultOrException(e, index));
842 break;
843
844 case SUCCESS:
845 builder.addResultOrException(getResultOrException(
846 ClientProtos.Result.getDefaultInstance(), index));
847 break;
848 }
849 }
850 } catch (IOException ie) {
851 for (int i = 0; i < mutations.size(); i++) {
852 builder.addResultOrException(getResultOrException(ie, mutations.get(i).getIndex()));
853 }
854 }
855 if (regionServer.metricsRegionServer != null) {
856 long after = EnvironmentEdgeManager.currentTime();
857 if (batchContainsPuts) {
858 regionServer.metricsRegionServer.updatePut(after - before);
859 }
860 if (batchContainsDelete) {
861 regionServer.metricsRegionServer.updateDelete(after - before);
862 }
863 }
864 }
865
866
867
868
869
870
871
872
873
874
875
876 private OperationStatus [] doReplayBatchOp(final Region region,
877 final List<WALSplitter.MutationReplay> mutations, long replaySeqId) throws IOException {
878 long before = EnvironmentEdgeManager.currentTime();
879 boolean batchContainsPuts = false, batchContainsDelete = false;
880 try {
881 for (Iterator<WALSplitter.MutationReplay> it = mutations.iterator(); it.hasNext();) {
882 WALSplitter.MutationReplay m = it.next();
883
884 if (m.type == MutationType.PUT) {
885 batchContainsPuts = true;
886 } else {
887 batchContainsDelete = true;
888 }
889
890 NavigableMap<byte[], List<Cell>> map = m.mutation.getFamilyCellMap();
891 List<Cell> metaCells = map.get(WALEdit.METAFAMILY);
892 if (metaCells != null && !metaCells.isEmpty()) {
893 for (Cell metaCell : metaCells) {
894 CompactionDescriptor compactionDesc = WALEdit.getCompaction(metaCell);
895 boolean isDefaultReplica = RegionReplicaUtil.isDefaultReplica(region.getRegionInfo());
896 HRegion hRegion = (HRegion)region;
897 if (compactionDesc != null) {
898
899
900 hRegion.replayWALCompactionMarker(compactionDesc, !isDefaultReplica, isDefaultReplica,
901 replaySeqId);
902 continue;
903 }
904 FlushDescriptor flushDesc = WALEdit.getFlushDescriptor(metaCell);
905 if (flushDesc != null && !isDefaultReplica) {
906 hRegion.replayWALFlushMarker(flushDesc, replaySeqId);
907 continue;
908 }
909 RegionEventDescriptor regionEvent = WALEdit.getRegionEventDescriptor(metaCell);
910 if (regionEvent != null && !isDefaultReplica) {
911 hRegion.replayWALRegionEventMarker(regionEvent);
912 continue;
913 }
914 BulkLoadDescriptor bulkLoadEvent = WALEdit.getBulkLoadDescriptor(metaCell);
915 if (bulkLoadEvent != null) {
916 hRegion.replayWALBulkLoadEventMarker(bulkLoadEvent);
917 continue;
918 }
919 }
920 it.remove();
921 }
922 }
923 requestCount.add(mutations.size());
924 if (!region.getRegionInfo().isMetaTable()) {
925 regionServer.cacheFlusher.reclaimMemStoreMemory();
926 }
927 return region.batchReplay(mutations.toArray(
928 new WALSplitter.MutationReplay[mutations.size()]), replaySeqId);
929 } finally {
930 if (regionServer.metricsRegionServer != null) {
931 long after = EnvironmentEdgeManager.currentTime();
932 if (batchContainsPuts) {
933 regionServer.metricsRegionServer.updatePut(after - before);
934 }
935 if (batchContainsDelete) {
936 regionServer.metricsRegionServer.updateDelete(after - before);
937 }
938 }
939 }
940 }
941
942 private void closeAllScanners() {
943
944
945 for (Map.Entry<String, RegionScannerHolder> e : scanners.entrySet()) {
946 try {
947 e.getValue().s.close();
948 } catch (IOException ioe) {
949 LOG.warn("Closing scanner " + e.getKey(), ioe);
950 }
951 }
952 }
953
954 public RSRpcServices(HRegionServer rs) throws IOException {
955 regionServer = rs;
956
957 RpcSchedulerFactory rpcSchedulerFactory;
958 try {
959 Class<?> rpcSchedulerFactoryClass = rs.conf.getClass(
960 REGION_SERVER_RPC_SCHEDULER_FACTORY_CLASS,
961 SimpleRpcSchedulerFactory.class);
962 rpcSchedulerFactory = ((RpcSchedulerFactory) rpcSchedulerFactoryClass.newInstance());
963 } catch (InstantiationException e) {
964 throw new IllegalArgumentException(e);
965 } catch (IllegalAccessException e) {
966 throw new IllegalArgumentException(e);
967 }
968
969 InetSocketAddress initialIsa;
970 InetSocketAddress bindAddress;
971 if(this instanceof MasterRpcServices) {
972 String hostname = getHostname(rs.conf, true);
973 int port = rs.conf.getInt(HConstants.MASTER_PORT, HConstants.DEFAULT_MASTER_PORT);
974
975 initialIsa = new InetSocketAddress(hostname, port);
976 bindAddress = new InetSocketAddress(rs.conf.get("hbase.master.ipc.address", hostname), port);
977 } else {
978 String hostname = getHostname(rs.conf, false);
979 int port = rs.conf.getInt(HConstants.REGIONSERVER_PORT,
980 HConstants.DEFAULT_REGIONSERVER_PORT);
981
982 initialIsa = new InetSocketAddress(hostname, port);
983 bindAddress = new InetSocketAddress(
984 rs.conf.get("hbase.regionserver.ipc.address", hostname), port);
985 }
986 if (initialIsa.getAddress() == null) {
987 throw new IllegalArgumentException("Failed resolve of " + initialIsa);
988 }
989 priority = createPriority();
990 String name = rs.getProcessName() + "/" + initialIsa.toString();
991
992 ConnectionUtils.setServerSideHConnectionRetriesConfig(rs.conf, name, LOG);
993 try {
994 rpcServer = new RpcServer(rs, name, getServices(),
995 bindAddress,
996 rs.conf,
997 rpcSchedulerFactory.create(rs.conf, this, rs));
998 rpcServer.setRsRpcServices(this);
999 } catch (BindException be) {
1000 String configName = (this instanceof MasterRpcServices) ? HConstants.MASTER_PORT :
1001 HConstants.REGIONSERVER_PORT;
1002 throw new IOException(be.getMessage() + ". To switch ports use the '" + configName +
1003 "' configuration property.", be.getCause() != null ? be.getCause() : be);
1004 }
1005
1006 scannerLeaseTimeoutPeriod = rs.conf.getInt(
1007 HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD,
1008 HConstants.DEFAULT_HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD);
1009 maxScannerResultSize = rs.conf.getLong(
1010 HConstants.HBASE_SERVER_SCANNER_MAX_RESULT_SIZE_KEY,
1011 HConstants.DEFAULT_HBASE_SERVER_SCANNER_MAX_RESULT_SIZE);
1012 rpcTimeout = rs.conf.getInt(
1013 HConstants.HBASE_RPC_TIMEOUT_KEY,
1014 HConstants.DEFAULT_HBASE_RPC_TIMEOUT);
1015 minimumScanTimeLimitDelta = rs.conf.getLong(
1016 REGION_SERVER_RPC_MINIMUM_SCAN_TIME_LIMIT_DELTA,
1017 DEFAULT_REGION_SERVER_RPC_MINIMUM_SCAN_TIME_LIMIT_DELTA);
1018
1019 InetSocketAddress address = rpcServer.getListenerAddress();
1020 if (address == null) {
1021 throw new IOException("Listener channel is closed");
1022 }
1023
1024 isa = new InetSocketAddress(initialIsa.getHostName(), address.getPort());
1025 rpcServer.setErrorHandler(this);
1026 rs.setName(name);
1027 }
1028
1029 @Override
1030 public void onConfigurationChange(Configuration newConf) {
1031 if (rpcServer instanceof ConfigurationObserver) {
1032 ((ConfigurationObserver)rpcServer).onConfigurationChange(newConf);
1033 }
1034 }
1035
1036 protected PriorityFunction createPriority() {
1037 return new AnnotationReadingPriorityFunction(this);
1038 }
1039
1040 public static String getHostname(Configuration conf, boolean isMaster)
1041 throws UnknownHostException {
1042 String hostname = conf.get(isMaster? HRegionServer.MASTER_HOSTNAME_KEY :
1043 HRegionServer.RS_HOSTNAME_KEY);
1044 if (hostname == null || hostname.isEmpty()) {
1045 String masterOrRS = isMaster ? "master" : "regionserver";
1046 return Strings.domainNamePointerToHostName(DNS.getDefaultHost(
1047 conf.get("hbase." + masterOrRS + ".dns.interface", "default"),
1048 conf.get("hbase." + masterOrRS + ".dns.nameserver", "default")));
1049 } else {
1050 LOG.info("hostname is configured to be " + hostname);
1051 return hostname;
1052 }
1053 }
1054
1055 public
1056 RegionScanner getScanner(long scannerId) {
1057 String scannerIdString = Long.toString(scannerId);
1058 RegionScannerHolder scannerHolder = scanners.get(scannerIdString);
1059 if (scannerHolder != null) {
1060 return scannerHolder.s;
1061 }
1062 return null;
1063 }
1064
1065 public String getScanDetailsWithId(long scannerId) {
1066 RegionScanner scanner = getScanner(scannerId);
1067 if (scanner == null) {
1068 return null;
1069 }
1070 StringBuilder builder = new StringBuilder();
1071 builder.append("table: ").append(scanner.getRegionInfo().getTable().getNameAsString());
1072 builder.append(" region: ").append(scanner.getRegionInfo().getRegionNameAsString());
1073 return builder.toString();
1074 }
1075
1076
1077
1078
1079
1080 long getScannerVirtualTime(long scannerId) {
1081 String scannerIdString = Long.toString(scannerId);
1082 RegionScannerHolder scannerHolder = scanners.get(scannerIdString);
1083 if (scannerHolder != null) {
1084 return scannerHolder.getNextCallSeq();
1085 }
1086 return 0L;
1087 }
1088
1089
1090
1091
1092
1093 Object addSize(RpcCallContext context, Result r, Object lastBlock) {
1094 if (context != null && r != null && !r.isEmpty()) {
1095 for (Cell c : r.rawCells()) {
1096 context.incrementResponseCellSize(CellUtil.estimatedHeapSizeOf(c));
1097
1098
1099
1100
1101
1102
1103 byte[] valueArray = c.getValueArray();
1104 if (valueArray != lastBlock) {
1105 context.incrementResponseBlockSize(valueArray.length);
1106 lastBlock = valueArray;
1107 }
1108 }
1109 }
1110 return lastBlock;
1111 }
1112
1113 private RegionScannerHolder addScanner(String scannerName, RegionScanner s, Region r)
1114 throws LeaseStillHeldException {
1115 regionServer.leases.createLease(scannerName, this.scannerLeaseTimeoutPeriod,
1116 new ScannerListener(scannerName));
1117 RegionScannerHolder rsh = new RegionScannerHolder(scannerName, s, r);
1118 RegionScannerHolder existing = scanners.putIfAbsent(scannerName, rsh);
1119 assert existing == null : "scannerId must be unique within regionserver's whole lifecycle!";
1120 return rsh;
1121 }
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131 @VisibleForTesting
1132 public Region getRegion(
1133 final RegionSpecifier regionSpecifier) throws IOException {
1134 ByteString value = regionSpecifier.getValue();
1135 RegionSpecifierType type = regionSpecifier.getType();
1136 switch (type) {
1137 case REGION_NAME:
1138 byte[] regionName = value.toByteArray();
1139 String encodedRegionName = HRegionInfo.encodeRegionName(regionName);
1140 return regionServer.getRegionByEncodedName(regionName, encodedRegionName);
1141 case ENCODED_REGION_NAME:
1142 return regionServer.getRegionByEncodedName(value.toStringUtf8());
1143 default:
1144 throw new DoNotRetryIOException(
1145 "Unsupported region specifier type: " + type);
1146 }
1147 }
1148
1149 @VisibleForTesting
1150 public PriorityFunction getPriority() {
1151 return priority;
1152 }
1153
1154 @VisibleForTesting
1155 public Configuration getConfiguration() {
1156 return regionServer.getConfiguration();
1157 }
1158
1159 private RegionServerQuotaManager getQuotaManager() {
1160 return regionServer.getRegionServerQuotaManager();
1161 }
1162
1163 void start() {
1164 rpcServer.start();
1165 }
1166
1167 void stop() {
1168 closeAllScanners();
1169 rpcServer.stop();
1170 }
1171
1172
1173
1174
1175
1176
1177 protected void checkOpen() throws IOException {
1178 if (regionServer.isAborted()) {
1179 throw new RegionServerAbortedException("Server " + regionServer.serverName + " aborting");
1180 }
1181 if (regionServer.isStopped()) {
1182 throw new RegionServerStoppedException("Server " + regionServer.serverName + " stopping");
1183 }
1184 if (!regionServer.fsOk) {
1185 throw new RegionServerStoppedException("File system not available");
1186 }
1187 if (!regionServer.isOnline()) {
1188 throw new ServerNotRunningYetException("Server is not running yet");
1189 }
1190 }
1191
1192
1193
1194
1195 protected List<BlockingServiceAndInterface> getServices() {
1196 List<BlockingServiceAndInterface> bssi = new ArrayList<BlockingServiceAndInterface>(2);
1197 bssi.add(new BlockingServiceAndInterface(
1198 ClientService.newReflectiveBlockingService(this),
1199 ClientService.BlockingInterface.class));
1200 bssi.add(new BlockingServiceAndInterface(
1201 AdminService.newReflectiveBlockingService(this),
1202 AdminService.BlockingInterface.class));
1203 return bssi;
1204 }
1205
1206 public InetSocketAddress getSocketAddress() {
1207 return isa;
1208 }
1209
1210 @Override
1211 public int getPriority(RequestHeader header, Message param, User user) {
1212 return priority.getPriority(header, param, user);
1213 }
1214
1215 @Override
1216 public long getDeadline(RequestHeader header, Message param) {
1217 return priority.getDeadline(header, param);
1218 }
1219
1220
1221
1222
1223
1224
1225
1226
1227 @Override
1228 public boolean checkOOME(final Throwable e) {
1229 return exitIfOOME(e);
1230 }
1231
1232 public static boolean exitIfOOME(final Throwable e ){
1233 boolean stop = false;
1234 try {
1235 if (e instanceof OutOfMemoryError
1236 || (e.getCause() != null && e.getCause() instanceof OutOfMemoryError)
1237 || (e.getMessage() != null && e.getMessage().contains(
1238 "java.lang.OutOfMemoryError"))) {
1239 stop = true;
1240 LOG.fatal("Run out of memory; " + RSRpcServices.class.getSimpleName()
1241 + " will abort itself immediately", e);
1242 }
1243 } finally {
1244 if (stop) {
1245 Runtime.getRuntime().halt(1);
1246 }
1247 }
1248 return stop;
1249 }
1250
1251
1252
1253
1254
1255
1256
1257
1258 @Override
1259 @QosPriority(priority=HConstants.ADMIN_QOS)
1260 public CloseRegionResponse closeRegion(final RpcController controller,
1261 final CloseRegionRequest request) throws ServiceException {
1262 final ServerName sn = (request.hasDestinationServer() ?
1263 ProtobufUtil.toServerName(request.getDestinationServer()) : null);
1264
1265 try {
1266 checkOpen();
1267 if (request.hasServerStartCode()) {
1268
1269 long serverStartCode = request.getServerStartCode();
1270 if (regionServer.serverName.getStartcode() != serverStartCode) {
1271 throw new ServiceException(new DoNotRetryIOException("This RPC was intended for a " +
1272 "different server with startCode: " + serverStartCode + ", this server is: "
1273 + regionServer.serverName));
1274 }
1275 }
1276 final String encodedRegionName = ProtobufUtil.getRegionEncodedName(request.getRegion());
1277
1278 requestCount.increment();
1279 LOG.info("Close " + encodedRegionName + ", moving to " + sn);
1280 CloseRegionCoordination.CloseRegionDetails crd = regionServer.getCoordinatedStateManager()
1281 .getCloseRegionCoordination().parseFromProtoRequest(request);
1282
1283 boolean closed = regionServer.closeRegion(encodedRegionName, false, crd, sn);
1284 CloseRegionResponse.Builder builder = CloseRegionResponse.newBuilder().setClosed(closed);
1285 return builder.build();
1286 } catch (IOException ie) {
1287 throw new ServiceException(ie);
1288 }
1289 }
1290
1291
1292
1293
1294
1295
1296
1297
1298 @Override
1299 @QosPriority(priority=HConstants.ADMIN_QOS)
1300 public CompactRegionResponse compactRegion(final RpcController controller,
1301 final CompactRegionRequest request) throws ServiceException {
1302 try {
1303 checkOpen();
1304 requestCount.increment();
1305 Region region = getRegion(request.getRegion());
1306 region.startRegionOperation(Operation.COMPACT_REGION);
1307 LOG.info("Compacting " + region.getRegionInfo().getRegionNameAsString());
1308 boolean major = false;
1309 byte [] family = null;
1310 Store store = null;
1311 if (request.hasFamily()) {
1312 family = request.getFamily().toByteArray();
1313 store = region.getStore(family);
1314 if (store == null) {
1315 throw new ServiceException(new IOException("column family " + Bytes.toString(family)
1316 + " does not exist in region " + region.getRegionInfo().getRegionNameAsString()));
1317 }
1318 }
1319 if (request.hasMajor()) {
1320 major = request.getMajor();
1321 }
1322 if (major) {
1323 if (family != null) {
1324 store.triggerMajorCompaction();
1325 } else {
1326 region.triggerMajorCompaction();
1327 }
1328 }
1329
1330 String familyLogMsg = (family != null)?" for column family: " + Bytes.toString(family):"";
1331 if (LOG.isTraceEnabled()) {
1332 LOG.trace("User-triggered compaction requested for region "
1333 + region.getRegionInfo().getRegionNameAsString() + familyLogMsg);
1334 }
1335 String log = "User-triggered " + (major ? "major " : "") + "compaction" + familyLogMsg;
1336 if(family != null) {
1337 regionServer.compactSplitThread.requestCompaction(region, store, log,
1338 Store.PRIORITY_USER, null, RpcServer.getRequestUser());
1339 } else {
1340 regionServer.compactSplitThread.requestCompaction(region, log,
1341 Store.PRIORITY_USER, null, RpcServer.getRequestUser());
1342 }
1343 return CompactRegionResponse.newBuilder().build();
1344 } catch (IOException ie) {
1345 throw new ServiceException(ie);
1346 }
1347 }
1348
1349
1350
1351
1352
1353
1354
1355
1356 @Override
1357 @QosPriority(priority=HConstants.ADMIN_QOS)
1358 public FlushRegionResponse flushRegion(final RpcController controller,
1359 final FlushRegionRequest request) throws ServiceException {
1360 try {
1361 checkOpen();
1362 requestCount.increment();
1363 Region region = getRegion(request.getRegion());
1364 LOG.info("Flushing " + region.getRegionInfo().getRegionNameAsString());
1365 boolean shouldFlush = true;
1366 if (request.hasIfOlderThanTs()) {
1367 shouldFlush = region.getEarliestFlushTimeForAllStores() < request.getIfOlderThanTs();
1368 }
1369 FlushRegionResponse.Builder builder = FlushRegionResponse.newBuilder();
1370 if (shouldFlush) {
1371 boolean writeFlushWalMarker = request.hasWriteFlushWalMarker() ?
1372 request.getWriteFlushWalMarker() : false;
1373
1374 HRegion.FlushResultImpl flushResult = (HRegion.FlushResultImpl)
1375 ((HRegion)region).flushcache(true, writeFlushWalMarker);
1376 boolean compactionNeeded = flushResult.isCompactionNeeded();
1377 if (compactionNeeded) {
1378 regionServer.compactSplitThread.requestSystemCompaction(region,
1379 "Compaction through user triggered flush");
1380 }
1381 builder.setFlushed(flushResult.isFlushSucceeded());
1382 builder.setWroteFlushWalMarker(flushResult.wroteFlushWalMarker);
1383 }
1384 builder.setLastFlushTime(region.getEarliestFlushTimeForAllStores());
1385 return builder.build();
1386 } catch (DroppedSnapshotException ex) {
1387
1388
1389
1390
1391 regionServer.abort("Replay of WAL required. Forcing server shutdown", ex);
1392 throw new ServiceException(ex);
1393 } catch (IOException ie) {
1394 throw new ServiceException(ie);
1395 }
1396 }
1397
1398 @Override
1399 @QosPriority(priority=HConstants.ADMIN_QOS)
1400 public GetOnlineRegionResponse getOnlineRegion(final RpcController controller,
1401 final GetOnlineRegionRequest request) throws ServiceException {
1402 try {
1403 checkOpen();
1404 requestCount.increment();
1405 Map<String, Region> onlineRegions = regionServer.onlineRegions;
1406 List<HRegionInfo> list = new ArrayList<HRegionInfo>(onlineRegions.size());
1407 for (Region region: onlineRegions.values()) {
1408 list.add(region.getRegionInfo());
1409 }
1410 Collections.sort(list);
1411 return ResponseConverter.buildGetOnlineRegionResponse(list);
1412 } catch (IOException ie) {
1413 throw new ServiceException(ie);
1414 }
1415 }
1416
1417 @Override
1418 @QosPriority(priority=HConstants.ADMIN_QOS)
1419 public GetRegionInfoResponse getRegionInfo(final RpcController controller,
1420 final GetRegionInfoRequest request) throws ServiceException {
1421 try {
1422 checkOpen();
1423 requestCount.increment();
1424 Region region = getRegion(request.getRegion());
1425 HRegionInfo info = region.getRegionInfo();
1426 GetRegionInfoResponse.Builder builder = GetRegionInfoResponse.newBuilder();
1427 builder.setRegionInfo(HRegionInfo.convert(info));
1428 if (request.hasCompactionState() && request.getCompactionState()) {
1429 builder.setCompactionState(region.getCompactionState());
1430 }
1431 builder.setIsRecovering(region.isRecovering());
1432 return builder.build();
1433 } catch (IOException ie) {
1434 throw new ServiceException(ie);
1435 }
1436 }
1437
1438
1439
1440
1441
1442
1443
1444
1445 @Override
1446 @QosPriority(priority=HConstants.ADMIN_QOS)
1447 public GetServerInfoResponse getServerInfo(final RpcController controller,
1448 final GetServerInfoRequest request) throws ServiceException {
1449 try {
1450 checkOpen();
1451 } catch (IOException ie) {
1452 throw new ServiceException(ie);
1453 }
1454 requestCount.increment();
1455 int infoPort = regionServer.infoServer != null ? regionServer.infoServer.getPort() : -1;
1456 return ResponseConverter.buildGetServerInfoResponse(regionServer.serverName, infoPort);
1457 }
1458
1459 @Override
1460 @QosPriority(priority=HConstants.ADMIN_QOS)
1461 public GetStoreFileResponse getStoreFile(final RpcController controller,
1462 final GetStoreFileRequest request) throws ServiceException {
1463 try {
1464 checkOpen();
1465 Region region = getRegion(request.getRegion());
1466 requestCount.increment();
1467 Set<byte[]> columnFamilies;
1468 if (request.getFamilyCount() == 0) {
1469 columnFamilies = region.getTableDesc().getFamiliesKeys();
1470 } else {
1471 columnFamilies = new TreeSet<byte[]>(Bytes.BYTES_RAWCOMPARATOR);
1472 for (ByteString cf: request.getFamilyList()) {
1473 columnFamilies.add(cf.toByteArray());
1474 }
1475 }
1476 int nCF = columnFamilies.size();
1477 List<String> fileList = region.getStoreFileList(
1478 columnFamilies.toArray(new byte[nCF][]));
1479 GetStoreFileResponse.Builder builder = GetStoreFileResponse.newBuilder();
1480 builder.addAllStoreFile(fileList);
1481 return builder.build();
1482 } catch (IOException ie) {
1483 throw new ServiceException(ie);
1484 }
1485 }
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495 @Override
1496 @QosPriority(priority = HConstants.ADMIN_QOS)
1497 public MergeRegionsResponse mergeRegions(final RpcController controller,
1498 final MergeRegionsRequest request) throws ServiceException {
1499 try {
1500 checkOpen();
1501 requestCount.increment();
1502 Region regionA = getRegion(request.getRegionA());
1503 Region regionB = getRegion(request.getRegionB());
1504 boolean forcible = request.getForcible();
1505 long masterSystemTime = request.hasMasterSystemTime() ? request.getMasterSystemTime() : -1;
1506 regionA.startRegionOperation(Operation.MERGE_REGION);
1507 regionB.startRegionOperation(Operation.MERGE_REGION);
1508 if (regionA.getRegionInfo().getReplicaId() != HRegionInfo.DEFAULT_REPLICA_ID ||
1509 regionB.getRegionInfo().getReplicaId() != HRegionInfo.DEFAULT_REPLICA_ID) {
1510 throw new ServiceException(new MergeRegionException("Can't merge non-default replicas"));
1511 }
1512 LOG.info("Receiving merging request for " + regionA + ", " + regionB
1513 + ",forcible=" + forcible);
1514 regionA.flush(true);
1515 regionB.flush(true);
1516 regionServer.compactSplitThread.requestRegionsMerge(regionA, regionB, forcible,
1517 masterSystemTime, RpcServer.getRequestUser());
1518 return MergeRegionsResponse.newBuilder().build();
1519 } catch (DroppedSnapshotException ex) {
1520 regionServer.abort("Replay of WAL required. Forcing server shutdown", ex);
1521 throw new ServiceException(ex);
1522 } catch (IOException ie) {
1523 throw new ServiceException(ie);
1524 }
1525 }
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550 @Override
1551 @QosPriority(priority=HConstants.ADMIN_QOS)
1552 public OpenRegionResponse openRegion(final RpcController controller,
1553 final OpenRegionRequest request) throws ServiceException {
1554 requestCount.increment();
1555 if (request.hasServerStartCode()) {
1556
1557 long serverStartCode = request.getServerStartCode();
1558 if (regionServer.serverName.getStartcode() != serverStartCode) {
1559 throw new ServiceException(new DoNotRetryIOException("This RPC was intended for a " +
1560 "different server with startCode: " + serverStartCode + ", this server is: "
1561 + regionServer.serverName));
1562 }
1563 }
1564
1565 OpenRegionResponse.Builder builder = OpenRegionResponse.newBuilder();
1566 final int regionCount = request.getOpenInfoCount();
1567 final Map<TableName, HTableDescriptor> htds =
1568 new HashMap<TableName, HTableDescriptor>(regionCount);
1569 final boolean isBulkAssign = regionCount > 1;
1570 try {
1571 checkOpen();
1572 } catch (IOException ie) {
1573 TableName tableName = null;
1574 if (regionCount == 1) {
1575 RegionInfo ri = request.getOpenInfo(0).getRegion();
1576 if (ri != null) {
1577 tableName = ProtobufUtil.toTableName(ri.getTableName());
1578 }
1579 }
1580 if (!TableName.META_TABLE_NAME.equals(tableName)) {
1581 throw new ServiceException(ie);
1582 }
1583
1584 int timeout = regionServer.conf.getInt(HConstants.HBASE_RPC_TIMEOUT_KEY,
1585 HConstants.DEFAULT_HBASE_RPC_TIMEOUT) >> 2;
1586 long endTime = System.currentTimeMillis() + timeout;
1587 synchronized (regionServer.online) {
1588 try {
1589 while (System.currentTimeMillis() <= endTime
1590 && !regionServer.isStopped() && !regionServer.isOnline()) {
1591 regionServer.online.wait(regionServer.msgInterval);
1592 }
1593 checkOpen();
1594 } catch (InterruptedException t) {
1595 Thread.currentThread().interrupt();
1596 throw new ServiceException(t);
1597 } catch (IOException e) {
1598 throw new ServiceException(e);
1599 }
1600 }
1601 }
1602
1603 long masterSystemTime = request.hasMasterSystemTime() ? request.getMasterSystemTime() : -1;
1604
1605 for (RegionOpenInfo regionOpenInfo : request.getOpenInfoList()) {
1606 final HRegionInfo region = HRegionInfo.convert(regionOpenInfo.getRegion());
1607 OpenRegionCoordination coordination = regionServer.getCoordinatedStateManager().
1608 getOpenRegionCoordination();
1609 OpenRegionCoordination.OpenRegionDetails ord =
1610 coordination.parseFromProtoRequest(regionOpenInfo);
1611
1612 HTableDescriptor htd;
1613 try {
1614 final Region onlineRegion = regionServer.getFromOnlineRegions(region.getEncodedName());
1615 if (onlineRegion != null) {
1616
1617 if (onlineRegion.getCoprocessorHost() != null) {
1618 onlineRegion.getCoprocessorHost().preOpen();
1619 }
1620
1621
1622 Pair<HRegionInfo, ServerName> p = MetaTableAccessor.getRegion(
1623 regionServer.getConnection(), region.getRegionName());
1624 if (regionServer.serverName.equals(p.getSecond())) {
1625 Boolean closing = regionServer.regionsInTransitionInRS.get(region.getEncodedNameAsBytes());
1626
1627
1628
1629
1630
1631 if (!Boolean.FALSE.equals(closing)
1632 && regionServer.getFromOnlineRegions(region.getEncodedName()) != null) {
1633 LOG.warn("Attempted open of " + region.getEncodedName()
1634 + " but already online on this server");
1635 builder.addOpeningState(RegionOpeningState.ALREADY_OPENED);
1636 continue;
1637 }
1638 } else {
1639 LOG.warn("The region " + region.getEncodedName() + " is online on this server"
1640 + " but hbase:meta does not have this server - continue opening.");
1641 regionServer.removeFromOnlineRegions(onlineRegion, null);
1642 }
1643 }
1644 LOG.info("Open " + region.getRegionNameAsString());
1645 htd = htds.get(region.getTable());
1646 if (htd == null) {
1647 htd = regionServer.tableDescriptors.get(region.getTable());
1648 if (htd == null) {
1649 throw new IOException("missing table descriptor for " + region.getEncodedName());
1650 }
1651 htds.put(region.getTable(), htd);
1652 }
1653
1654 final Boolean previous = regionServer.regionsInTransitionInRS.putIfAbsent(
1655 region.getEncodedNameAsBytes(), Boolean.TRUE);
1656
1657 if (Boolean.FALSE.equals(previous)) {
1658
1659
1660 coordination.tryTransitionFromOfflineToFailedOpen(regionServer, region, ord);
1661
1662 throw new RegionAlreadyInTransitionException("Received OPEN for the region:"
1663 + region.getRegionNameAsString() + " , which we are already trying to CLOSE ");
1664 }
1665
1666 if (Boolean.TRUE.equals(previous)) {
1667
1668 LOG.info("Receiving OPEN for the region:" +
1669 region.getRegionNameAsString() + " , which we are already trying to OPEN"
1670 + " - ignoring this new request for this region.");
1671 }
1672
1673
1674
1675 regionServer.removeFromMovedRegions(region.getEncodedName());
1676
1677 if (previous == null) {
1678
1679 if (ZKSplitLog.isRegionMarkedRecoveringInZK(regionServer.getZooKeeper(),
1680 region.getEncodedName())) {
1681
1682
1683 if (!regionOpenInfo.hasOpenForDistributedLogReplay()
1684 || regionOpenInfo.getOpenForDistributedLogReplay()) {
1685 regionServer.recoveringRegions.put(region.getEncodedName(), null);
1686 } else {
1687
1688
1689 List<String> tmpRegions = new ArrayList<String>();
1690 tmpRegions.add(region.getEncodedName());
1691 ZKSplitLog.deleteRecoveringRegionZNodes(regionServer.getZooKeeper(),
1692 tmpRegions);
1693 }
1694 }
1695
1696
1697 if (region.isMetaRegion()) {
1698 regionServer.service.submit(new OpenMetaHandler(
1699 regionServer, regionServer, region, htd, masterSystemTime, coordination, ord));
1700 } else {
1701 regionServer.updateRegionFavoredNodesMapping(region.getEncodedName(),
1702 regionOpenInfo.getFavoredNodesList());
1703 if (htd.getPriority() >= HConstants.ADMIN_QOS || region.getTable().isSystemTable()) {
1704 regionServer.service.submit(new OpenPriorityRegionHandler(
1705 regionServer, regionServer, region, htd, masterSystemTime, coordination, ord));
1706 } else {
1707 regionServer.service.submit(new OpenRegionHandler(
1708 regionServer, regionServer, region, htd, masterSystemTime, coordination, ord));
1709 }
1710 }
1711 }
1712
1713 builder.addOpeningState(RegionOpeningState.OPENED);
1714
1715 } catch (KeeperException zooKeeperEx) {
1716 LOG.error("Can't retrieve recovering state from zookeeper", zooKeeperEx);
1717 throw new ServiceException(zooKeeperEx);
1718 } catch (IOException ie) {
1719 LOG.warn("Failed opening region " + region.getRegionNameAsString(), ie);
1720 if (isBulkAssign) {
1721 builder.addOpeningState(RegionOpeningState.FAILED_OPENING);
1722 } else {
1723 throw new ServiceException(ie);
1724 }
1725 }
1726 }
1727 return builder.build();
1728 }
1729
1730
1731
1732
1733
1734
1735
1736
1737
1738
1739
1740
1741 @Override
1742 public WarmupRegionResponse warmupRegion(final RpcController controller,
1743 final WarmupRegionRequest request) throws ServiceException {
1744
1745 RegionInfo regionInfo = request.getRegionInfo();
1746 final HRegionInfo region = HRegionInfo.convert(regionInfo);
1747 HTableDescriptor htd;
1748 WarmupRegionResponse response = WarmupRegionResponse.getDefaultInstance();
1749
1750 try {
1751 checkOpen();
1752 String encodedName = region.getEncodedName();
1753 byte[] encodedNameBytes = region.getEncodedNameAsBytes();
1754 final Region onlineRegion = regionServer.getFromOnlineRegions(encodedName);
1755
1756 if (onlineRegion != null) {
1757 LOG.info("Region already online. Skipping warming up " + region);
1758 return response;
1759 }
1760
1761 if (LOG.isDebugEnabled()) {
1762 LOG.debug("Warming up Region " + region.getRegionNameAsString());
1763 }
1764
1765 htd = regionServer.tableDescriptors.get(region.getTable());
1766
1767 if (regionServer.getRegionsInTransitionInRS().containsKey(encodedNameBytes)) {
1768 LOG.info("Region is in transition. Skipping warmup " + region);
1769 return response;
1770 }
1771
1772 HRegion.warmupHRegion(region, htd, regionServer.getWAL(region),
1773 regionServer.getConfiguration(), regionServer, null);
1774
1775 } catch (IOException ie) {
1776 LOG.error("Failed warming up region " + region.getRegionNameAsString(), ie);
1777 throw new ServiceException(ie);
1778 }
1779
1780 return response;
1781 }
1782
1783
1784
1785
1786
1787
1788
1789
1790
1791 @Override
1792 @QosPriority(priority = HConstants.REPLAY_QOS)
1793 public ReplicateWALEntryResponse replay(final RpcController controller,
1794 final ReplicateWALEntryRequest request) throws ServiceException {
1795 long before = EnvironmentEdgeManager.currentTime();
1796 CellScanner cells = ((PayloadCarryingRpcController) controller).cellScanner();
1797 try {
1798 checkOpen();
1799 List<WALEntry> entries = request.getEntryList();
1800 if (entries == null || entries.isEmpty()) {
1801
1802 return ReplicateWALEntryResponse.newBuilder().build();
1803 }
1804 ByteString regionName = entries.get(0).getKey().getEncodedRegionName();
1805 Region region = regionServer.getRegionByEncodedName(regionName.toStringUtf8());
1806 RegionCoprocessorHost coprocessorHost =
1807 ServerRegionReplicaUtil.isDefaultReplica(region.getRegionInfo())
1808 ? region.getCoprocessorHost()
1809 : null;
1810 List<Pair<WALKey, WALEdit>> walEntries = new ArrayList<Pair<WALKey, WALEdit>>();
1811
1812
1813 boolean isPrimary = RegionReplicaUtil.isDefaultReplica(region.getRegionInfo());
1814 Durability durability = isPrimary ? Durability.USE_DEFAULT : Durability.SKIP_WAL;
1815
1816 for (WALEntry entry : entries) {
1817 if (!regionName.equals(entry.getKey().getEncodedRegionName())) {
1818 throw new NotServingRegionException("Replay request contains entries from multiple " +
1819 "regions. First region:" + regionName.toStringUtf8() + " , other region:"
1820 + entry.getKey().getEncodedRegionName());
1821 }
1822 if (regionServer.nonceManager != null && isPrimary) {
1823 long nonceGroup = entry.getKey().hasNonceGroup()
1824 ? entry.getKey().getNonceGroup() : HConstants.NO_NONCE;
1825 long nonce = entry.getKey().hasNonce() ? entry.getKey().getNonce() : HConstants.NO_NONCE;
1826 regionServer.nonceManager.reportOperationFromWal(
1827 nonceGroup,
1828 nonce,
1829 entry.getKey().getWriteTime());
1830 }
1831 Pair<WALKey, WALEdit> walEntry = (coprocessorHost == null) ? null :
1832 new Pair<WALKey, WALEdit>();
1833 List<WALSplitter.MutationReplay> edits = WALSplitter.getMutationsFromWALEntry(entry,
1834 cells, walEntry, durability);
1835 if (coprocessorHost != null) {
1836
1837
1838 if (coprocessorHost.preWALRestore(region.getRegionInfo(), walEntry.getFirst(),
1839 walEntry.getSecond())) {
1840
1841 continue;
1842 }
1843 walEntries.add(walEntry);
1844 }
1845 if(edits!=null && !edits.isEmpty()) {
1846 long replaySeqId = (entry.getKey().hasOrigSequenceNumber()) ?
1847 entry.getKey().getOrigSequenceNumber() : entry.getKey().getLogSequenceNumber();
1848 OperationStatus[] result = doReplayBatchOp(region, edits, replaySeqId);
1849
1850 for (int i = 0; result != null && i < result.length; i++) {
1851 if (result[i] != OperationStatus.SUCCESS) {
1852 throw new IOException(result[i].getExceptionMsg());
1853 }
1854 }
1855 }
1856 }
1857
1858
1859 WAL wal = getWAL(region);
1860 if (wal != null) {
1861 wal.sync();
1862 }
1863
1864 if (coprocessorHost != null) {
1865 for (Pair<WALKey, WALEdit> entry : walEntries) {
1866 coprocessorHost.postWALRestore(region.getRegionInfo(), entry.getFirst(),
1867 entry.getSecond());
1868 }
1869 }
1870 return ReplicateWALEntryResponse.newBuilder().build();
1871 } catch (IOException ie) {
1872 throw new ServiceException(ie);
1873 } finally {
1874 if (regionServer.metricsRegionServer != null) {
1875 regionServer.metricsRegionServer.updateReplay(
1876 EnvironmentEdgeManager.currentTime() - before);
1877 }
1878 }
1879 }
1880
1881 WAL getWAL(Region region) {
1882 return ((HRegion)region).getWAL();
1883 }
1884
1885
1886
1887
1888
1889
1890
1891
1892 @Override
1893 @QosPriority(priority=HConstants.REPLICATION_QOS)
1894 public ReplicateWALEntryResponse replicateWALEntry(final RpcController controller,
1895 final ReplicateWALEntryRequest request) throws ServiceException {
1896 try {
1897 checkOpen();
1898 if (regionServer.replicationSinkHandler != null) {
1899 requestCount.increment();
1900 List<WALEntry> entries = request.getEntryList();
1901 CellScanner cellScanner = ((PayloadCarryingRpcController)controller).cellScanner();
1902 regionServer.getRegionServerCoprocessorHost().preReplicateLogEntries(entries, cellScanner);
1903 regionServer.replicationSinkHandler.replicateLogEntries(entries, cellScanner,
1904 request.getReplicationClusterId(), request.getSourceBaseNamespaceDirPath(),
1905 request.getSourceHFileArchiveDirPath());
1906 regionServer.getRegionServerCoprocessorHost().postReplicateLogEntries(entries, cellScanner);
1907 return ReplicateWALEntryResponse.newBuilder().build();
1908 } else {
1909 throw new ServiceException("Replication services are not initialized yet");
1910 }
1911 } catch (IOException ie) {
1912 throw new ServiceException(ie);
1913 }
1914 }
1915
1916
1917
1918
1919
1920
1921
1922 @Override
1923 public RollWALWriterResponse rollWALWriter(final RpcController controller,
1924 final RollWALWriterRequest request) throws ServiceException {
1925 try {
1926 checkOpen();
1927 requestCount.increment();
1928 regionServer.getRegionServerCoprocessorHost().preRollWALWriterRequest();
1929 regionServer.walRoller.requestRollAll();
1930 regionServer.getRegionServerCoprocessorHost().postRollWALWriterRequest();
1931 RollWALWriterResponse.Builder builder = RollWALWriterResponse.newBuilder();
1932 return builder.build();
1933 } catch (IOException ie) {
1934 throw new ServiceException(ie);
1935 }
1936 }
1937
1938
1939
1940
1941
1942
1943
1944
1945 @Override
1946 @QosPriority(priority=HConstants.ADMIN_QOS)
1947 public SplitRegionResponse splitRegion(final RpcController controller,
1948 final SplitRegionRequest request) throws ServiceException {
1949 try {
1950 checkOpen();
1951 requestCount.increment();
1952 Region region = getRegion(request.getRegion());
1953 region.startRegionOperation(Operation.SPLIT_REGION);
1954 if (region.getRegionInfo().getReplicaId() != HRegionInfo.DEFAULT_REPLICA_ID) {
1955 throw new IOException("Can't split replicas directly. "
1956 + "Replicas are auto-split when their primary is split.");
1957 }
1958 LOG.info("Splitting " + region.getRegionInfo().getRegionNameAsString());
1959 region.flush(true);
1960 byte[] splitPoint = null;
1961 if (request.hasSplitPoint()) {
1962 splitPoint = request.getSplitPoint().toByteArray();
1963 }
1964 ((HRegion)region).forceSplit(splitPoint);
1965 regionServer.compactSplitThread.requestSplit(region, ((HRegion)region).checkSplit(),
1966 RpcServer.getRequestUser());
1967 return SplitRegionResponse.newBuilder().build();
1968 } catch (DroppedSnapshotException ex) {
1969 regionServer.abort("Replay of WAL required. Forcing server shutdown", ex);
1970 throw new ServiceException(ex);
1971 } catch (IOException ie) {
1972 throw new ServiceException(ie);
1973 }
1974 }
1975
1976
1977
1978
1979
1980
1981
1982
1983 @Override
1984 @QosPriority(priority=HConstants.ADMIN_QOS)
1985 public StopServerResponse stopServer(final RpcController controller,
1986 final StopServerRequest request) throws ServiceException {
1987 requestCount.increment();
1988 String reason = request.getReason();
1989 regionServer.stop(reason);
1990 return StopServerResponse.newBuilder().build();
1991 }
1992
1993 @Override
1994 public UpdateFavoredNodesResponse updateFavoredNodes(RpcController controller,
1995 UpdateFavoredNodesRequest request) throws ServiceException {
1996 List<UpdateFavoredNodesRequest.RegionUpdateInfo> openInfoList = request.getUpdateInfoList();
1997 UpdateFavoredNodesResponse.Builder respBuilder = UpdateFavoredNodesResponse.newBuilder();
1998 for (UpdateFavoredNodesRequest.RegionUpdateInfo regionUpdateInfo : openInfoList) {
1999 HRegionInfo hri = HRegionInfo.convert(regionUpdateInfo.getRegion());
2000 regionServer.updateRegionFavoredNodesMapping(hri.getEncodedName(),
2001 regionUpdateInfo.getFavoredNodesList());
2002 }
2003 respBuilder.setResponse(openInfoList.size());
2004 return respBuilder.build();
2005 }
2006
2007
2008
2009
2010
2011
2012 @Override
2013 public BulkLoadHFileResponse bulkLoadHFile(final RpcController controller,
2014 final BulkLoadHFileRequest request) throws ServiceException {
2015 try {
2016 checkOpen();
2017 requestCount.increment();
2018 Region region = getRegion(request.getRegion());
2019 List<Pair<byte[], String>> familyPaths = new ArrayList<Pair<byte[], String>>();
2020 for (FamilyPath familyPath: request.getFamilyPathList()) {
2021 familyPaths.add(new Pair<byte[], String>(familyPath.getFamily().toByteArray(),
2022 familyPath.getPath()));
2023 }
2024 boolean bypass = false;
2025 if (region.getCoprocessorHost() != null) {
2026 bypass = region.getCoprocessorHost().preBulkLoadHFile(familyPaths);
2027 }
2028 boolean loaded = false;
2029 if (!bypass) {
2030 loaded = region.bulkLoadHFiles(familyPaths, request.getAssignSeqNum(), null);
2031 }
2032 if (region.getCoprocessorHost() != null) {
2033 loaded = region.getCoprocessorHost().postBulkLoadHFile(familyPaths, loaded);
2034 }
2035 BulkLoadHFileResponse.Builder builder = BulkLoadHFileResponse.newBuilder();
2036 builder.setLoaded(loaded);
2037 return builder.build();
2038 } catch (IOException ie) {
2039 throw new ServiceException(ie);
2040 }
2041 }
2042
2043 @Override
2044 public CoprocessorServiceResponse execService(final RpcController controller,
2045 final CoprocessorServiceRequest request) throws ServiceException {
2046 try {
2047 checkOpen();
2048 requestCount.increment();
2049 Region region = getRegion(request.getRegion());
2050 Message result = execServiceOnRegion(region, request.getCall());
2051 CoprocessorServiceResponse.Builder builder =
2052 CoprocessorServiceResponse.newBuilder();
2053 builder.setRegion(RequestConverter.buildRegionSpecifier(
2054 RegionSpecifierType.REGION_NAME, region.getRegionInfo().getRegionName()));
2055 builder.setValue(
2056 builder.getValueBuilder().setName(result.getClass().getName())
2057 .setValue(result.toByteString()));
2058 return builder.build();
2059 } catch (IOException ie) {
2060 throw new ServiceException(ie);
2061 }
2062 }
2063
2064 private Message execServiceOnRegion(Region region,
2065 final ClientProtos.CoprocessorServiceCall serviceCall) throws IOException {
2066
2067 ServerRpcController execController = new ServerRpcController();
2068 return region.execService(execController, serviceCall);
2069 }
2070
2071
2072
2073
2074
2075
2076
2077
2078 @Override
2079 public GetResponse get(final RpcController controller,
2080 final GetRequest request) throws ServiceException {
2081 long before = EnvironmentEdgeManager.currentTime();
2082 OperationQuota quota = null;
2083 try {
2084 checkOpen();
2085 requestCount.increment();
2086 rpcGetRequestCount.increment();
2087 Region region = getRegion(request.getRegion());
2088
2089 GetResponse.Builder builder = GetResponse.newBuilder();
2090 ClientProtos.Get get = request.getGet();
2091 Boolean existence = null;
2092 Result r = null;
2093 quota = getQuotaManager().checkQuota(region, OperationQuota.OperationType.GET);
2094
2095 if (get.hasClosestRowBefore() && get.getClosestRowBefore()) {
2096 if (get.getColumnCount() != 1) {
2097 throw new DoNotRetryIOException(
2098 "get ClosestRowBefore supports one and only one family now, not "
2099 + get.getColumnCount() + " families");
2100 }
2101 byte[] row = get.getRow().toByteArray();
2102 byte[] family = get.getColumn(0).getFamily().toByteArray();
2103 r = region.getClosestRowBefore(row, family);
2104 } else {
2105 Get clientGet = ProtobufUtil.toGet(get);
2106 if (get.getExistenceOnly() && region.getCoprocessorHost() != null) {
2107 existence = region.getCoprocessorHost().preExists(clientGet);
2108 }
2109 if (existence == null) {
2110 r = region.get(clientGet);
2111 if (get.getExistenceOnly()) {
2112 boolean exists = r.getExists();
2113 if (region.getCoprocessorHost() != null) {
2114 exists = region.getCoprocessorHost().postExists(clientGet, exists);
2115 }
2116 existence = exists;
2117 }
2118 }
2119 }
2120 if (existence != null){
2121 ClientProtos.Result pbr =
2122 ProtobufUtil.toResult(existence, region.getRegionInfo().getReplicaId() != 0);
2123 builder.setResult(pbr);
2124 } else if (r != null) {
2125 ClientProtos.Result pbr;
2126 RpcCallContext call = RpcServer.getCurrentCall();
2127 if (isClientCellBlockSupport(call) && controller instanceof PayloadCarryingRpcController
2128 && VersionInfoUtil.hasMinimumVersion(call.getClientVersionInfo(), 1, 3)) {
2129 pbr = ProtobufUtil.toResultNoData(r);
2130 ((PayloadCarryingRpcController) controller)
2131 .setCellScanner(CellUtil.createCellScanner(r.rawCells()));
2132 addSize(call, r, null);
2133 } else {
2134 pbr = ProtobufUtil.toResult(r);
2135 }
2136 builder.setResult(pbr);
2137 }
2138 if (r != null) {
2139 quota.addGetResult(r);
2140 }
2141 return builder.build();
2142 } catch (IOException ie) {
2143 throw new ServiceException(ie);
2144 } finally {
2145 if (regionServer.metricsRegionServer != null) {
2146 regionServer.metricsRegionServer.updateGet(
2147 EnvironmentEdgeManager.currentTime() - before);
2148 }
2149 if (quota != null) {
2150 quota.close();
2151 }
2152 }
2153 }
2154
2155
2156
2157
2158
2159
2160
2161
2162 @Override
2163 public MultiResponse multi(final RpcController rpcc, final MultiRequest request)
2164 throws ServiceException {
2165 try {
2166 checkOpen();
2167 } catch (IOException ie) {
2168 throw new ServiceException(ie);
2169 }
2170
2171
2172
2173 PayloadCarryingRpcController controller = (PayloadCarryingRpcController)rpcc;
2174 CellScanner cellScanner = controller != null ? controller.cellScanner(): null;
2175 if (controller != null) {
2176 controller.setCellScanner(null);
2177 }
2178
2179 long nonceGroup = request.hasNonceGroup() ? request.getNonceGroup() : HConstants.NO_NONCE;
2180
2181
2182 List<CellScannable> cellsToReturn = null;
2183 MultiResponse.Builder responseBuilder = MultiResponse.newBuilder();
2184 RegionActionResult.Builder regionActionResultBuilder = RegionActionResult.newBuilder();
2185 Boolean processed = null;
2186
2187 this.rpcMultiRequestCount.increment();
2188 Map<RegionSpecifier, ClientProtos.RegionLoadStats> regionStats = new HashMap<>(request
2189 .getRegionActionCount());
2190 for (RegionAction regionAction : request.getRegionActionList()) {
2191 this.requestCount.add(regionAction.getActionCount());
2192 OperationQuota quota;
2193 Region region;
2194 regionActionResultBuilder.clear();
2195 RegionSpecifier regionSpecifier = regionAction.getRegion();
2196 try {
2197 region = getRegion(regionSpecifier);
2198 quota = getQuotaManager().checkQuota(region, regionAction.getActionList());
2199 } catch (IOException e) {
2200 rpcServer.getMetrics().exception(e);
2201 regionActionResultBuilder.setException(ResponseConverter.buildException(e));
2202 responseBuilder.addRegionActionResult(regionActionResultBuilder.build());
2203
2204
2205
2206 if (cellScanner != null) {
2207 skipCellsForMutations(regionAction.getActionList(), cellScanner);
2208 }
2209 continue;
2210 }
2211
2212 if (regionAction.hasAtomic() && regionAction.getAtomic()) {
2213
2214
2215 try {
2216 if (request.hasCondition()) {
2217 Condition condition = request.getCondition();
2218 byte[] row = condition.getRow().toByteArray();
2219 byte[] family = condition.getFamily().toByteArray();
2220 byte[] qualifier = condition.getQualifier().toByteArray();
2221 CompareOp compareOp = CompareOp.valueOf(condition.getCompareType().name());
2222 ByteArrayComparable comparator =
2223 ProtobufUtil.toComparator(condition.getComparator());
2224 processed = checkAndRowMutate(region, regionAction.getActionList(),
2225 cellScanner, row, family, qualifier, compareOp,
2226 comparator, regionActionResultBuilder);
2227 } else {
2228 mutateRows(region, regionAction.getActionList(), cellScanner,
2229 regionActionResultBuilder);
2230 processed = Boolean.TRUE;
2231 }
2232 } catch (IOException e) {
2233 rpcServer.getMetrics().exception(e);
2234
2235 regionActionResultBuilder.setException(ResponseConverter.buildException(e));
2236 }
2237 } else {
2238
2239 cellsToReturn = doNonAtomicRegionMutation(region, quota, regionAction, cellScanner,
2240 regionActionResultBuilder, cellsToReturn, nonceGroup);
2241 }
2242 responseBuilder.addRegionActionResult(regionActionResultBuilder.build());
2243 quota.close();
2244 ClientProtos.RegionLoadStats regionLoadStats = ((HRegion)region).getLoadStatistics();
2245 if(regionLoadStats != null) {
2246 regionStats.put(regionSpecifier, regionLoadStats);
2247 }
2248 }
2249
2250 if (cellsToReturn != null && !cellsToReturn.isEmpty() && controller != null) {
2251 controller.setCellScanner(CellUtil.createCellScanner(cellsToReturn));
2252 }
2253
2254 if (processed != null) {
2255 responseBuilder.setProcessed(processed);
2256 }
2257
2258 MultiRegionLoadStats.Builder builder = MultiRegionLoadStats.newBuilder();
2259 for(Entry<RegionSpecifier, ClientProtos.RegionLoadStats> stat: regionStats.entrySet()){
2260 builder.addRegion(stat.getKey());
2261 builder.addStat(stat.getValue());
2262 }
2263 responseBuilder.setRegionStatistics(builder);
2264 return responseBuilder.build();
2265 }
2266
2267 private void skipCellsForMutations(List<Action> actions, CellScanner cellScanner) {
2268 for (Action action : actions) {
2269 skipCellsForMutation(action, cellScanner);
2270 }
2271 }
2272
2273 private void skipCellsForMutation(Action action, CellScanner cellScanner) {
2274 try {
2275 if (action.hasMutation()) {
2276 MutationProto m = action.getMutation();
2277 if (m.hasAssociatedCellCount()) {
2278 for (int i = 0; i < m.getAssociatedCellCount(); i++) {
2279 cellScanner.advance();
2280 }
2281 }
2282 }
2283 } catch (IOException e) {
2284
2285
2286
2287 LOG.error("Error while skipping Cells in CellScanner for invalid Region Mutations", e);
2288 }
2289 }
2290
2291
2292
2293
2294
2295
2296
2297
2298 @Override
2299 public MutateResponse mutate(final RpcController rpcc,
2300 final MutateRequest request) throws ServiceException {
2301
2302
2303 PayloadCarryingRpcController controller = (PayloadCarryingRpcController)rpcc;
2304 CellScanner cellScanner = controller != null ? controller.cellScanner() : null;
2305 OperationQuota quota = null;
2306 RpcCallContext context = RpcServer.getCurrentCall();
2307
2308 if (controller != null) {
2309 controller.setCellScanner(null);
2310 }
2311 try {
2312 checkOpen();
2313 requestCount.increment();
2314 rpcMutateRequestCount.increment();
2315 Region region = getRegion(request.getRegion());
2316 MutateResponse.Builder builder = MutateResponse.newBuilder();
2317 MutationProto mutation = request.getMutation();
2318 if (!region.getRegionInfo().isMetaTable()) {
2319 regionServer.cacheFlusher.reclaimMemStoreMemory();
2320 }
2321 long nonceGroup = request.hasNonceGroup() ? request.getNonceGroup() : HConstants.NO_NONCE;
2322 Result r = null;
2323 Boolean processed = null;
2324 MutationType type = mutation.getMutateType();
2325 long mutationSize = 0;
2326 quota = getQuotaManager().checkQuota(region, OperationQuota.OperationType.MUTATE);
2327 switch (type) {
2328 case APPEND:
2329
2330 r = append(region, quota, mutation, cellScanner, nonceGroup);
2331 break;
2332 case INCREMENT:
2333
2334 r = increment(region, quota, mutation, cellScanner, nonceGroup);
2335 break;
2336 case PUT:
2337 Put put = ProtobufUtil.toPut(mutation, cellScanner);
2338 quota.addMutation(put);
2339 if (request.hasCondition()) {
2340 Condition condition = request.getCondition();
2341 byte[] row = condition.getRow().toByteArray();
2342 byte[] family = condition.getFamily().toByteArray();
2343 byte[] qualifier = condition.getQualifier().toByteArray();
2344 CompareOp compareOp = CompareOp.valueOf(condition.getCompareType().name());
2345 ByteArrayComparable comparator =
2346 ProtobufUtil.toComparator(condition.getComparator());
2347 if (region.getCoprocessorHost() != null) {
2348 processed = region.getCoprocessorHost().preCheckAndPut(
2349 row, family, qualifier, compareOp, comparator, put);
2350 }
2351 if (processed == null) {
2352 boolean result = region.checkAndMutate(row, family,
2353 qualifier, compareOp, comparator, put, true);
2354 if (region.getCoprocessorHost() != null) {
2355 result = region.getCoprocessorHost().postCheckAndPut(row, family,
2356 qualifier, compareOp, comparator, put, result);
2357 }
2358 processed = result;
2359 }
2360 } else {
2361 region.put(put);
2362 processed = Boolean.TRUE;
2363 }
2364 break;
2365 case DELETE:
2366 Delete delete = ProtobufUtil.toDelete(mutation, cellScanner);
2367 quota.addMutation(delete);
2368 if (request.hasCondition()) {
2369 Condition condition = request.getCondition();
2370 byte[] row = condition.getRow().toByteArray();
2371 byte[] family = condition.getFamily().toByteArray();
2372 byte[] qualifier = condition.getQualifier().toByteArray();
2373 CompareOp compareOp = CompareOp.valueOf(condition.getCompareType().name());
2374 ByteArrayComparable comparator =
2375 ProtobufUtil.toComparator(condition.getComparator());
2376 if (region.getCoprocessorHost() != null) {
2377 processed = region.getCoprocessorHost().preCheckAndDelete(
2378 row, family, qualifier, compareOp, comparator, delete);
2379 }
2380 if (processed == null) {
2381 boolean result = region.checkAndMutate(row, family,
2382 qualifier, compareOp, comparator, delete, true);
2383 if (region.getCoprocessorHost() != null) {
2384 result = region.getCoprocessorHost().postCheckAndDelete(row, family,
2385 qualifier, compareOp, comparator, delete, result);
2386 }
2387 processed = result;
2388 }
2389 } else {
2390 region.delete(delete);
2391 processed = Boolean.TRUE;
2392 }
2393 break;
2394 default:
2395 throw new DoNotRetryIOException(
2396 "Unsupported mutate type: " + type.name());
2397 }
2398 if (processed != null) builder.setProcessed(processed.booleanValue());
2399 addResult(builder, r, controller);
2400 boolean clientCellBlockSupported = isClientCellBlockSupport(context);
2401 if (clientCellBlockSupported) {
2402 addSize(context, r, null);
2403 }
2404 return builder.build();
2405 } catch (IOException ie) {
2406 regionServer.checkFileSystem();
2407 throw new ServiceException(ie);
2408 } finally {
2409 if (quota != null) {
2410 quota.close();
2411 }
2412 }
2413 }
2414
2415
2416
2417
2418 @Deprecated
2419 private static final IOException SCANNER_ALREADY_CLOSED = new IOException() {
2420
2421 private static final long serialVersionUID = -4305297078988180130L;
2422
2423 @Override
2424 public Throwable fillInStackTrace() {
2425 return this;
2426 }
2427 };
2428
2429 private RegionScannerHolder getRegionScanner(ScanRequest request) throws IOException {
2430 String scannerName = Long.toString(request.getScannerId());
2431 RegionScannerHolder rsh = scanners.get(scannerName);
2432 if (rsh == null) {
2433
2434 if (request.hasCloseScanner() && request.getCloseScanner()) {
2435 throw SCANNER_ALREADY_CLOSED;
2436 } else {
2437 LOG.warn("Client tried to access missing scanner " + scannerName);
2438 throw new UnknownScannerException(
2439 "Unknown scanner '" + scannerName + "'. This can happen due to any of the following "
2440 + "reasons: a) Scanner id given is wrong, b) Scanner lease expired because of "
2441 + "long wait between consecutive client checkins, c) Server may be closing down, "
2442 + "d) RegionServer restart during upgrade.\nIf the issue is due to reason (b), a "
2443 + "possible fix would be increasing the value of"
2444 + "'hbase.client.scanner.timeout.period' configuration.");
2445 }
2446 }
2447 HRegionInfo hri = rsh.s.getRegionInfo();
2448
2449 if (regionServer.getOnlineRegion(hri.getRegionName()) != rsh.r) {
2450 String msg = "Region was re-opened after the scanner" + scannerName + " was created: "
2451 + hri.getRegionNameAsString();
2452 LOG.warn(msg + ", closing...");
2453 scanners.remove(scannerName);
2454 try {
2455 rsh.s.close();
2456 } catch (IOException e) {
2457 LOG.warn("Getting exception closing " + scannerName, e);
2458 } finally {
2459 try {
2460 regionServer.leases.cancelLease(scannerName);
2461 } catch (LeaseException e) {
2462 LOG.warn("Getting exception closing " + scannerName, e);
2463 }
2464 }
2465 throw new NotServingRegionException(msg);
2466 }
2467 return rsh;
2468 }
2469
2470 private Pair<RegionScannerHolder, Boolean> newRegionScanner(ScanRequest request,
2471 ScanResponse.Builder builder) throws IOException {
2472 Region region = getRegion(request.getRegion());
2473 ClientProtos.Scan protoScan = request.getScan();
2474 boolean isLoadingCfsOnDemandSet = protoScan.hasLoadColumnFamiliesOnDemand();
2475 Scan scan = ProtobufUtil.toScan(protoScan);
2476
2477 if (!isLoadingCfsOnDemandSet) {
2478 scan.setLoadColumnFamiliesOnDemand(region.isLoadingCfsOnDemandDefault());
2479 }
2480 if (!scan.hasFamilies()) {
2481
2482 for (byte[] family : region.getTableDesc().getFamiliesKeys()) {
2483 scan.addFamily(family);
2484 }
2485 }
2486 RegionScanner scanner = null;
2487 if (region.getCoprocessorHost() != null) {
2488 scanner = region.getCoprocessorHost().preScannerOpen(scan);
2489 }
2490 if (scanner == null) {
2491 scanner = region.getScanner(scan);
2492 }
2493 if (region.getCoprocessorHost() != null) {
2494 scanner = region.getCoprocessorHost().postScannerOpen(scan, scanner);
2495 }
2496 long scannerId = this.scannerIdGen.incrementAndGet();
2497 builder.setScannerId(scannerId);
2498 builder.setTtl(scannerLeaseTimeoutPeriod);
2499 String scannerName = String.valueOf(scannerId);
2500 return Pair.newPair(addScanner(scannerName, scanner, region), scan.isSmall());
2501 }
2502
2503 private void checkScanNextCallSeq(ScanRequest request, RegionScannerHolder rsh)
2504 throws OutOfOrderScannerNextException {
2505
2506
2507
2508 if (request.hasNextCallSeq()) {
2509 long callSeq = request.getNextCallSeq();
2510 if (!rsh.incNextCallSeq(callSeq)) {
2511 throw new OutOfOrderScannerNextException("Expected nextCallSeq: " + rsh.getNextCallSeq()
2512 + " But the nextCallSeq got from client: " + request.getNextCallSeq() + "; request="
2513 + TextFormat.shortDebugString(request));
2514 }
2515 }
2516 }
2517
2518 private void addScannerLeaseBack(Leases.Lease lease) {
2519 try {
2520 regionServer.leases.addLease(lease);
2521 } catch (LeaseStillHeldException e) {
2522
2523 throw new AssertionError(e);
2524 }
2525 }
2526
2527 private long getTimeLimit(PayloadCarryingRpcController controller,
2528 boolean allowHeartbeatMessages) {
2529
2530
2531
2532 if (allowHeartbeatMessages && (scannerLeaseTimeoutPeriod > 0 || rpcTimeout > 0)) {
2533 long timeLimitDelta;
2534 if (scannerLeaseTimeoutPeriod > 0 && rpcTimeout > 0) {
2535 timeLimitDelta = Math.min(scannerLeaseTimeoutPeriod, rpcTimeout);
2536 } else {
2537 timeLimitDelta = scannerLeaseTimeoutPeriod > 0 ? scannerLeaseTimeoutPeriod : rpcTimeout;
2538 }
2539 if (controller != null && controller.getCallTimeout() > 0) {
2540 timeLimitDelta = Math.min(timeLimitDelta, controller.getCallTimeout());
2541 }
2542
2543
2544
2545 timeLimitDelta = Math.max(timeLimitDelta / 2, minimumScanTimeLimitDelta);
2546
2547
2548 return System.currentTimeMillis() + timeLimitDelta;
2549 }
2550
2551
2552 return -1L;
2553 }
2554
2555
2556 private boolean scan(PayloadCarryingRpcController controller, ScanRequest request,
2557 RegionScannerHolder rsh, boolean isSmallScan, long maxQuotaResultSize, int rows,
2558 List<Result> results, ScanResponse.Builder builder, MutableObject lastBlock,
2559 RpcCallContext context) throws IOException {
2560 Region region = rsh.r;
2561 RegionScanner scanner = rsh.s;
2562 long maxResultSize;
2563 if (scanner.getMaxResultSize() > 0) {
2564 maxResultSize = Math.min(scanner.getMaxResultSize(), maxQuotaResultSize);
2565 } else {
2566 maxResultSize = maxQuotaResultSize;
2567 }
2568
2569
2570
2571 List<Cell> values = new ArrayList<Cell>(32);
2572 region.startRegionOperation(Operation.SCAN);
2573 try {
2574 int i = 0;
2575 long before = EnvironmentEdgeManager.currentTime();
2576 synchronized (scanner) {
2577 boolean stale = (region.getRegionInfo().getReplicaId() != 0);
2578 boolean clientHandlesPartials =
2579 request.hasClientHandlesPartials() && request.getClientHandlesPartials();
2580 boolean clientHandlesHeartbeats =
2581 request.hasClientHandlesHeartbeats() && request.getClientHandlesHeartbeats();
2582
2583
2584
2585
2586
2587
2588 boolean serverGuaranteesOrderOfPartials = results.isEmpty();
2589 boolean allowPartialResults =
2590 clientHandlesPartials && serverGuaranteesOrderOfPartials && !isSmallScan;
2591 boolean moreRows = false;
2592
2593
2594
2595
2596
2597
2598
2599
2600 boolean allowHeartbeatMessages = clientHandlesHeartbeats && allowPartialResults;
2601
2602 long timeLimit = getTimeLimit(controller, allowHeartbeatMessages);
2603
2604 final LimitScope sizeScope =
2605 allowPartialResults ? LimitScope.BETWEEN_CELLS : LimitScope.BETWEEN_ROWS;
2606 final LimitScope timeScope =
2607 allowHeartbeatMessages ? LimitScope.BETWEEN_CELLS : LimitScope.BETWEEN_ROWS;
2608
2609 boolean trackMetrics = request.hasTrackScanMetrics() && request.getTrackScanMetrics();
2610
2611
2612
2613 ScannerContext.Builder contextBuilder = ScannerContext.newBuilder(true);
2614 contextBuilder.setSizeLimit(sizeScope, maxResultSize);
2615 contextBuilder.setBatchLimit(scanner.getBatch());
2616 contextBuilder.setTimeLimit(timeScope, timeLimit);
2617 contextBuilder.setTrackMetrics(trackMetrics);
2618 ScannerContext scannerContext = contextBuilder.build();
2619 boolean limitReached = false;
2620 while (i < rows) {
2621
2622
2623
2624
2625
2626 scannerContext.setBatchProgress(0);
2627
2628
2629 moreRows = scanner.nextRaw(values, scannerContext);
2630
2631 if (!values.isEmpty()) {
2632 final boolean partial = scannerContext.partialResultFormed();
2633 Result r = Result.create(values, null, stale, partial);
2634 lastBlock.setValue(addSize(context, r, lastBlock.getValue()));
2635 results.add(r);
2636 i++;
2637 }
2638
2639 boolean sizeLimitReached = scannerContext.checkSizeLimit(LimitScope.BETWEEN_ROWS);
2640 boolean timeLimitReached = scannerContext.checkTimeLimit(LimitScope.BETWEEN_ROWS);
2641 boolean rowLimitReached = i >= rows;
2642 limitReached = sizeLimitReached || timeLimitReached || rowLimitReached;
2643
2644 if (limitReached || !moreRows) {
2645 if (LOG.isTraceEnabled()) {
2646 LOG.trace("Done scanning. limitReached: " + limitReached + " moreRows: " + moreRows
2647 + " scannerContext: " + scannerContext);
2648 }
2649
2650
2651
2652
2653 if (moreRows) {
2654
2655 builder.setHeartbeatMessage(timeLimitReached);
2656 }
2657 break;
2658 }
2659 values.clear();
2660 }
2661 if (limitReached || moreRows) {
2662
2663 builder.setMoreResultsInRegion(true);
2664 } else {
2665
2666 builder.setMoreResultsInRegion(false);
2667 }
2668
2669
2670
2671 if (trackMetrics) {
2672 Map<String, Long> metrics = scannerContext.getMetrics().getMetricsMap();
2673 ScanMetrics.Builder metricBuilder = ScanMetrics.newBuilder();
2674 NameInt64Pair.Builder pairBuilder = NameInt64Pair.newBuilder();
2675
2676 for (Entry<String, Long> entry : metrics.entrySet()) {
2677 pairBuilder.setName(entry.getKey());
2678 pairBuilder.setValue(entry.getValue());
2679 metricBuilder.addMetrics(pairBuilder.build());
2680 }
2681
2682 builder.setScanMetrics(metricBuilder.build());
2683 }
2684 }
2685 region.updateReadRequestsCount(i);
2686 long end = EnvironmentEdgeManager.currentTime();
2687 long responseCellSize = context != null ? context.getResponseCellSize() : 0;
2688 region.getMetrics().updateScanTime(end - before);
2689 if (regionServer.metricsRegionServer != null) {
2690 regionServer.metricsRegionServer.updateScanSize(responseCellSize);
2691 regionServer.metricsRegionServer.updateScanTime(end - before);
2692 }
2693 } finally {
2694 region.closeRegionOperation();
2695 }
2696
2697 if (region.getCoprocessorHost() != null) {
2698 region.getCoprocessorHost().postScannerNext(scanner, results, rows, true);
2699 }
2700 return builder.getMoreResultsInRegion();
2701 }
2702
2703
2704
2705
2706
2707
2708
2709
2710 @Override
2711 public ScanResponse scan(final RpcController controller, final ScanRequest request)
2712 throws ServiceException {
2713 if (controller != null && !(controller instanceof PayloadCarryingRpcController)) {
2714 throw new UnsupportedOperationException(
2715 "We only do PayloadCarryingRpcControllers! FIX IF A PROBLEM: " + controller);
2716 }
2717 if (!request.hasScannerId() && !request.hasScan()) {
2718 throw new ServiceException(
2719 new DoNotRetryIOException("Missing required input: scannerId or scan"));
2720 }
2721 try {
2722 checkOpen();
2723 } catch (IOException e) {
2724 if (request.hasScannerId()) {
2725 String scannerName = Long.toString(request.getScannerId());
2726 if (LOG.isDebugEnabled()) {
2727 LOG.debug(
2728 "Server shutting down and client tried to access missing scanner " + scannerName);
2729 }
2730 if (regionServer.leases != null) {
2731 try {
2732 regionServer.leases.cancelLease(scannerName);
2733 } catch (LeaseException le) {
2734
2735 if (LOG.isTraceEnabled()) {
2736 LOG.trace("Un-able to cancel lease of scanner. It could already be closed.");
2737 }
2738 }
2739 }
2740 }
2741 throw new ServiceException(e);
2742 }
2743 requestCount.increment();
2744 rpcScanRequestCount.increment();
2745 RegionScannerHolder rsh;
2746 ScanResponse.Builder builder = ScanResponse.newBuilder();
2747 boolean isSmallScan;
2748 try {
2749 if (request.hasScannerId()) {
2750 rsh = getRegionScanner(request);
2751 isSmallScan = false;
2752 } else {
2753 Pair<RegionScannerHolder, Boolean> pair = newRegionScanner(request, builder);
2754 rsh = pair.getFirst();
2755 isSmallScan = pair.getSecond().booleanValue();
2756 }
2757 } catch (IOException e) {
2758 if (e == SCANNER_ALREADY_CLOSED) {
2759
2760
2761 return builder.build();
2762 }
2763 throw new ServiceException(e);
2764 }
2765 Region region = rsh.r;
2766 String scannerName = rsh.scannerName;
2767 Leases.Lease lease;
2768 try {
2769
2770
2771 lease = regionServer.leases.removeLease(scannerName);
2772 } catch (LeaseException e) {
2773 throw new ServiceException(e);
2774 }
2775 if (request.hasRenew() && request.getRenew()) {
2776
2777 addScannerLeaseBack(lease);
2778 try {
2779 checkScanNextCallSeq(request, rsh);
2780 } catch (OutOfOrderScannerNextException e) {
2781 throw new ServiceException(e);
2782 }
2783 return builder.build();
2784 }
2785 OperationQuota quota;
2786 try {
2787 quota = getQuotaManager().checkQuota(region, OperationQuota.OperationType.SCAN);
2788 } catch (IOException e) {
2789 addScannerLeaseBack(lease);
2790 throw new ServiceException(e);
2791 };
2792 try {
2793 checkScanNextCallSeq(request, rsh);
2794 } catch (OutOfOrderScannerNextException e) {
2795 addScannerLeaseBack(lease);
2796 throw new ServiceException(e);
2797 }
2798
2799
2800
2801 boolean closeScanner = request.hasCloseScanner() ? request.getCloseScanner() : false;
2802 int rows;
2803 if (request.hasNumberOfRows()) {
2804 rows = request.getNumberOfRows();
2805 } else {
2806 rows = closeScanner ? 0 : 1;
2807 }
2808 RpcCallContext context = RpcServer.getCurrentCall();
2809
2810 long maxQuotaResultSize = Math.min(maxScannerResultSize, quota.getReadAvailable());
2811 RegionScanner scanner = rsh.s;
2812 boolean moreResults = true;
2813 boolean moreResultsInRegion = true;
2814 MutableObject lastBlock = new MutableObject();
2815 boolean scannerClosed = false;
2816 try {
2817 List<Result> results = new ArrayList<>();
2818 if (rows > 0) {
2819 boolean done = false;
2820
2821 if (region.getCoprocessorHost() != null) {
2822 Boolean bypass = region.getCoprocessorHost().preScannerNext(scanner, results, rows);
2823 if (!results.isEmpty()) {
2824 for (Result r : results) {
2825 lastBlock.setValue(addSize(context, r, lastBlock.getValue()));
2826 }
2827 }
2828 if (bypass != null && bypass.booleanValue()) {
2829 done = true;
2830 }
2831 }
2832 if (!done) {
2833 moreResultsInRegion = scan((PayloadCarryingRpcController) controller, request, rsh,
2834 isSmallScan, maxQuotaResultSize, rows, results, builder, lastBlock, context);
2835 }
2836 }
2837
2838 quota.addScanResult(results);
2839
2840 if (scanner.isFilterDone() && results.isEmpty()) {
2841
2842
2843
2844
2845 moreResults = false;
2846 }
2847 addResults(builder, results, (PayloadCarryingRpcController) controller,
2848 RegionReplicaUtil.isDefaultReplica(region.getRegionInfo()));
2849 if (!moreResults || !moreResultsInRegion || closeScanner) {
2850 scannerClosed = true;
2851 closeScanner(region, scanner, scannerName, context);
2852 }
2853 builder.setMoreResults(moreResults);
2854 return builder.build();
2855 } catch (Exception e) {
2856 try {
2857
2858 scannerClosed = true;
2859
2860
2861
2862 closeScanner(region, scanner, scannerName, context);
2863
2864
2865
2866
2867
2868
2869
2870
2871
2872
2873
2874
2875 if (e instanceof DoNotRetryIOException) {
2876 throw e;
2877 }
2878
2879
2880
2881 if (e instanceof FileNotFoundException) {
2882 throw new DoNotRetryIOException(e);
2883 }
2884
2885
2886
2887
2888 if (VersionInfoUtil.hasMinimumVersion(context.getClientVersionInfo(), 1, 4)) {
2889
2890 throw new ScannerResetException("Scanner is closed on the server-side", e);
2891 } else {
2892
2893 throw new UnknownScannerException("Throwing UnknownScannerException to reset the client"
2894 + " scanner state for clients older than 1.3.", e);
2895 }
2896 } catch (IOException ioe) {
2897 throw new ServiceException(ioe);
2898 }
2899 } finally {
2900 if (!scannerClosed) {
2901
2902 addScannerLeaseBack(lease);
2903 }
2904 quota.close();
2905 }
2906 }
2907
2908 private void closeScanner(Region region, RegionScanner scanner, String scannerName,
2909 RpcCallContext context) throws IOException {
2910 if (region.getCoprocessorHost() != null) {
2911 if (region.getCoprocessorHost().preScannerClose(scanner)) {
2912
2913 return;
2914 }
2915 }
2916 RegionScannerHolder rsh = scanners.remove(scannerName);
2917 if (rsh != null) {
2918 rsh.s.close();
2919 if (region.getCoprocessorHost() != null) {
2920 region.getCoprocessorHost().postScannerClose(scanner);
2921 }
2922 }
2923 }
2924
2925 @Override
2926 public CoprocessorServiceResponse execRegionServerService(RpcController controller,
2927 CoprocessorServiceRequest request) throws ServiceException {
2928 return regionServer.execRegionServerService(controller, request);
2929 }
2930
2931 @Override
2932 public UpdateConfigurationResponse updateConfiguration(
2933 RpcController controller, UpdateConfigurationRequest request)
2934 throws ServiceException {
2935 try {
2936 this.regionServer.updateConfiguration();
2937 } catch (Exception e) {
2938 throw new ServiceException(e);
2939 }
2940 return UpdateConfigurationResponse.getDefaultInstance();
2941 }
2942 }