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