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