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