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