1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.client;
20
21 import java.io.IOException;
22 import java.io.InterruptedIOException;
23 import java.util.ArrayList;
24 import java.util.Collections;
25 import java.util.List;
26 import java.util.Map;
27 import java.util.NavigableMap;
28 import java.util.TreeMap;
29 import java.util.concurrent.Callable;
30 import java.util.concurrent.ExecutionException;
31 import java.util.concurrent.ExecutorService;
32 import java.util.concurrent.Future;
33 import java.util.concurrent.SynchronousQueue;
34 import java.util.concurrent.ThreadPoolExecutor;
35 import java.util.concurrent.TimeUnit;
36
37 import org.apache.commons.logging.Log;
38 import org.apache.commons.logging.LogFactory;
39 import org.apache.hadoop.hbase.classification.InterfaceAudience;
40 import org.apache.hadoop.hbase.classification.InterfaceStability;
41 import org.apache.hadoop.conf.Configuration;
42 import org.apache.hadoop.hbase.Cell;
43 import org.apache.hadoop.hbase.HBaseConfiguration;
44 import org.apache.hadoop.hbase.DoNotRetryIOException;
45 import org.apache.hadoop.hbase.HConstants;
46 import org.apache.hadoop.hbase.HRegionInfo;
47 import org.apache.hadoop.hbase.HRegionLocation;
48 import org.apache.hadoop.hbase.HTableDescriptor;
49 import org.apache.hadoop.hbase.KeyValueUtil;
50 import org.apache.hadoop.hbase.ServerName;
51 import org.apache.hadoop.hbase.TableName;
52 import org.apache.hadoop.hbase.client.AsyncProcess.AsyncRequestFuture;
53 import org.apache.hadoop.hbase.client.coprocessor.Batch;
54 import org.apache.hadoop.hbase.client.coprocessor.Batch.Callback;
55 import org.apache.hadoop.hbase.filter.BinaryComparator;
56 import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
57 import org.apache.hadoop.hbase.ipc.CoprocessorRpcChannel;
58 import org.apache.hadoop.hbase.ipc.PayloadCarryingRpcController;
59 import org.apache.hadoop.hbase.ipc.RegionCoprocessorRpcChannel;
60 import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
61 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
62 import org.apache.hadoop.hbase.protobuf.RequestConverter;
63 import org.apache.hadoop.hbase.protobuf.ResponseConverter;
64 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
65 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MultiRequest;
66 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutateRequest;
67 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutateResponse;
68 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.RegionAction;
69 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.CompareType;
70 import org.apache.hadoop.hbase.util.Bytes;
71 import org.apache.hadoop.hbase.util.Pair;
72 import org.apache.hadoop.hbase.util.ReflectionUtils;
73 import org.apache.hadoop.hbase.util.Threads;
74
75 import com.google.common.annotations.VisibleForTesting;
76 import com.google.protobuf.Descriptors;
77 import com.google.protobuf.Message;
78 import com.google.protobuf.Service;
79 import com.google.protobuf.ServiceException;
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113 @InterfaceAudience.Private
114 @InterfaceStability.Stable
115 public class HTable implements HTableInterface, RegionLocator {
116 private static final Log LOG = LogFactory.getLog(HTable.class);
117 protected ClusterConnection connection;
118 private final TableName tableName;
119 private volatile Configuration configuration;
120 private ConnectionConfiguration connConfiguration;
121 protected BufferedMutatorImpl mutator;
122 private boolean autoFlush = true;
123 private boolean closed = false;
124 protected int scannerCaching;
125 protected long scannerMaxResultSize;
126 private ExecutorService pool;
127 private int operationTimeout;
128 private int rpcTimeout;
129 private final boolean cleanupPoolOnClose;
130 private final boolean cleanupConnectionOnClose;
131 private Consistency defaultConsistency = Consistency.STRONG;
132 private HRegionLocator locator;
133
134
135 protected AsyncProcess multiAp;
136 private RpcRetryingCallerFactory rpcCallerFactory;
137 private RpcControllerFactory rpcControllerFactory;
138
139
140
141
142
143
144
145
146
147 @Deprecated
148 public HTable(Configuration conf, final String tableName)
149 throws IOException {
150 this(conf, TableName.valueOf(tableName));
151 }
152
153
154
155
156
157
158
159
160
161 @Deprecated
162 public HTable(Configuration conf, final byte[] tableName)
163 throws IOException {
164 this(conf, TableName.valueOf(tableName));
165 }
166
167
168
169
170
171
172
173
174
175 @Deprecated
176 public HTable(Configuration conf, final TableName tableName)
177 throws IOException {
178 this.tableName = tableName;
179 this.cleanupPoolOnClose = this.cleanupConnectionOnClose = true;
180 if (conf == null) {
181 this.connection = null;
182 return;
183 }
184 this.connection = ConnectionManager.getConnectionInternal(conf);
185 this.configuration = conf;
186
187 this.pool = getDefaultExecutor(conf);
188 this.finishSetup();
189 }
190
191
192
193
194
195
196
197
198 @Deprecated
199 public HTable(TableName tableName, Connection connection) throws IOException {
200 this.tableName = tableName;
201 this.cleanupPoolOnClose = true;
202 this.cleanupConnectionOnClose = false;
203 this.connection = (ClusterConnection)connection;
204 this.configuration = connection.getConfiguration();
205
206 this.pool = getDefaultExecutor(this.configuration);
207 this.finishSetup();
208 }
209
210
211 @InterfaceAudience.Private
212 public static ThreadPoolExecutor getDefaultExecutor(Configuration conf) {
213 int maxThreads = conf.getInt("hbase.htable.threads.max", Integer.MAX_VALUE);
214 if (maxThreads == 0) {
215 maxThreads = 1;
216 }
217 long keepAliveTime = conf.getLong("hbase.htable.threads.keepalivetime", 60);
218
219
220
221
222
223 ThreadPoolExecutor pool = new ThreadPoolExecutor(1, maxThreads, keepAliveTime, TimeUnit.SECONDS,
224 new SynchronousQueue<Runnable>(), Threads.newDaemonThreadFactory("htable"));
225 pool.allowCoreThreadTimeOut(true);
226 return pool;
227 }
228
229
230
231
232
233
234
235
236
237
238 @Deprecated
239 public HTable(Configuration conf, final byte[] tableName, final ExecutorService pool)
240 throws IOException {
241 this(conf, TableName.valueOf(tableName), pool);
242 }
243
244
245
246
247
248
249
250
251
252
253 @Deprecated
254 public HTable(Configuration conf, final TableName tableName, final ExecutorService pool)
255 throws IOException {
256 this.connection = ConnectionManager.getConnectionInternal(conf);
257 this.configuration = conf;
258 this.pool = pool;
259 if (pool == null) {
260 this.pool = getDefaultExecutor(conf);
261 this.cleanupPoolOnClose = true;
262 } else {
263 this.cleanupPoolOnClose = false;
264 }
265 this.tableName = tableName;
266 this.cleanupConnectionOnClose = true;
267 this.finishSetup();
268 }
269
270
271
272
273
274
275
276
277
278 @Deprecated
279 public HTable(final byte[] tableName, final Connection connection,
280 final ExecutorService pool) throws IOException {
281 this(TableName.valueOf(tableName), connection, pool);
282 }
283
284
285 @Deprecated
286 public HTable(TableName tableName, final Connection connection,
287 final ExecutorService pool) throws IOException {
288 this(tableName, (ClusterConnection)connection, null, null, null, pool);
289 }
290
291
292
293
294
295
296
297
298
299
300 @InterfaceAudience.Private
301 public HTable(TableName tableName, final ClusterConnection connection,
302 final ConnectionConfiguration tableConfig,
303 final RpcRetryingCallerFactory rpcCallerFactory,
304 final RpcControllerFactory rpcControllerFactory,
305 final ExecutorService pool) throws IOException {
306 if (connection == null || connection.isClosed()) {
307 throw new IllegalArgumentException("Connection is null or closed.");
308 }
309 this.tableName = tableName;
310 this.cleanupConnectionOnClose = false;
311 this.connection = connection;
312 this.configuration = connection.getConfiguration();
313 this.connConfiguration = tableConfig;
314 this.pool = pool;
315 if (pool == null) {
316 this.pool = getDefaultExecutor(this.configuration);
317 this.cleanupPoolOnClose = true;
318 } else {
319 this.cleanupPoolOnClose = false;
320 }
321
322 this.rpcCallerFactory = rpcCallerFactory;
323 this.rpcControllerFactory = rpcControllerFactory;
324
325 this.finishSetup();
326 }
327
328
329
330
331
332 @VisibleForTesting
333 protected HTable(ClusterConnection conn, BufferedMutatorParams params) throws IOException {
334 connection = conn;
335 tableName = params.getTableName();
336 connConfiguration = new ConnectionConfiguration(connection.getConfiguration());
337 cleanupPoolOnClose = false;
338 cleanupConnectionOnClose = false;
339
340 this.mutator = new BufferedMutatorImpl(conn, null, null, params);
341 }
342
343
344
345
346 public static int getMaxKeyValueSize(Configuration conf) {
347 return conf.getInt("hbase.client.keyvalue.maxsize", -1);
348 }
349
350
351
352
353 private void finishSetup() throws IOException {
354 if (connConfiguration == null) {
355 connConfiguration = new ConnectionConfiguration(configuration);
356 }
357 this.operationTimeout = tableName.isSystemTable() ?
358 connConfiguration.getMetaOperationTimeout() : connConfiguration.getOperationTimeout();
359 this.rpcTimeout = configuration.getInt(HConstants.HBASE_RPC_TIMEOUT_KEY,
360 HConstants.DEFAULT_HBASE_RPC_TIMEOUT);
361 this.scannerCaching = connConfiguration.getScannerCaching();
362 this.scannerMaxResultSize = connConfiguration.getScannerMaxResultSize();
363 if (this.rpcCallerFactory == null) {
364 this.rpcCallerFactory = connection.getNewRpcRetryingCallerFactory(configuration);
365 }
366 if (this.rpcControllerFactory == null) {
367 this.rpcControllerFactory = RpcControllerFactory.instantiate(configuration);
368 }
369
370
371 multiAp = this.connection.getAsyncProcess();
372
373 this.closed = false;
374
375 this.locator = new HRegionLocator(tableName, connection);
376 }
377
378
379
380
381 @Override
382 public Configuration getConfiguration() {
383 return configuration;
384 }
385
386
387
388
389
390
391
392
393
394
395 @Deprecated
396 public static boolean isTableEnabled(String tableName) throws IOException {
397 return isTableEnabled(TableName.valueOf(tableName));
398 }
399
400
401
402
403
404
405
406
407
408
409 @Deprecated
410 public static boolean isTableEnabled(byte[] tableName) throws IOException {
411 return isTableEnabled(TableName.valueOf(tableName));
412 }
413
414
415
416
417
418
419
420
421
422
423 @Deprecated
424 public static boolean isTableEnabled(TableName tableName) throws IOException {
425 return isTableEnabled(HBaseConfiguration.create(), tableName);
426 }
427
428
429
430
431
432
433
434
435
436 @Deprecated
437 public static boolean isTableEnabled(Configuration conf, String tableName)
438 throws IOException {
439 return isTableEnabled(conf, TableName.valueOf(tableName));
440 }
441
442
443
444
445
446
447
448
449
450 @Deprecated
451 public static boolean isTableEnabled(Configuration conf, byte[] tableName)
452 throws IOException {
453 return isTableEnabled(conf, TableName.valueOf(tableName));
454 }
455
456
457
458
459
460
461
462
463
464 @Deprecated
465 public static boolean isTableEnabled(Configuration conf,
466 final TableName tableName) throws IOException {
467 return HConnectionManager.execute(new HConnectable<Boolean>(conf) {
468 @Override
469 public Boolean connect(HConnection connection) throws IOException {
470 return connection.isTableEnabled(tableName);
471 }
472 });
473 }
474
475
476
477
478
479
480
481
482 @Deprecated
483 public HRegionLocation getRegionLocation(final String row)
484 throws IOException {
485 return getRegionLocation(Bytes.toBytes(row), false);
486 }
487
488
489
490
491 @Override
492 @Deprecated
493 public HRegionLocation getRegionLocation(final byte [] row)
494 throws IOException {
495 return locator.getRegionLocation(row);
496 }
497
498
499
500
501 @Override
502 @Deprecated
503 public HRegionLocation getRegionLocation(final byte [] row, boolean reload)
504 throws IOException {
505 return locator.getRegionLocation(row, reload);
506 }
507
508
509
510
511 @Override
512 public byte [] getTableName() {
513 return this.tableName.getName();
514 }
515
516 @Override
517 public TableName getName() {
518 return tableName;
519 }
520
521
522
523
524
525
526
527
528 @Deprecated
529 @VisibleForTesting
530 public HConnection getConnection() {
531 return this.connection;
532 }
533
534
535
536
537
538
539
540 @Deprecated
541 public int getScannerCaching() {
542 return scannerCaching;
543 }
544
545
546
547
548
549 @Deprecated
550 public List<Row> getWriteBuffer() {
551 return mutator == null ? null : mutator.getWriteBuffer();
552 }
553
554
555
556
557
558
559
560
561
562
563
564
565 @Deprecated
566 public void setScannerCaching(int scannerCaching) {
567 this.scannerCaching = scannerCaching;
568 }
569
570
571
572
573 @Override
574 public HTableDescriptor getTableDescriptor() throws IOException {
575 HTableDescriptor htd = HBaseAdmin.getTableDescriptor(tableName, connection, rpcCallerFactory,
576 rpcControllerFactory, operationTimeout, rpcTimeout);
577 if (htd != null) {
578 return new UnmodifyableHTableDescriptor(htd);
579 }
580 return null;
581 }
582
583
584
585
586
587 @Override
588 @Deprecated
589 public byte [][] getStartKeys() throws IOException {
590 return locator.getStartKeys();
591 }
592
593
594
595
596
597 @Override
598 @Deprecated
599 public byte[][] getEndKeys() throws IOException {
600 return locator.getEndKeys();
601 }
602
603
604
605
606
607 @Override
608 @Deprecated
609 public Pair<byte[][],byte[][]> getStartEndKeys() throws IOException {
610 return locator.getStartEndKeys();
611 }
612
613
614
615
616
617
618
619
620
621 @Deprecated
622 public NavigableMap<HRegionInfo, ServerName> getRegionLocations() throws IOException {
623
624 return MetaScanner.allTableRegions(this.connection, getName());
625 }
626
627
628
629
630
631
632
633
634
635
636 @Override
637 @Deprecated
638 public List<HRegionLocation> getAllRegionLocations() throws IOException {
639 return locator.getAllRegionLocations();
640 }
641
642
643
644
645
646
647
648
649
650
651
652 @Deprecated
653 public List<HRegionLocation> getRegionsInRange(final byte [] startKey,
654 final byte [] endKey) throws IOException {
655 return getRegionsInRange(startKey, endKey, false);
656 }
657
658
659
660
661
662
663
664
665
666
667
668
669 @Deprecated
670 public List<HRegionLocation> getRegionsInRange(final byte [] startKey,
671 final byte [] endKey, final boolean reload) throws IOException {
672 return getKeysAndRegionsInRange(startKey, endKey, false, reload).getSecond();
673 }
674
675
676
677
678
679
680
681
682
683
684
685
686
687 @Deprecated
688 private Pair<List<byte[]>, List<HRegionLocation>> getKeysAndRegionsInRange(
689 final byte[] startKey, final byte[] endKey, final boolean includeEndKey)
690 throws IOException {
691 return getKeysAndRegionsInRange(startKey, endKey, includeEndKey, false);
692 }
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707 @Deprecated
708 private Pair<List<byte[]>, List<HRegionLocation>> getKeysAndRegionsInRange(
709 final byte[] startKey, final byte[] endKey, final boolean includeEndKey,
710 final boolean reload) throws IOException {
711 final boolean endKeyIsEndOfTable = Bytes.equals(endKey,HConstants.EMPTY_END_ROW);
712 if ((Bytes.compareTo(startKey, endKey) > 0) && !endKeyIsEndOfTable) {
713 throw new IllegalArgumentException(
714 "Invalid range: " + Bytes.toStringBinary(startKey) +
715 " > " + Bytes.toStringBinary(endKey));
716 }
717 List<byte[]> keysInRange = new ArrayList<byte[]>();
718 List<HRegionLocation> regionsInRange = new ArrayList<HRegionLocation>();
719 byte[] currentKey = startKey;
720 do {
721 HRegionLocation regionLocation = getRegionLocation(currentKey, reload);
722 keysInRange.add(currentKey);
723 regionsInRange.add(regionLocation);
724 currentKey = regionLocation.getRegionInfo().getEndKey();
725 } while (!Bytes.equals(currentKey, HConstants.EMPTY_END_ROW)
726 && (endKeyIsEndOfTable || Bytes.compareTo(currentKey, endKey) < 0
727 || (includeEndKey && Bytes.compareTo(currentKey, endKey) == 0)));
728 return new Pair<List<byte[]>, List<HRegionLocation>>(keysInRange,
729 regionsInRange);
730 }
731
732
733
734
735
736 @Override
737 @Deprecated
738 public Result getRowOrBefore(final byte[] row, final byte[] family)
739 throws IOException {
740 RegionServerCallable<Result> callable = new RegionServerCallable<Result>(this.connection,
741 tableName, row) {
742 @Override
743 public Result call(int callTimeout) throws IOException {
744 PayloadCarryingRpcController controller = rpcControllerFactory.newController();
745 controller.setPriority(tableName);
746 controller.setCallTimeout(callTimeout);
747 ClientProtos.GetRequest request = RequestConverter.buildGetRowOrBeforeRequest(
748 getLocation().getRegionInfo().getRegionName(), row, family);
749 try {
750 ClientProtos.GetResponse response = getStub().get(controller, request);
751 if (!response.hasResult()) return null;
752 return ProtobufUtil.toResult(response.getResult(), controller.cellScanner());
753 } catch (ServiceException se) {
754 throw ProtobufUtil.getRemoteException(se);
755 }
756 }
757 };
758 return rpcCallerFactory.<Result>newCaller(rpcTimeout).callWithRetries(callable,
759 this.operationTimeout);
760 }
761
762
763
764
765
766 @Override
767 public ResultScanner getScanner(final Scan scan) throws IOException {
768 if (scan.getBatch() > 0 && scan.isSmall()) {
769 throw new IllegalArgumentException("Small scan should not be used with batching");
770 }
771
772 if (scan.getCaching() <= 0) {
773 scan.setCaching(getScannerCaching());
774 }
775 if (scan.getMaxResultSize() <= 0) {
776 scan.setMaxResultSize(scannerMaxResultSize);
777 }
778
779 if (scan.isReversed()) {
780 if (scan.isSmall()) {
781 return new ClientSmallReversedScanner(getConfiguration(), scan, getName(),
782 this.connection, this.rpcCallerFactory, this.rpcControllerFactory,
783 pool, connConfiguration.getReplicaCallTimeoutMicroSecondScan());
784 } else {
785 return new ReversedClientScanner(getConfiguration(), scan, getName(),
786 this.connection, this.rpcCallerFactory, this.rpcControllerFactory,
787 pool, connConfiguration.getReplicaCallTimeoutMicroSecondScan());
788 }
789 }
790
791 if (scan.isSmall()) {
792 return new ClientSmallScanner(getConfiguration(), scan, getName(),
793 this.connection, this.rpcCallerFactory, this.rpcControllerFactory,
794 pool, connConfiguration.getReplicaCallTimeoutMicroSecondScan());
795 } else {
796 return new ClientScanner(getConfiguration(), scan, getName(), this.connection,
797 this.rpcCallerFactory, this.rpcControllerFactory,
798 pool, connConfiguration.getReplicaCallTimeoutMicroSecondScan());
799 }
800 }
801
802
803
804
805
806 @Override
807 public ResultScanner getScanner(byte [] family) throws IOException {
808 Scan scan = new Scan();
809 scan.addFamily(family);
810 return getScanner(scan);
811 }
812
813
814
815
816
817 @Override
818 public ResultScanner getScanner(byte [] family, byte [] qualifier)
819 throws IOException {
820 Scan scan = new Scan();
821 scan.addColumn(family, qualifier);
822 return getScanner(scan);
823 }
824
825
826
827
828 @Override
829 public Result get(final Get get) throws IOException {
830 return get(get, get.isCheckExistenceOnly());
831 }
832
833 private Result get(Get get, final boolean checkExistenceOnly) throws IOException {
834
835 if (get.isCheckExistenceOnly() != checkExistenceOnly || get.getConsistency() == null) {
836 get = ReflectionUtils.newInstance(get.getClass(), get);
837 get.setCheckExistenceOnly(checkExistenceOnly);
838 if (get.getConsistency() == null){
839 get.setConsistency(defaultConsistency);
840 }
841 }
842
843 if (get.getConsistency() == Consistency.STRONG) {
844
845 final Get getReq = get;
846 RegionServerCallable<Result> callable = new RegionServerCallable<Result>(this.connection,
847 getName(), get.getRow()) {
848 @Override
849 public Result call(int callTimeout) throws IOException {
850 ClientProtos.GetRequest request =
851 RequestConverter.buildGetRequest(getLocation().getRegionInfo().getRegionName(), getReq);
852 PayloadCarryingRpcController controller = rpcControllerFactory.newController();
853 controller.setPriority(tableName);
854 controller.setCallTimeout(callTimeout);
855 try {
856 ClientProtos.GetResponse response = getStub().get(controller, request);
857 if (response == null) return null;
858 return ProtobufUtil.toResult(response.getResult(), controller.cellScanner());
859 } catch (ServiceException se) {
860 throw ProtobufUtil.getRemoteException(se);
861 }
862 }
863 };
864 return rpcCallerFactory.<Result>newCaller(rpcTimeout).callWithRetries(callable,
865 this.operationTimeout);
866 }
867
868
869 RpcRetryingCallerWithReadReplicas callable = new RpcRetryingCallerWithReadReplicas(
870 rpcControllerFactory, tableName, this.connection, get, pool,
871 connConfiguration.getRetriesNumber(),
872 operationTimeout,
873 connConfiguration.getPrimaryCallTimeoutMicroSecond());
874 return callable.call();
875 }
876
877
878
879
880
881 @Override
882 public Result[] get(List<Get> gets) throws IOException {
883 if (gets.size() == 1) {
884 return new Result[]{get(gets.get(0))};
885 }
886 try {
887 Object [] r1 = batch((List)gets);
888
889
890 Result [] results = new Result[r1.length];
891 int i=0;
892 for (Object o : r1) {
893
894 results[i++] = (Result) o;
895 }
896
897 return results;
898 } catch (InterruptedException e) {
899 throw (InterruptedIOException)new InterruptedIOException().initCause(e);
900 }
901 }
902
903
904
905
906 @Override
907 public void batch(final List<? extends Row> actions, final Object[] results)
908 throws InterruptedException, IOException {
909 AsyncRequestFuture ars = multiAp.submitAll(pool, tableName, actions, null, results);
910 ars.waitUntilDone();
911 if (ars.hasError()) {
912 throw ars.getErrors();
913 }
914 }
915
916
917
918
919
920
921 @Deprecated
922 @Override
923 public Object[] batch(final List<? extends Row> actions)
924 throws InterruptedException, IOException {
925 Object[] results = new Object[actions.size()];
926 batch(actions, results);
927 return results;
928 }
929
930
931
932
933 @Override
934 public <R> void batchCallback(
935 final List<? extends Row> actions, final Object[] results, final Batch.Callback<R> callback)
936 throws IOException, InterruptedException {
937 connection.processBatchCallback(actions, tableName, pool, results, callback);
938 }
939
940
941
942
943
944
945
946
947 @Deprecated
948 @Override
949 public <R> Object[] batchCallback(
950 final List<? extends Row> actions, final Batch.Callback<R> callback) throws IOException,
951 InterruptedException {
952 Object[] results = new Object[actions.size()];
953 batchCallback(actions, results, callback);
954 return results;
955 }
956
957
958
959
960 @Override
961 public void delete(final Delete delete)
962 throws IOException {
963 RegionServerCallable<Boolean> callable = new RegionServerCallable<Boolean>(connection,
964 tableName, delete.getRow()) {
965 @Override
966 public Boolean call(int callTimeout) throws IOException {
967 PayloadCarryingRpcController controller = rpcControllerFactory.newController();
968 controller.setPriority(tableName);
969 controller.setCallTimeout(callTimeout);
970
971 try {
972 MutateRequest request = RequestConverter.buildMutateRequest(
973 getLocation().getRegionInfo().getRegionName(), delete);
974 MutateResponse response = getStub().mutate(controller, request);
975 return Boolean.valueOf(response.getProcessed());
976 } catch (ServiceException se) {
977 throw ProtobufUtil.getRemoteException(se);
978 }
979 }
980 };
981 rpcCallerFactory.<Boolean> newCaller(rpcTimeout).callWithRetries(callable,
982 this.operationTimeout);
983 }
984
985
986
987
988 @Override
989 public void delete(final List<Delete> deletes)
990 throws IOException {
991 Object[] results = new Object[deletes.size()];
992 try {
993 batch(deletes, results);
994 } catch (InterruptedException e) {
995 throw (InterruptedIOException)new InterruptedIOException().initCause(e);
996 } finally {
997
998
999
1000 for (int i = results.length - 1; i>=0; i--) {
1001
1002 if (results[i] instanceof Result) {
1003 deletes.remove(i);
1004 }
1005 }
1006 }
1007 }
1008
1009
1010
1011
1012
1013 @Override
1014 public void put(final Put put) throws IOException {
1015 getBufferedMutator().mutate(put);
1016 if (autoFlush) {
1017 flushCommits();
1018 }
1019 }
1020
1021
1022
1023
1024
1025 @Override
1026 public void put(final List<Put> puts) throws IOException {
1027 getBufferedMutator().mutate(puts);
1028 if (autoFlush) {
1029 flushCommits();
1030 }
1031 }
1032
1033
1034
1035
1036 @Override
1037 public void mutateRow(final RowMutations rm) throws IOException {
1038 final RetryingTimeTracker tracker = new RetryingTimeTracker().start();
1039 PayloadCarryingServerCallable<MultiResponse> callable =
1040 new PayloadCarryingServerCallable<MultiResponse>(connection, getName(), rm.getRow(),
1041 rpcControllerFactory) {
1042 @Override
1043 public MultiResponse call(int callTimeout) throws IOException {
1044 controller.setPriority(tableName);
1045 int remainingTime = tracker.getRemainingTime(callTimeout);
1046 if (remainingTime == 0) {
1047 throw new DoNotRetryIOException("Timeout for mutate row");
1048 }
1049 controller.setCallTimeout(remainingTime);
1050 try {
1051 RegionAction.Builder regionMutationBuilder = RequestConverter.buildRegionAction(
1052 getLocation().getRegionInfo().getRegionName(), rm);
1053 regionMutationBuilder.setAtomic(true);
1054 MultiRequest request =
1055 MultiRequest.newBuilder().addRegionAction(regionMutationBuilder.build()).build();
1056 ClientProtos.MultiResponse response = getStub().multi(controller, request);
1057 ClientProtos.RegionActionResult res = response.getRegionActionResultList().get(0);
1058 if (res.hasException()) {
1059 Throwable ex = ProtobufUtil.toException(res.getException());
1060 if (ex instanceof IOException) {
1061 throw (IOException) ex;
1062 }
1063 throw new IOException("Failed to mutate row: " +
1064 Bytes.toStringBinary(rm.getRow()), ex);
1065 }
1066 return ResponseConverter.getResults(request, response, controller.cellScanner());
1067 } catch (ServiceException se) {
1068 throw ProtobufUtil.getRemoteException(se);
1069 }
1070 }
1071 };
1072 AsyncRequestFuture ars = multiAp.submitAll(pool, tableName, rm.getMutations(),
1073 null, null, callable, operationTimeout, rpcTimeout);
1074 ars.waitUntilDone();
1075 if (ars.hasError()) {
1076 throw ars.getErrors();
1077 }
1078 }
1079
1080
1081
1082
1083 @Override
1084 public Result append(final Append append) throws IOException {
1085 if (append.numFamilies() == 0) {
1086 throw new IOException(
1087 "Invalid arguments to append, no columns specified");
1088 }
1089
1090 NonceGenerator ng = this.connection.getNonceGenerator();
1091 final long nonceGroup = ng.getNonceGroup(), nonce = ng.newNonce();
1092 RegionServerCallable<Result> callable =
1093 new RegionServerCallable<Result>(this.connection, getName(), append.getRow()) {
1094 @Override
1095 public Result call(int callTimeout) throws IOException {
1096 PayloadCarryingRpcController controller = rpcControllerFactory.newController();
1097 controller.setPriority(getTableName());
1098 controller.setCallTimeout(callTimeout);
1099 try {
1100 MutateRequest request = RequestConverter.buildMutateRequest(
1101 getLocation().getRegionInfo().getRegionName(), append, nonceGroup, nonce);
1102 MutateResponse response = getStub().mutate(controller, request);
1103 if (!response.hasResult()) return null;
1104 return ProtobufUtil.toResult(response.getResult(), controller.cellScanner());
1105 } catch (ServiceException se) {
1106 throw ProtobufUtil.getRemoteException(se);
1107 }
1108 }
1109 };
1110 return rpcCallerFactory.<Result> newCaller(rpcTimeout).callWithRetries(callable,
1111 this.operationTimeout);
1112 }
1113
1114
1115
1116
1117 @Override
1118 public Result increment(final Increment increment) throws IOException {
1119 if (!increment.hasFamilies()) {
1120 throw new IOException(
1121 "Invalid arguments to increment, no columns specified");
1122 }
1123 NonceGenerator ng = this.connection.getNonceGenerator();
1124 final long nonceGroup = ng.getNonceGroup(), nonce = ng.newNonce();
1125 RegionServerCallable<Result> callable = new RegionServerCallable<Result>(this.connection,
1126 getName(), increment.getRow()) {
1127 @Override
1128 public Result call(int callTimeout) throws IOException {
1129 PayloadCarryingRpcController controller = rpcControllerFactory.newController();
1130 controller.setPriority(getTableName());
1131 controller.setCallTimeout(callTimeout);
1132 try {
1133 MutateRequest request = RequestConverter.buildMutateRequest(
1134 getLocation().getRegionInfo().getRegionName(), increment, nonceGroup, nonce);
1135 MutateResponse response = getStub().mutate(controller, request);
1136 return ProtobufUtil.toResult(response.getResult(), controller.cellScanner());
1137 } catch (ServiceException se) {
1138 throw ProtobufUtil.getRemoteException(se);
1139 }
1140 }
1141 };
1142 return rpcCallerFactory.<Result> newCaller(rpcTimeout).callWithRetries(callable,
1143 this.operationTimeout);
1144 }
1145
1146
1147
1148
1149 @Override
1150 public long incrementColumnValue(final byte [] row, final byte [] family,
1151 final byte [] qualifier, final long amount)
1152 throws IOException {
1153 return incrementColumnValue(row, family, qualifier, amount, Durability.SYNC_WAL);
1154 }
1155
1156
1157
1158
1159
1160
1161
1162 @Deprecated
1163 @Override
1164 public long incrementColumnValue(final byte [] row, final byte [] family,
1165 final byte [] qualifier, final long amount, final boolean writeToWAL)
1166 throws IOException {
1167 return incrementColumnValue(row, family, qualifier, amount,
1168 writeToWAL? Durability.SYNC_WAL: Durability.SKIP_WAL);
1169 }
1170
1171
1172
1173
1174 @Override
1175 public long incrementColumnValue(final byte [] row, final byte [] family,
1176 final byte [] qualifier, final long amount, final Durability durability)
1177 throws IOException {
1178 NullPointerException npe = null;
1179 if (row == null) {
1180 npe = new NullPointerException("row is null");
1181 } else if (family == null) {
1182 npe = new NullPointerException("family is null");
1183 } else if (qualifier == null) {
1184 npe = new NullPointerException("qualifier is null");
1185 }
1186 if (npe != null) {
1187 throw new IOException(
1188 "Invalid arguments to incrementColumnValue", npe);
1189 }
1190
1191 NonceGenerator ng = this.connection.getNonceGenerator();
1192 final long nonceGroup = ng.getNonceGroup(), nonce = ng.newNonce();
1193 RegionServerCallable<Long> callable =
1194 new RegionServerCallable<Long>(connection, getName(), row) {
1195 @Override
1196 public Long call(int callTimeout) throws IOException {
1197 PayloadCarryingRpcController controller = rpcControllerFactory.newController();
1198 controller.setPriority(getTableName());
1199 controller.setCallTimeout(callTimeout);
1200 try {
1201 MutateRequest request = RequestConverter.buildIncrementRequest(
1202 getLocation().getRegionInfo().getRegionName(), row, family,
1203 qualifier, amount, durability, nonceGroup, nonce);
1204 MutateResponse response = getStub().mutate(controller, request);
1205 Result result =
1206 ProtobufUtil.toResult(response.getResult(), controller.cellScanner());
1207 return Long.valueOf(Bytes.toLong(result.getValue(family, qualifier)));
1208 } catch (ServiceException se) {
1209 throw ProtobufUtil.getRemoteException(se);
1210 }
1211 }
1212 };
1213 return rpcCallerFactory.<Long> newCaller(rpcTimeout).callWithRetries(callable,
1214 this.operationTimeout);
1215 }
1216
1217
1218
1219
1220 @Override
1221 public boolean checkAndPut(final byte [] row,
1222 final byte [] family, final byte [] qualifier, final byte [] value,
1223 final Put put)
1224 throws IOException {
1225 RegionServerCallable<Boolean> callable =
1226 new RegionServerCallable<Boolean>(connection, getName(), row) {
1227 @Override
1228 public Boolean call(int callTimeout) throws IOException {
1229 PayloadCarryingRpcController controller = rpcControllerFactory.newController();
1230 controller.setPriority(tableName);
1231 controller.setCallTimeout(callTimeout);
1232 try {
1233 MutateRequest request = RequestConverter.buildMutateRequest(
1234 getLocation().getRegionInfo().getRegionName(), row, family, qualifier,
1235 new BinaryComparator(value), CompareType.EQUAL, put);
1236 MutateResponse response = getStub().mutate(controller, request);
1237 return Boolean.valueOf(response.getProcessed());
1238 } catch (ServiceException se) {
1239 throw ProtobufUtil.getRemoteException(se);
1240 }
1241 }
1242 };
1243 return rpcCallerFactory.<Boolean> newCaller(rpcTimeout).callWithRetries(callable,
1244 this.operationTimeout);
1245 }
1246
1247
1248
1249
1250 @Override
1251 public boolean checkAndPut(final byte [] row, final byte [] family,
1252 final byte [] qualifier, final CompareOp compareOp, final byte [] value,
1253 final Put put)
1254 throws IOException {
1255 RegionServerCallable<Boolean> callable =
1256 new RegionServerCallable<Boolean>(connection, getName(), row) {
1257 @Override
1258 public Boolean call(int callTimeout) throws IOException {
1259 PayloadCarryingRpcController controller = new PayloadCarryingRpcController();
1260 controller.setPriority(tableName);
1261 controller.setCallTimeout(callTimeout);
1262 try {
1263 CompareType compareType = CompareType.valueOf(compareOp.name());
1264 MutateRequest request = RequestConverter.buildMutateRequest(
1265 getLocation().getRegionInfo().getRegionName(), row, family, qualifier,
1266 new BinaryComparator(value), compareType, put);
1267 MutateResponse response = getStub().mutate(controller, request);
1268 return Boolean.valueOf(response.getProcessed());
1269 } catch (ServiceException se) {
1270 throw ProtobufUtil.getRemoteException(se);
1271 }
1272 }
1273 };
1274 return rpcCallerFactory.<Boolean> newCaller(rpcTimeout).callWithRetries(callable,
1275 this.operationTimeout);
1276 }
1277
1278
1279
1280
1281 @Override
1282 public boolean checkAndDelete(final byte [] row,
1283 final byte [] family, final byte [] qualifier, final byte [] value,
1284 final Delete delete)
1285 throws IOException {
1286 RegionServerCallable<Boolean> callable =
1287 new RegionServerCallable<Boolean>(connection, getName(), row) {
1288 @Override
1289 public Boolean call(int callTimeout) throws IOException {
1290 PayloadCarryingRpcController controller = rpcControllerFactory.newController();
1291 controller.setPriority(tableName);
1292 controller.setCallTimeout(callTimeout);
1293 try {
1294 MutateRequest request = RequestConverter.buildMutateRequest(
1295 getLocation().getRegionInfo().getRegionName(), row, family, qualifier,
1296 new BinaryComparator(value), CompareType.EQUAL, delete);
1297 MutateResponse response = getStub().mutate(controller, request);
1298 return Boolean.valueOf(response.getProcessed());
1299 } catch (ServiceException se) {
1300 throw ProtobufUtil.getRemoteException(se);
1301 }
1302 }
1303 };
1304 return rpcCallerFactory.<Boolean> newCaller(rpcTimeout).callWithRetries(callable,
1305 this.operationTimeout);
1306 }
1307
1308
1309
1310
1311 @Override
1312 public boolean checkAndDelete(final byte [] row, final byte [] family,
1313 final byte [] qualifier, final CompareOp compareOp, final byte [] value,
1314 final Delete delete)
1315 throws IOException {
1316 RegionServerCallable<Boolean> callable =
1317 new RegionServerCallable<Boolean>(connection, getName(), row) {
1318 @Override
1319 public Boolean call(int callTimeout) throws IOException {
1320 PayloadCarryingRpcController controller = rpcControllerFactory.newController();
1321 controller.setPriority(tableName);
1322 controller.setCallTimeout(callTimeout);
1323 try {
1324 CompareType compareType = CompareType.valueOf(compareOp.name());
1325 MutateRequest request = RequestConverter.buildMutateRequest(
1326 getLocation().getRegionInfo().getRegionName(), row, family, qualifier,
1327 new BinaryComparator(value), compareType, delete);
1328 MutateResponse response = getStub().mutate(controller, request);
1329 return Boolean.valueOf(response.getProcessed());
1330 } catch (ServiceException se) {
1331 throw ProtobufUtil.getRemoteException(se);
1332 }
1333 }
1334 };
1335 return rpcCallerFactory.<Boolean> newCaller(rpcTimeout).callWithRetries(callable,
1336 this.operationTimeout);
1337 }
1338
1339
1340
1341
1342 @Override
1343 public boolean checkAndMutate(final byte [] row, final byte [] family, final byte [] qualifier,
1344 final CompareOp compareOp, final byte [] value, final RowMutations rm)
1345 throws IOException {
1346 final RetryingTimeTracker tracker = new RetryingTimeTracker().start();
1347 PayloadCarryingServerCallable<MultiResponse> callable =
1348 new PayloadCarryingServerCallable<MultiResponse>(connection, getName(), row,
1349 rpcControllerFactory) {
1350 @Override
1351 public MultiResponse call(int callTimeout) throws IOException {
1352 controller.setPriority(tableName);
1353 int remainingTime = tracker.getRemainingTime(callTimeout);
1354 if (remainingTime == 0) {
1355 throw new DoNotRetryIOException("Timeout for mutate row");
1356 }
1357 controller.setCallTimeout(remainingTime);
1358 try {
1359 CompareType compareType = CompareType.valueOf(compareOp.name());
1360 MultiRequest request = RequestConverter.buildMutateRequest(
1361 getLocation().getRegionInfo().getRegionName(), row, family, qualifier,
1362 new BinaryComparator(value), compareType, rm);
1363 ClientProtos.MultiResponse response = getStub().multi(controller, request);
1364 ClientProtos.RegionActionResult res = response.getRegionActionResultList().get(0);
1365 if (res.hasException()) {
1366 Throwable ex = ProtobufUtil.toException(res.getException());
1367 if(ex instanceof IOException) {
1368 throw (IOException)ex;
1369 }
1370 throw new IOException("Failed to checkAndMutate row: "+
1371 Bytes.toStringBinary(row), ex);
1372 }
1373 return ResponseConverter.getResults(request, response, controller.cellScanner());
1374 } catch (ServiceException se) {
1375 throw ProtobufUtil.getRemoteException(se);
1376 }
1377 }
1378 };
1379
1380
1381
1382
1383 Object[] results = new Object[rm.getMutations().size()];
1384 AsyncRequestFuture ars = multiAp.submitAll(pool, tableName, rm.getMutations(),
1385 null, results, callable, operationTimeout, rpcTimeout);
1386 ars.waitUntilDone();
1387 if (ars.hasError()) {
1388 throw ars.getErrors();
1389 }
1390
1391 return ((Result)results[0]).getExists();
1392 }
1393
1394
1395
1396
1397 @Override
1398 public boolean exists(final Get get) throws IOException {
1399 Result r = get(get, true);
1400 assert r.getExists() != null;
1401 return r.getExists();
1402 }
1403
1404
1405
1406
1407 @Override
1408 public boolean[] existsAll(final List<Get> gets) throws IOException {
1409 if (gets.isEmpty()) return new boolean[]{};
1410 if (gets.size() == 1) return new boolean[]{exists(gets.get(0))};
1411
1412 ArrayList<Get> exists = new ArrayList<Get>(gets.size());
1413 for (Get g: gets){
1414 Get ge = new Get(g);
1415 ge.setCheckExistenceOnly(true);
1416 exists.add(ge);
1417 }
1418
1419 Object[] r1;
1420 try {
1421 r1 = batch(exists);
1422 } catch (InterruptedException e) {
1423 throw (InterruptedIOException)new InterruptedIOException().initCause(e);
1424 }
1425
1426
1427 boolean[] results = new boolean[r1.length];
1428 int i = 0;
1429 for (Object o : r1) {
1430
1431 results[i++] = ((Result)o).getExists();
1432 }
1433
1434 return results;
1435 }
1436
1437
1438
1439
1440 @Override
1441 @Deprecated
1442 public Boolean[] exists(final List<Get> gets) throws IOException {
1443 boolean[] results = existsAll(gets);
1444 Boolean[] objectResults = new Boolean[results.length];
1445 for (int i = 0; i < results.length; ++i) {
1446 objectResults[i] = results[i];
1447 }
1448 return objectResults;
1449 }
1450
1451
1452
1453
1454
1455 @Override
1456 public void flushCommits() throws IOException {
1457 if (mutator == null) {
1458
1459 return;
1460 }
1461 getBufferedMutator().flush();
1462 }
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475 public <R> void processBatchCallback(
1476 final List<? extends Row> list, final Object[] results, final Batch.Callback<R> callback)
1477 throws IOException, InterruptedException {
1478 this.batchCallback(list, results, callback);
1479 }
1480
1481
1482
1483
1484
1485
1486 public void processBatch(final List<? extends Row> list, final Object[] results)
1487 throws IOException, InterruptedException {
1488 this.batch(list, results);
1489 }
1490
1491
1492 @Override
1493 public void close() throws IOException {
1494 if (this.closed) {
1495 return;
1496 }
1497 flushCommits();
1498 if (cleanupPoolOnClose) {
1499 this.pool.shutdown();
1500 try {
1501 boolean terminated = false;
1502 do {
1503
1504 terminated = this.pool.awaitTermination(60, TimeUnit.SECONDS);
1505 } while (!terminated);
1506 } catch (InterruptedException e) {
1507 this.pool.shutdownNow();
1508 LOG.warn("waitForTermination interrupted");
1509 }
1510 }
1511 if (cleanupConnectionOnClose) {
1512 if (this.connection != null) {
1513 this.connection.close();
1514 }
1515 }
1516 this.closed = true;
1517 }
1518
1519
1520 public void validatePut(final Put put) throws IllegalArgumentException {
1521 validatePut(put, connConfiguration.getMaxKeyValueSize());
1522 }
1523
1524
1525 public static void validatePut(Put put, int maxKeyValueSize) throws IllegalArgumentException {
1526 if (put.isEmpty()) {
1527 throw new IllegalArgumentException("No columns to insert");
1528 }
1529 if (maxKeyValueSize > 0) {
1530 for (List<Cell> list : put.getFamilyCellMap().values()) {
1531 for (Cell cell : list) {
1532 if (KeyValueUtil.length(cell) > maxKeyValueSize) {
1533 throw new IllegalArgumentException("KeyValue size too large");
1534 }
1535 }
1536 }
1537 }
1538 }
1539
1540
1541
1542
1543 @Override
1544 public boolean isAutoFlush() {
1545 return autoFlush;
1546 }
1547
1548
1549
1550
1551 @Deprecated
1552 @Override
1553 public void setAutoFlush(boolean autoFlush) {
1554 this.autoFlush = autoFlush;
1555 }
1556
1557
1558
1559
1560 @Override
1561 public void setAutoFlushTo(boolean autoFlush) {
1562 this.autoFlush = autoFlush;
1563 }
1564
1565
1566
1567
1568 @Override
1569 public void setAutoFlush(boolean autoFlush, boolean clearBufferOnFail) {
1570 this.autoFlush = autoFlush;
1571 }
1572
1573
1574
1575
1576
1577
1578
1579
1580 @Override
1581 public long getWriteBufferSize() {
1582 if (mutator == null) {
1583 return connConfiguration.getWriteBufferSize();
1584 } else {
1585 return mutator.getWriteBufferSize();
1586 }
1587 }
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597 @Override
1598 public void setWriteBufferSize(long writeBufferSize) throws IOException {
1599 getBufferedMutator();
1600 mutator.setWriteBufferSize(writeBufferSize);
1601 }
1602
1603
1604
1605
1606
1607 ExecutorService getPool() {
1608 return this.pool;
1609 }
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621 @Deprecated
1622 public static void setRegionCachePrefetch(final byte[] tableName,
1623 final boolean enable) throws IOException {
1624 }
1625
1626
1627
1628
1629 @Deprecated
1630 public static void setRegionCachePrefetch(
1631 final TableName tableName,
1632 final boolean enable) throws IOException {
1633 }
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646 @Deprecated
1647 public static void setRegionCachePrefetch(final Configuration conf,
1648 final byte[] tableName, final boolean enable) throws IOException {
1649 }
1650
1651
1652
1653
1654 @Deprecated
1655 public static void setRegionCachePrefetch(final Configuration conf,
1656 final TableName tableName,
1657 final boolean enable) throws IOException {
1658 }
1659
1660
1661
1662
1663
1664
1665
1666
1667
1668
1669 @Deprecated
1670 public static boolean getRegionCachePrefetch(final Configuration conf,
1671 final byte[] tableName) throws IOException {
1672 return false;
1673 }
1674
1675
1676
1677
1678 @Deprecated
1679 public static boolean getRegionCachePrefetch(final Configuration conf,
1680 final TableName tableName) throws IOException {
1681 return false;
1682 }
1683
1684
1685
1686
1687
1688
1689
1690
1691
1692 @Deprecated
1693 public static boolean getRegionCachePrefetch(final byte[] tableName) throws IOException {
1694 return false;
1695 }
1696
1697
1698
1699
1700 @Deprecated
1701 public static boolean getRegionCachePrefetch(
1702 final TableName tableName) throws IOException {
1703 return false;
1704 }
1705
1706
1707
1708
1709
1710 public void clearRegionCache() {
1711 this.connection.clearRegionCache();
1712 }
1713
1714
1715
1716
1717 @Override
1718 public CoprocessorRpcChannel coprocessorService(byte[] row) {
1719 return new RegionCoprocessorRpcChannel(connection, tableName, row);
1720 }
1721
1722
1723
1724
1725 @Override
1726 public <T extends Service, R> Map<byte[],R> coprocessorService(final Class<T> service,
1727 byte[] startKey, byte[] endKey, final Batch.Call<T,R> callable)
1728 throws ServiceException, Throwable {
1729 final Map<byte[],R> results = Collections.synchronizedMap(
1730 new TreeMap<byte[], R>(Bytes.BYTES_COMPARATOR));
1731 coprocessorService(service, startKey, endKey, callable, new Batch.Callback<R>() {
1732 @Override
1733 public void update(byte[] region, byte[] row, R value) {
1734 if (region != null) {
1735 results.put(region, value);
1736 }
1737 }
1738 });
1739 return results;
1740 }
1741
1742
1743
1744
1745 @Override
1746 public <T extends Service, R> void coprocessorService(final Class<T> service,
1747 byte[] startKey, byte[] endKey, final Batch.Call<T,R> callable,
1748 final Batch.Callback<R> callback) throws ServiceException, Throwable {
1749
1750
1751 List<byte[]> keys = getStartKeysInRange(startKey, endKey);
1752
1753 Map<byte[],Future<R>> futures =
1754 new TreeMap<byte[],Future<R>>(Bytes.BYTES_COMPARATOR);
1755 for (final byte[] r : keys) {
1756 final RegionCoprocessorRpcChannel channel =
1757 new RegionCoprocessorRpcChannel(connection, tableName, r);
1758 Future<R> future = pool.submit(
1759 new Callable<R>() {
1760 @Override
1761 public R call() throws Exception {
1762 T instance = ProtobufUtil.newServiceStub(service, channel);
1763 R result = callable.call(instance);
1764 byte[] region = channel.getLastRegion();
1765 if (callback != null) {
1766 callback.update(region, r, result);
1767 }
1768 return result;
1769 }
1770 });
1771 futures.put(r, future);
1772 }
1773 for (Map.Entry<byte[],Future<R>> e : futures.entrySet()) {
1774 try {
1775 e.getValue().get();
1776 } catch (ExecutionException ee) {
1777 LOG.warn("Error calling coprocessor service " + service.getName() + " for row "
1778 + Bytes.toStringBinary(e.getKey()), ee);
1779 throw ee.getCause();
1780 } catch (InterruptedException ie) {
1781 throw new InterruptedIOException("Interrupted calling coprocessor service " + service.getName()
1782 + " for row " + Bytes.toStringBinary(e.getKey()))
1783 .initCause(ie);
1784 }
1785 }
1786 }
1787
1788 private List<byte[]> getStartKeysInRange(byte[] start, byte[] end)
1789 throws IOException {
1790 if (start == null) {
1791 start = HConstants.EMPTY_START_ROW;
1792 }
1793 if (end == null) {
1794 end = HConstants.EMPTY_END_ROW;
1795 }
1796 return getKeysAndRegionsInRange(start, end, true).getFirst();
1797 }
1798
1799 public void setOperationTimeout(int operationTimeout) {
1800 this.operationTimeout = operationTimeout;
1801 if (mutator != null) {
1802 mutator.setOperationTimeout(operationTimeout);
1803 }
1804 multiAp.setOperationTimeout(operationTimeout);
1805 }
1806
1807 public int getOperationTimeout() {
1808 return operationTimeout;
1809 }
1810
1811 @Override public void setRpcTimeout(int rpcTimeout) {
1812 this.rpcTimeout = rpcTimeout;
1813 if (mutator != null) {
1814 mutator.setRpcTimeout(rpcTimeout);
1815 }
1816 multiAp.setRpcTimeout(rpcTimeout);
1817 }
1818
1819 @Override public int getRpcTimeout() {
1820 return rpcTimeout;
1821 }
1822
1823 @Override
1824 public String toString() {
1825 return tableName + ";" + connection;
1826 }
1827
1828
1829
1830
1831 @Override
1832 public <R extends Message> Map<byte[], R> batchCoprocessorService(
1833 Descriptors.MethodDescriptor methodDescriptor, Message request,
1834 byte[] startKey, byte[] endKey, R responsePrototype) throws ServiceException, Throwable {
1835 final Map<byte[], R> results = Collections.synchronizedMap(new TreeMap<byte[], R>(
1836 Bytes.BYTES_COMPARATOR));
1837 batchCoprocessorService(methodDescriptor, request, startKey, endKey, responsePrototype,
1838 new Callback<R>() {
1839
1840 @Override
1841 public void update(byte[] region, byte[] row, R result) {
1842 if (region != null) {
1843 results.put(region, result);
1844 }
1845 }
1846 });
1847 return results;
1848 }
1849
1850
1851
1852
1853 @Override
1854 public <R extends Message> void batchCoprocessorService(
1855 final Descriptors.MethodDescriptor methodDescriptor, final Message request,
1856 byte[] startKey, byte[] endKey, final R responsePrototype, final Callback<R> callback)
1857 throws ServiceException, Throwable {
1858
1859 if (startKey == null) {
1860 startKey = HConstants.EMPTY_START_ROW;
1861 }
1862 if (endKey == null) {
1863 endKey = HConstants.EMPTY_END_ROW;
1864 }
1865
1866 Pair<List<byte[]>, List<HRegionLocation>> keysAndRegions =
1867 getKeysAndRegionsInRange(startKey, endKey, true);
1868 List<byte[]> keys = keysAndRegions.getFirst();
1869 List<HRegionLocation> regions = keysAndRegions.getSecond();
1870
1871
1872 if (keys.isEmpty()) {
1873 LOG.info("No regions were selected by key range start=" + Bytes.toStringBinary(startKey) +
1874 ", end=" + Bytes.toStringBinary(endKey));
1875 return;
1876 }
1877
1878 List<RegionCoprocessorServiceExec> execs = new ArrayList<RegionCoprocessorServiceExec>();
1879 final Map<byte[], RegionCoprocessorServiceExec> execsByRow =
1880 new TreeMap<byte[], RegionCoprocessorServiceExec>(Bytes.BYTES_COMPARATOR);
1881 for (int i = 0; i < keys.size(); i++) {
1882 final byte[] rowKey = keys.get(i);
1883 final byte[] region = regions.get(i).getRegionInfo().getRegionName();
1884 RegionCoprocessorServiceExec exec =
1885 new RegionCoprocessorServiceExec(region, rowKey, methodDescriptor, request);
1886 execs.add(exec);
1887 execsByRow.put(rowKey, exec);
1888 }
1889
1890
1891
1892 final List<Throwable> callbackErrorExceptions = new ArrayList<Throwable>();
1893 final List<Row> callbackErrorActions = new ArrayList<Row>();
1894 final List<String> callbackErrorServers = new ArrayList<String>();
1895 Object[] results = new Object[execs.size()];
1896
1897 AsyncProcess asyncProcess =
1898 new AsyncProcess(connection, configuration, pool,
1899 RpcRetryingCallerFactory.instantiate(configuration, connection.getStatisticsTracker()),
1900 true, RpcControllerFactory.instantiate(configuration), rpcTimeout);
1901
1902 AsyncRequestFuture future = asyncProcess.submitAll(tableName, execs,
1903 new Callback<ClientProtos.CoprocessorServiceResult>() {
1904 @Override
1905 public void update(byte[] region, byte[] row,
1906 ClientProtos.CoprocessorServiceResult serviceResult) {
1907 if (LOG.isTraceEnabled()) {
1908 LOG.trace("Received result for endpoint " + methodDescriptor.getFullName() +
1909 ": region=" + Bytes.toStringBinary(region) +
1910 ", row=" + Bytes.toStringBinary(row) +
1911 ", value=" + serviceResult.getValue().getValue());
1912 }
1913 try {
1914 Message.Builder builder = responsePrototype.newBuilderForType();
1915 ProtobufUtil.mergeFrom(builder, serviceResult.getValue().getValue());
1916 callback.update(region, row, (R) builder.build());
1917 } catch (IOException e) {
1918 LOG.error("Unexpected response type from endpoint " + methodDescriptor.getFullName(),
1919 e);
1920 callbackErrorExceptions.add(e);
1921 callbackErrorActions.add(execsByRow.get(row));
1922 callbackErrorServers.add("null");
1923 }
1924 }
1925 }, results);
1926
1927 future.waitUntilDone();
1928
1929 if (future.hasError()) {
1930 throw future.getErrors();
1931 } else if (!callbackErrorExceptions.isEmpty()) {
1932 throw new RetriesExhaustedWithDetailsException(callbackErrorExceptions, callbackErrorActions,
1933 callbackErrorServers);
1934 }
1935 }
1936
1937 public RegionLocator getRegionLocator() {
1938 return this.locator;
1939 }
1940
1941 @VisibleForTesting
1942 BufferedMutator getBufferedMutator() throws IOException {
1943 if (mutator == null) {
1944 this.mutator = (BufferedMutatorImpl) connection.getBufferedMutator(
1945 new BufferedMutatorParams(tableName)
1946 .pool(pool)
1947 .writeBufferSize(connConfiguration.getWriteBufferSize())
1948 .maxKeyValueSize(connConfiguration.getMaxKeyValueSize())
1949 );
1950 mutator.setRpcTimeout(rpcTimeout);
1951 mutator.setOperationTimeout(operationTimeout);
1952 }
1953 return mutator;
1954 }
1955 }