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 static org.apache.hadoop.hbase.client.MetricsConnection.CLIENT_SIDE_METRICS_ENABLED_KEY;
22
23 import java.io.Closeable;
24 import java.io.IOException;
25 import java.io.InterruptedIOException;
26 import java.lang.reflect.UndeclaredThrowableException;
27 import java.net.InetAddress;
28 import java.net.InetSocketAddress;
29 import java.util.ArrayList;
30 import java.util.Date;
31 import java.util.HashSet;
32 import java.util.LinkedHashMap;
33 import java.util.List;
34 import java.util.concurrent.BlockingQueue;
35 import java.util.Map;
36 import java.util.Map.Entry;
37 import java.util.NavigableMap;
38 import java.util.Set;
39 import java.util.concurrent.ConcurrentHashMap;
40 import java.util.concurrent.ConcurrentMap;
41 import java.util.concurrent.ExecutorService;
42 import java.util.concurrent.CopyOnWriteArraySet;
43 import java.util.concurrent.LinkedBlockingQueue;
44 import java.util.concurrent.ThreadPoolExecutor;
45 import java.util.concurrent.TimeUnit;
46 import java.util.concurrent.atomic.AtomicBoolean;
47 import java.util.concurrent.atomic.AtomicInteger;
48
49 import org.apache.commons.logging.Log;
50 import org.apache.commons.logging.LogFactory;
51 import org.apache.hadoop.conf.Configuration;
52 import org.apache.hadoop.hbase.DoNotRetryIOException;
53 import org.apache.hadoop.hbase.HBaseConfiguration;
54 import org.apache.hadoop.hbase.HConstants;
55 import org.apache.hadoop.hbase.HRegionInfo;
56 import org.apache.hadoop.hbase.HRegionLocation;
57 import org.apache.hadoop.hbase.HTableDescriptor;
58 import org.apache.hadoop.hbase.MasterNotRunningException;
59 import org.apache.hadoop.hbase.MetaTableAccessor;
60 import org.apache.hadoop.hbase.RegionLocations;
61 import org.apache.hadoop.hbase.ServerName;
62 import org.apache.hadoop.hbase.TableName;
63 import org.apache.hadoop.hbase.TableNotEnabledException;
64 import org.apache.hadoop.hbase.TableNotFoundException;
65 import org.apache.hadoop.hbase.ZooKeeperConnectionException;
66 import org.apache.hadoop.hbase.classification.InterfaceAudience;
67 import org.apache.hadoop.hbase.client.AsyncProcess.AsyncRequestFuture;
68 import org.apache.hadoop.hbase.client.MetaScanner.MetaScannerVisitor;
69 import org.apache.hadoop.hbase.client.MetaScanner.MetaScannerVisitorBase;
70 import org.apache.hadoop.hbase.client.backoff.ClientBackoffPolicy;
71 import org.apache.hadoop.hbase.client.backoff.ClientBackoffPolicyFactory;
72 import org.apache.hadoop.hbase.client.coprocessor.Batch;
73 import org.apache.hadoop.hbase.exceptions.ClientExceptionsUtil;
74 import org.apache.hadoop.hbase.exceptions.RegionMovedException;
75 import org.apache.hadoop.hbase.ipc.RpcClient;
76 import org.apache.hadoop.hbase.ipc.RpcClientFactory;
77 import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
78 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
79 import org.apache.hadoop.hbase.protobuf.RequestConverter;
80 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.AdminService;
81 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService;
82 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.CoprocessorServiceRequest;
83 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.CoprocessorServiceResponse;
84 import org.apache.hadoop.hbase.protobuf.generated.*;
85 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.AddColumnRequest;
86 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.AddColumnResponse;
87 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.AssignRegionRequest;
88 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.AssignRegionResponse;
89 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.BalanceRequest;
90 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.BalanceResponse;
91 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.CreateNamespaceRequest;
92 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.CreateNamespaceResponse;
93 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.CreateTableRequest;
94 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.CreateTableResponse;
95 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.DeleteColumnRequest;
96 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.DeleteColumnResponse;
97 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.DeleteNamespaceRequest;
98 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.DeleteNamespaceResponse;
99 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.DeleteSnapshotRequest;
100 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.DeleteSnapshotResponse;
101 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.DeleteTableRequest;
102 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.DeleteTableResponse;
103 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.DisableTableRequest;
104 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.DisableTableResponse;
105 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.DispatchMergingRegionsRequest;
106 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.DispatchMergingRegionsResponse;
107 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.EnableCatalogJanitorRequest;
108 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.EnableCatalogJanitorResponse;
109 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.EnableTableRequest;
110 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.EnableTableResponse;
111 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.ExecProcedureRequest;
112 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.ExecProcedureResponse;
113 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.GetClusterStatusRequest;
114 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.GetClusterStatusResponse;
115 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.GetCompletedSnapshotsRequest;
116 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.GetCompletedSnapshotsResponse;
117 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.GetNamespaceDescriptorRequest;
118 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.GetNamespaceDescriptorResponse;
119 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.GetSchemaAlterStatusRequest;
120 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.GetSchemaAlterStatusResponse;
121 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.GetTableDescriptorsRequest;
122 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.GetTableDescriptorsResponse;
123 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.GetTableNamesRequest;
124 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.GetTableNamesResponse;
125 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.GetProcedureResultRequest;
126 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.GetProcedureResultResponse;
127 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.IsBalancerEnabledRequest;
128 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.IsBalancerEnabledResponse;
129 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.IsCatalogJanitorEnabledRequest;
130 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.IsCatalogJanitorEnabledResponse;
131 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.IsMasterRunningRequest;
132 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.IsMasterRunningResponse;
133 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.IsNormalizerEnabledRequest;
134 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.IsNormalizerEnabledResponse;
135 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.IsProcedureDoneRequest;
136 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.IsProcedureDoneResponse;
137 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.IsRestoreSnapshotDoneRequest;
138 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.IsRestoreSnapshotDoneResponse;
139 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.IsSnapshotDoneRequest;
140 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.IsSnapshotDoneResponse;
141 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.ListNamespaceDescriptorsRequest;
142 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.ListNamespaceDescriptorsResponse;
143 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.ListTableDescriptorsByNamespaceRequest;
144 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.ListTableDescriptorsByNamespaceResponse;
145 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.ListTableNamesByNamespaceRequest;
146 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.ListTableNamesByNamespaceResponse;
147 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.MajorCompactionTimestampForRegionRequest;
148 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.MajorCompactionTimestampRequest;
149 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.MajorCompactionTimestampResponse;
150 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.MasterService;
151 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.ModifyColumnRequest;
152 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.ModifyColumnResponse;
153 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.ModifyNamespaceRequest;
154 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.ModifyNamespaceResponse;
155 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.ModifyTableRequest;
156 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.ModifyTableResponse;
157 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.MoveRegionRequest;
158 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.MoveRegionResponse;
159 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.NormalizeRequest;
160 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.NormalizeResponse;
161 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.OfflineRegionRequest;
162 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.OfflineRegionResponse;
163 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.RestoreSnapshotRequest;
164 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.RestoreSnapshotResponse;
165 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.RunCatalogScanRequest;
166 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.RunCatalogScanResponse;
167 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.SecurityCapabilitiesRequest;
168 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.SecurityCapabilitiesResponse;
169 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.SetBalancerRunningRequest;
170 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.SetBalancerRunningResponse;
171 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.SetNormalizerRunningRequest;
172 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.SetNormalizerRunningResponse;
173 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.SetQuotaRequest;
174 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.SetQuotaResponse;
175 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.ShutdownRequest;
176 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.ShutdownResponse;
177 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.SnapshotRequest;
178 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.SnapshotResponse;
179 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.StopMasterRequest;
180 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.StopMasterResponse;
181 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.TruncateTableRequest;
182 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.TruncateTableResponse;
183 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.UnassignRegionRequest;
184 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.UnassignRegionResponse;
185 import org.apache.hadoop.hbase.regionserver.RegionServerStoppedException;
186 import org.apache.hadoop.hbase.security.User;
187 import org.apache.hadoop.hbase.security.UserProvider;
188 import org.apache.hadoop.hbase.util.Bytes;
189 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
190 import org.apache.hadoop.hbase.util.ExceptionUtil;
191 import org.apache.hadoop.hbase.util.Threads;
192 import org.apache.hadoop.hbase.zookeeper.MasterAddressTracker;
193 import org.apache.hadoop.hbase.zookeeper.ZKUtil;
194 import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
195 import org.apache.hadoop.ipc.RemoteException;
196 import org.apache.zookeeper.KeeperException;
197
198 import com.google.common.annotations.VisibleForTesting;
199 import com.google.protobuf.BlockingRpcChannel;
200 import com.google.protobuf.RpcController;
201 import com.google.protobuf.ServiceException;
202
203
204
205
206 @SuppressWarnings("serial")
207 @InterfaceAudience.Private
208
209 class ConnectionManager {
210 static final Log LOG = LogFactory.getLog(ConnectionManager.class);
211
212 public static final String RETRIES_BY_SERVER_KEY = "hbase.client.retries.by.server";
213 private static final String CLIENT_NONCES_ENABLED_KEY = "hbase.client.nonces.enabled";
214 private static final String RESOLVE_HOSTNAME_ON_FAIL_KEY = "hbase.resolve.hostnames.on.failure";
215
216
217
218
219 static final Map<HConnectionKey, HConnectionImplementation> CONNECTION_INSTANCES;
220
221 public static final int MAX_CACHED_CONNECTION_INSTANCES;
222
223
224
225
226
227 private static volatile NonceGenerator nonceGenerator = null;
228
229
230 private static Object nonceGeneratorCreateLock = new Object();
231
232 static {
233
234
235
236
237 MAX_CACHED_CONNECTION_INSTANCES = HBaseConfiguration.create().getInt(
238 HConstants.ZOOKEEPER_MAX_CLIENT_CNXNS, HConstants.DEFAULT_ZOOKEPER_MAX_CLIENT_CNXNS) + 1;
239 CONNECTION_INSTANCES = new LinkedHashMap<HConnectionKey, HConnectionImplementation>(
240 (int) (MAX_CACHED_CONNECTION_INSTANCES / 0.75F) + 1, 0.75F, true) {
241 @Override
242 protected boolean removeEldestEntry(
243 Map.Entry<HConnectionKey, HConnectionImplementation> eldest) {
244 return size() > MAX_CACHED_CONNECTION_INSTANCES;
245 }
246 };
247 }
248
249
250 static class NoNonceGenerator implements NonceGenerator {
251 @Override
252 public long getNonceGroup() {
253 return HConstants.NO_NONCE;
254 }
255 @Override
256 public long newNonce() {
257 return HConstants.NO_NONCE;
258 }
259 }
260
261
262
263
264 private ConnectionManager() {
265 super();
266 }
267
268
269
270
271
272
273 @VisibleForTesting
274 static NonceGenerator injectNonceGeneratorForTesting(
275 ClusterConnection conn, NonceGenerator cnm) {
276 HConnectionImplementation connImpl = (HConnectionImplementation)conn;
277 NonceGenerator ng = connImpl.getNonceGenerator();
278 LOG.warn("Nonce generator is being replaced by test code for " + cnm.getClass().getName());
279 connImpl.nonceGenerator = cnm;
280 return ng;
281 }
282
283
284
285
286
287
288
289
290
291
292 @Deprecated
293 public static HConnection getConnection(final Configuration conf) throws IOException {
294 return getConnectionInternal(conf);
295 }
296
297
298 static ClusterConnection getConnectionInternal(final Configuration conf)
299 throws IOException {
300 HConnectionKey connectionKey = new HConnectionKey(conf);
301 synchronized (CONNECTION_INSTANCES) {
302 HConnectionImplementation connection = CONNECTION_INSTANCES.get(connectionKey);
303 if (connection == null) {
304 connection = (HConnectionImplementation)createConnection(conf, true);
305 CONNECTION_INSTANCES.put(connectionKey, connection);
306 } else if (connection.isClosed()) {
307 ConnectionManager.deleteConnection(connectionKey, true);
308 connection = (HConnectionImplementation)createConnection(conf, true);
309 CONNECTION_INSTANCES.put(connectionKey, connection);
310 }
311 connection.incCount();
312 return connection;
313 }
314 }
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336 public static HConnection createConnection(Configuration conf) throws IOException {
337 return createConnectionInternal(conf);
338 }
339
340 static ClusterConnection createConnectionInternal(Configuration conf) throws IOException {
341 UserProvider provider = UserProvider.instantiate(conf);
342 return createConnection(conf, false, null, provider.getCurrent());
343 }
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365 public static HConnection createConnection(Configuration conf, ExecutorService pool)
366 throws IOException {
367 UserProvider provider = UserProvider.instantiate(conf);
368 return createConnection(conf, false, pool, provider.getCurrent());
369 }
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391 public static HConnection createConnection(Configuration conf, User user)
392 throws IOException {
393 return createConnection(conf, false, null, user);
394 }
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417 public static HConnection createConnection(Configuration conf, ExecutorService pool, User user)
418 throws IOException {
419 return createConnection(conf, false, pool, user);
420 }
421
422 @Deprecated
423 static HConnection createConnection(final Configuration conf, final boolean managed)
424 throws IOException {
425 UserProvider provider = UserProvider.instantiate(conf);
426 return createConnection(conf, managed, null, provider.getCurrent());
427 }
428
429 @Deprecated
430 static ClusterConnection createConnection(final Configuration conf, final boolean managed,
431 final ExecutorService pool, final User user)
432 throws IOException {
433 return (ClusterConnection) ConnectionFactory.createConnection(conf, managed, pool, user);
434 }
435
436
437
438
439
440
441
442
443
444 @Deprecated
445 public static void deleteConnection(Configuration conf) {
446 deleteConnection(new HConnectionKey(conf), false);
447 }
448
449
450
451
452
453
454
455
456 @Deprecated
457 public static void deleteStaleConnection(HConnection connection) {
458 deleteConnection(connection, true);
459 }
460
461
462
463
464
465
466
467 @Deprecated
468 public static void deleteAllConnections(boolean staleConnection) {
469 synchronized (CONNECTION_INSTANCES) {
470 Set<HConnectionKey> connectionKeys = new HashSet<HConnectionKey>();
471 connectionKeys.addAll(CONNECTION_INSTANCES.keySet());
472 for (HConnectionKey connectionKey : connectionKeys) {
473 deleteConnection(connectionKey, staleConnection);
474 }
475 CONNECTION_INSTANCES.clear();
476 }
477 }
478
479
480
481
482
483 @Deprecated
484 public static void deleteAllConnections() {
485 deleteAllConnections(false);
486 }
487
488
489 @Deprecated
490 private static void deleteConnection(HConnection connection, boolean staleConnection) {
491 synchronized (CONNECTION_INSTANCES) {
492 for (Entry<HConnectionKey, HConnectionImplementation> e: CONNECTION_INSTANCES.entrySet()) {
493 if (e.getValue() == connection) {
494 deleteConnection(e.getKey(), staleConnection);
495 break;
496 }
497 }
498 }
499 }
500
501 @Deprecated
502 private static void deleteConnection(HConnectionKey connectionKey, boolean staleConnection) {
503 synchronized (CONNECTION_INSTANCES) {
504 HConnectionImplementation connection = CONNECTION_INSTANCES.get(connectionKey);
505 if (connection != null) {
506 connection.decCount();
507 if (connection.isZeroReference() || staleConnection) {
508 CONNECTION_INSTANCES.remove(connectionKey);
509 connection.internalClose();
510 }
511 } else {
512 LOG.error("Connection not found in the list, can't delete it "+
513 "(connection key=" + connectionKey + "). May be the key was modified?", new Exception());
514 }
515 }
516 }
517
518
519
520
521
522
523
524
525
526
527
528
529 @InterfaceAudience.Private
530 public static <T> T execute(HConnectable<T> connectable) throws IOException {
531 if (connectable == null || connectable.conf == null) {
532 return null;
533 }
534 Configuration conf = connectable.conf;
535 HConnection connection = getConnection(conf);
536 boolean connectSucceeded = false;
537 try {
538 T returnValue = connectable.connect(connection);
539 connectSucceeded = true;
540 return returnValue;
541 } finally {
542 try {
543 connection.close();
544 } catch (Exception e) {
545 ExceptionUtil.rethrowIfInterrupt(e);
546 if (connectSucceeded) {
547 throw new IOException("The connection to " + connection
548 + " could not be deleted.", e);
549 }
550 }
551 }
552 }
553
554
555 @edu.umd.cs.findbugs.annotations.SuppressWarnings(
556 value="AT_OPERATION_SEQUENCE_ON_CONCURRENT_ABSTRACTION",
557 justification="Access to the conncurrent hash map is under a lock so should be fine.")
558 static class HConnectionImplementation implements ClusterConnection, Closeable {
559 static final Log LOG = LogFactory.getLog(HConnectionImplementation.class);
560 private final boolean hostnamesCanChange;
561 private final long pause;
562 private final boolean useMetaReplicas;
563 private final int numTries;
564 final int rpcTimeout;
565 private NonceGenerator nonceGenerator = null;
566 private final boolean usePrefetch;
567 private final int prefetchRegionLimit;
568 private final AsyncProcess asyncProcess;
569
570 private final ServerStatisticTracker stats;
571
572 private volatile boolean closed;
573 private volatile boolean aborted;
574
575
576 ClusterStatusListener clusterStatusListener;
577
578
579 private final Object metaRegionLock = new Object();
580
581
582
583
584
585
586 private final Object masterAndZKLock = new Object();
587
588 private long keepZooKeeperWatcherAliveUntil = Long.MAX_VALUE;
589
590
591
592 private volatile ExecutorService batchPool = null;
593
594
595 private volatile ExecutorService metaLookupPool = null;
596 private volatile boolean cleanupPool = false;
597
598 private final Configuration conf;
599
600
601
602 private final TableConfiguration tableConfig;
603
604
605 private RpcClient rpcClient;
606
607
608
609 private final Set<Integer> regionCachePrefetchDisabledTables =
610 new CopyOnWriteArraySet<Integer>();
611
612 private final MetaCache metaCache;
613 private final MetricsConnection metrics;
614
615 private int refCount;
616
617
618 private boolean managed;
619
620 protected User user;
621
622 private RpcRetryingCallerFactory rpcCallerFactory;
623
624 private RpcControllerFactory rpcControllerFactory;
625
626 private final RetryingCallerInterceptor interceptor;
627
628
629
630
631 Registry registry;
632
633 private final ClientBackoffPolicy backoffPolicy;
634
635 HConnectionImplementation(Configuration conf, boolean managed) throws IOException {
636 this(conf, managed, null, null);
637 }
638
639
640
641
642
643
644
645
646
647
648
649
650 HConnectionImplementation(Configuration conf, boolean managed,
651 ExecutorService pool, User user) throws IOException {
652 this(conf);
653 this.user = user;
654 this.batchPool = pool;
655 this.managed = managed;
656 this.registry = setupRegistry();
657 retrieveClusterId();
658
659 this.rpcClient = RpcClientFactory.createClient(this.conf, this.clusterId, this.metrics);
660 this.rpcControllerFactory = RpcControllerFactory.instantiate(conf);
661
662
663 boolean shouldListen = conf.getBoolean(HConstants.STATUS_PUBLISHED,
664 HConstants.STATUS_PUBLISHED_DEFAULT);
665 Class<? extends ClusterStatusListener.Listener> listenerClass =
666 conf.getClass(ClusterStatusListener.STATUS_LISTENER_CLASS,
667 ClusterStatusListener.DEFAULT_STATUS_LISTENER_CLASS,
668 ClusterStatusListener.Listener.class);
669 if (shouldListen) {
670 if (listenerClass == null) {
671 LOG.warn(HConstants.STATUS_PUBLISHED + " is true, but " +
672 ClusterStatusListener.STATUS_LISTENER_CLASS + " is not set - not listening status");
673 } else {
674 clusterStatusListener = new ClusterStatusListener(
675 new ClusterStatusListener.DeadServerHandler() {
676 @Override
677 public void newDead(ServerName sn) {
678 clearCaches(sn);
679 rpcClient.cancelConnections(sn);
680 }
681 }, conf, listenerClass);
682 }
683 }
684 }
685
686
687
688
689 protected HConnectionImplementation(Configuration conf) {
690 this.conf = conf;
691 this.tableConfig = new TableConfiguration(conf);
692 this.closed = false;
693 this.pause = conf.getLong(HConstants.HBASE_CLIENT_PAUSE,
694 HConstants.DEFAULT_HBASE_CLIENT_PAUSE);
695 this.useMetaReplicas = conf.getBoolean(HConstants.USE_META_REPLICAS,
696 HConstants.DEFAULT_USE_META_REPLICAS);
697 this.numTries = tableConfig.getRetriesNumber();
698 this.rpcTimeout = conf.getInt(
699 HConstants.HBASE_RPC_TIMEOUT_KEY,
700 HConstants.DEFAULT_HBASE_RPC_TIMEOUT);
701 if (conf.getBoolean(CLIENT_NONCES_ENABLED_KEY, true)) {
702 synchronized (nonceGeneratorCreateLock) {
703 if (ConnectionManager.nonceGenerator == null) {
704 ConnectionManager.nonceGenerator = new PerClientRandomNonceGenerator();
705 }
706 this.nonceGenerator = ConnectionManager.nonceGenerator;
707 }
708 } else {
709 this.nonceGenerator = new NoNonceGenerator();
710 }
711 stats = ServerStatisticTracker.create(conf);
712 this.asyncProcess = createAsyncProcess(this.conf);
713 this.usePrefetch = conf.getBoolean(HConstants.HBASE_CLIENT_PREFETCH,
714 HConstants.DEFAULT_HBASE_CLIENT_PREFETCH);
715 this.prefetchRegionLimit = conf.getInt(
716 HConstants.HBASE_CLIENT_PREFETCH_LIMIT,
717 HConstants.DEFAULT_HBASE_CLIENT_PREFETCH_LIMIT);
718 this.interceptor = (new RetryingCallerInterceptorFactory(conf)).build();
719 this.rpcCallerFactory = RpcRetryingCallerFactory.instantiate(conf, interceptor, this.stats);
720 this.backoffPolicy = ClientBackoffPolicyFactory.create(conf);
721 if (conf.getBoolean(CLIENT_SIDE_METRICS_ENABLED_KEY, false)) {
722 this.metrics = new MetricsConnection(this);
723 } else {
724 this.metrics = null;
725 }
726
727 this.hostnamesCanChange = conf.getBoolean(RESOLVE_HOSTNAME_ON_FAIL_KEY, true);
728 this.metaCache = new MetaCache(this.metrics);
729 }
730
731 @Override
732 public HTableInterface getTable(String tableName) throws IOException {
733 return getTable(TableName.valueOf(tableName));
734 }
735
736 @Override
737 public HTableInterface getTable(byte[] tableName) throws IOException {
738 return getTable(TableName.valueOf(tableName));
739 }
740
741 @Override
742 public HTableInterface getTable(TableName tableName) throws IOException {
743 return getTable(tableName, getBatchPool());
744 }
745
746 @Override
747 public HTableInterface getTable(String tableName, ExecutorService pool) throws IOException {
748 return getTable(TableName.valueOf(tableName), pool);
749 }
750
751 @Override
752 public HTableInterface getTable(byte[] tableName, ExecutorService pool) throws IOException {
753 return getTable(TableName.valueOf(tableName), pool);
754 }
755
756 @Override
757 public HTableInterface getTable(TableName tableName, ExecutorService pool) throws IOException {
758 if (managed) {
759 throw new NeedUnmanagedConnectionException();
760 }
761 return new HTable(tableName, this, tableConfig, rpcCallerFactory, rpcControllerFactory, pool);
762 }
763
764 @Override
765 public BufferedMutator getBufferedMutator(BufferedMutatorParams params) {
766 if (params.getTableName() == null) {
767 throw new IllegalArgumentException("TableName cannot be null.");
768 }
769 if (params.getPool() == null) {
770 params.pool(HTable.getDefaultExecutor(getConfiguration()));
771 }
772 if (params.getWriteBufferSize() == BufferedMutatorParams.UNSET) {
773 params.writeBufferSize(tableConfig.getWriteBufferSize());
774 }
775 if (params.getMaxKeyValueSize() == BufferedMutatorParams.UNSET) {
776 params.maxKeyValueSize(tableConfig.getMaxKeyValueSize());
777 }
778 return new BufferedMutatorImpl(this, rpcCallerFactory, rpcControllerFactory, params);
779 }
780
781 @Override
782 public BufferedMutator getBufferedMutator(TableName tableName) {
783 return getBufferedMutator(new BufferedMutatorParams(tableName));
784 }
785
786 @Override
787 public RegionLocator getRegionLocator(TableName tableName) throws IOException {
788 return new HRegionLocator(tableName, this);
789 }
790
791 @Override
792 public Admin getAdmin() throws IOException {
793 if (managed) {
794 throw new NeedUnmanagedConnectionException();
795 }
796 return new HBaseAdmin(this);
797 }
798
799 @Override
800 public MetricsConnection getConnectionMetrics() {
801 return this.metrics;
802 }
803
804 private ExecutorService getBatchPool() {
805 if (batchPool == null) {
806 synchronized (this) {
807 if (batchPool == null) {
808 this.batchPool = getThreadPool(conf.getInt("hbase.hconnection.threads.max", 256),
809 conf.getInt("hbase.hconnection.threads.core", 256), "-shared-", null);
810 this.cleanupPool = true;
811 }
812 }
813 }
814 return this.batchPool;
815 }
816
817 private ExecutorService getThreadPool(int maxThreads, int coreThreads, String nameHint,
818 BlockingQueue<Runnable> passedWorkQueue) {
819
820 if (maxThreads == 0) {
821 maxThreads = Runtime.getRuntime().availableProcessors() * 8;
822 }
823 if (coreThreads == 0) {
824 coreThreads = Runtime.getRuntime().availableProcessors() * 8;
825 }
826 long keepAliveTime = conf.getLong("hbase.hconnection.threads.keepalivetime", 60);
827 BlockingQueue<Runnable> workQueue = passedWorkQueue;
828 if (workQueue == null) {
829 workQueue =
830 new LinkedBlockingQueue<Runnable>(maxThreads *
831 conf.getInt(HConstants.HBASE_CLIENT_MAX_TOTAL_TASKS,
832 HConstants.DEFAULT_HBASE_CLIENT_MAX_TOTAL_TASKS));
833 }
834 ThreadPoolExecutor tpe = new ThreadPoolExecutor(
835 coreThreads,
836 maxThreads,
837 keepAliveTime,
838 TimeUnit.SECONDS,
839 workQueue,
840 Threads.newDaemonThreadFactory(toString() + nameHint));
841 tpe.allowCoreThreadTimeOut(true);
842 return tpe;
843 }
844
845 private ExecutorService getMetaLookupPool() {
846 if (this.metaLookupPool == null) {
847 synchronized (this) {
848 if (this.metaLookupPool == null) {
849
850
851
852
853 this.metaLookupPool = getThreadPool(
854 conf.getInt("hbase.hconnection.meta.lookup.threads.max", 128),
855 conf.getInt("hbase.hconnection.meta.lookup.threads.core", 10),
856 "-metaLookup-shared-", new LinkedBlockingQueue<Runnable>());
857 }
858 }
859 }
860 return this.metaLookupPool;
861 }
862
863 protected ExecutorService getCurrentMetaLookupPool() {
864 return metaLookupPool;
865 }
866
867 protected ExecutorService getCurrentBatchPool() {
868 return batchPool;
869 }
870
871 private void shutdownPools() {
872 if (this.cleanupPool && this.batchPool != null && !this.batchPool.isShutdown()) {
873 shutdownBatchPool(this.batchPool);
874 }
875 if (this.metaLookupPool != null && !this.metaLookupPool.isShutdown()) {
876 shutdownBatchPool(this.metaLookupPool);
877 }
878 }
879
880 private void shutdownBatchPool(ExecutorService pool) {
881 pool.shutdown();
882 try {
883 if (!pool.awaitTermination(10, TimeUnit.SECONDS)) {
884 pool.shutdownNow();
885 }
886 } catch (InterruptedException e) {
887 pool.shutdownNow();
888 }
889 }
890
891
892
893
894
895 private Registry setupRegistry() throws IOException {
896 return RegistryFactory.getRegistry(this);
897 }
898
899
900
901
902 @VisibleForTesting
903 RpcClient getRpcClient() {
904 return rpcClient;
905 }
906
907
908
909
910 @Override
911 public String toString(){
912 return "hconnection-0x" + Integer.toHexString(hashCode());
913 }
914
915 protected String clusterId = null;
916
917 void retrieveClusterId() {
918 if (clusterId != null) return;
919 this.clusterId = this.registry.getClusterId();
920 if (clusterId == null) {
921 clusterId = HConstants.CLUSTER_ID_DEFAULT;
922 LOG.debug("clusterid came back null, using default " + clusterId);
923 }
924 }
925
926 @Override
927 public Configuration getConfiguration() {
928 return this.conf;
929 }
930
931 private void checkIfBaseNodeAvailable(ZooKeeperWatcher zkw)
932 throws MasterNotRunningException {
933 String errorMsg;
934 try {
935 if (ZKUtil.checkExists(zkw, zkw.baseZNode) == -1) {
936 errorMsg = "The node " + zkw.baseZNode+" is not in ZooKeeper. "
937 + "It should have been written by the master. "
938 + "Check the value configured in 'zookeeper.znode.parent'. "
939 + "There could be a mismatch with the one configured in the master.";
940 LOG.error(errorMsg);
941 throw new MasterNotRunningException(errorMsg);
942 }
943 } catch (KeeperException e) {
944 errorMsg = "Can't get connection to ZooKeeper: " + e.getMessage();
945 LOG.error(errorMsg);
946 throw new MasterNotRunningException(errorMsg, e);
947 }
948 }
949
950
951
952
953
954
955 @Deprecated
956 @Override
957 public boolean isMasterRunning()
958 throws MasterNotRunningException, ZooKeeperConnectionException {
959
960
961
962 MasterKeepAliveConnection m = getKeepAliveMasterService();
963 m.close();
964 return true;
965 }
966
967 @Override
968 public HRegionLocation getRegionLocation(final TableName tableName,
969 final byte [] row, boolean reload)
970 throws IOException {
971 return reload? relocateRegion(tableName, row): locateRegion(tableName, row);
972 }
973
974 @Override
975 public HRegionLocation getRegionLocation(final byte[] tableName,
976 final byte [] row, boolean reload)
977 throws IOException {
978 return getRegionLocation(TableName.valueOf(tableName), row, reload);
979 }
980
981 @Override
982 public boolean isTableEnabled(TableName tableName) throws IOException {
983 return this.registry.isTableOnlineState(tableName, true);
984 }
985
986 @Override
987 public boolean isTableEnabled(byte[] tableName) throws IOException {
988 return isTableEnabled(TableName.valueOf(tableName));
989 }
990
991 @Override
992 public boolean isTableDisabled(TableName tableName) throws IOException {
993 return this.registry.isTableOnlineState(tableName, false);
994 }
995
996 @Override
997 public boolean isTableDisabled(byte[] tableName) throws IOException {
998 return isTableDisabled(TableName.valueOf(tableName));
999 }
1000
1001 @Override
1002 public boolean isTableAvailable(final TableName tableName) throws IOException {
1003 final AtomicBoolean available = new AtomicBoolean(true);
1004 final AtomicInteger regionCount = new AtomicInteger(0);
1005 MetaScannerVisitor visitor = new MetaScannerVisitorBase() {
1006 @Override
1007 public boolean processRow(Result row) throws IOException {
1008 HRegionInfo info = MetaScanner.getHRegionInfo(row);
1009 if (info != null && !info.isSplitParent()) {
1010 if (tableName.equals(info.getTable())) {
1011 ServerName server = HRegionInfo.getServerName(row);
1012 if (server == null) {
1013 available.set(false);
1014 return false;
1015 }
1016 regionCount.incrementAndGet();
1017 } else if (tableName.compareTo(info.getTable()) < 0) {
1018
1019 return false;
1020 }
1021 }
1022 return true;
1023 }
1024 };
1025 MetaScanner.metaScan(this, visitor, tableName);
1026 return available.get() && (regionCount.get() > 0);
1027 }
1028
1029 @Override
1030 public boolean isTableAvailable(final byte[] tableName) throws IOException {
1031 return isTableAvailable(TableName.valueOf(tableName));
1032 }
1033
1034 @Override
1035 public boolean isTableAvailable(final TableName tableName, final byte[][] splitKeys)
1036 throws IOException {
1037 final AtomicBoolean available = new AtomicBoolean(true);
1038 final AtomicInteger regionCount = new AtomicInteger(0);
1039 MetaScannerVisitor visitor = new MetaScannerVisitorBase() {
1040 @Override
1041 public boolean processRow(Result row) throws IOException {
1042 HRegionInfo info = MetaScanner.getHRegionInfo(row);
1043 if (info != null && !info.isSplitParent()) {
1044 if (tableName.equals(info.getTable())) {
1045 ServerName server = HRegionInfo.getServerName(row);
1046 if (server == null) {
1047 available.set(false);
1048 return false;
1049 }
1050 if (!Bytes.equals(info.getStartKey(), HConstants.EMPTY_BYTE_ARRAY)) {
1051 for (byte[] splitKey : splitKeys) {
1052
1053 if (Bytes.equals(info.getStartKey(), splitKey)) {
1054 regionCount.incrementAndGet();
1055 break;
1056 }
1057 }
1058 } else {
1059
1060 regionCount.incrementAndGet();
1061 }
1062 } else if (tableName.compareTo(info.getTable()) < 0) {
1063
1064 return false;
1065 }
1066 }
1067 return true;
1068 }
1069 };
1070 MetaScanner.metaScan(this, visitor, tableName);
1071
1072 return available.get() && (regionCount.get() == splitKeys.length + 1);
1073 }
1074
1075 @Override
1076 public boolean isTableAvailable(final byte[] tableName, final byte[][] splitKeys)
1077 throws IOException {
1078 return isTableAvailable(TableName.valueOf(tableName), splitKeys);
1079 }
1080
1081 @Override
1082 public HRegionLocation locateRegion(final byte[] regionName) throws IOException {
1083 RegionLocations locations = locateRegion(HRegionInfo.getTable(regionName),
1084 HRegionInfo.getStartKey(regionName), false, true);
1085 return locations == null ? null : locations.getRegionLocation();
1086 }
1087
1088 @Override
1089 public boolean isDeadServer(ServerName sn) {
1090 if (clusterStatusListener == null) {
1091 return false;
1092 } else {
1093 return clusterStatusListener.isDeadServer(sn);
1094 }
1095 }
1096
1097 @Override
1098 public List<HRegionLocation> locateRegions(final TableName tableName)
1099 throws IOException {
1100 return locateRegions (tableName, false, true);
1101 }
1102
1103 @Override
1104 public List<HRegionLocation> locateRegions(final byte[] tableName)
1105 throws IOException {
1106 return locateRegions(TableName.valueOf(tableName));
1107 }
1108
1109 @Override
1110 public List<HRegionLocation> locateRegions(final TableName tableName,
1111 final boolean useCache, final boolean offlined) throws IOException {
1112 NavigableMap<HRegionInfo, ServerName> regions = MetaScanner.allTableRegions(this, tableName);
1113 final List<HRegionLocation> locations = new ArrayList<HRegionLocation>();
1114 for (HRegionInfo regionInfo : regions.keySet()) {
1115 RegionLocations list = locateRegion(tableName, regionInfo.getStartKey(), useCache, true);
1116 if (list != null) {
1117 for (HRegionLocation loc : list.getRegionLocations()) {
1118 if (loc != null) {
1119 locations.add(loc);
1120 }
1121 }
1122 }
1123 }
1124 return locations;
1125 }
1126
1127 @Override
1128 public List<HRegionLocation> locateRegions(final byte[] tableName,
1129 final boolean useCache, final boolean offlined) throws IOException {
1130 return locateRegions(TableName.valueOf(tableName), useCache, offlined);
1131 }
1132
1133 @Override
1134 public HRegionLocation locateRegion(
1135 final TableName tableName, final byte[] row) throws IOException{
1136 RegionLocations locations = locateRegion(tableName, row, true, true);
1137 return locations == null ? null : locations.getRegionLocation();
1138 }
1139
1140 @Override
1141 public HRegionLocation locateRegion(final byte[] tableName,
1142 final byte [] row)
1143 throws IOException{
1144 return locateRegion(TableName.valueOf(tableName), row);
1145 }
1146
1147 @Override
1148 public HRegionLocation relocateRegion(final TableName tableName,
1149 final byte [] row) throws IOException{
1150 RegionLocations locations = relocateRegion(tableName, row,
1151 RegionReplicaUtil.DEFAULT_REPLICA_ID);
1152 return locations == null ? null :
1153 locations.getRegionLocation(RegionReplicaUtil.DEFAULT_REPLICA_ID);
1154 }
1155
1156 @Override
1157 public RegionLocations relocateRegion(final TableName tableName,
1158 final byte [] row, int replicaId) throws IOException{
1159
1160
1161
1162 if (!tableName.equals(TableName.META_TABLE_NAME) && isTableDisabled(tableName)) {
1163 throw new TableNotEnabledException(tableName.getNameAsString() + " is disabled.");
1164 }
1165
1166 return locateRegion(tableName, row, false, true, replicaId);
1167 }
1168
1169 @Override
1170 public HRegionLocation relocateRegion(final byte[] tableName,
1171 final byte [] row) throws IOException {
1172 return relocateRegion(TableName.valueOf(tableName), row);
1173 }
1174
1175 @Override
1176 public RegionLocations locateRegion(final TableName tableName,
1177 final byte [] row, boolean useCache, boolean retry)
1178 throws IOException {
1179 return locateRegion(tableName, row, useCache, retry, RegionReplicaUtil.DEFAULT_REPLICA_ID);
1180 }
1181
1182 @Override
1183 public RegionLocations locateRegion(final TableName tableName,
1184 final byte [] row, boolean useCache, boolean retry, int replicaId)
1185 throws IOException {
1186 if (this.closed) throw new IOException(toString() + " closed");
1187 if (tableName== null || tableName.getName().length == 0) {
1188 throw new IllegalArgumentException(
1189 "table name cannot be null or zero length");
1190 }
1191 if (tableName.equals(TableName.META_TABLE_NAME)) {
1192 if (useMetaReplicas) {
1193 return locateMeta(tableName, useCache, replicaId);
1194 } else {
1195 return this.registry.getMetaRegionLocation();
1196 }
1197 } else {
1198
1199 return locateRegionInMeta(tableName, row, useCache, retry, replicaId);
1200 }
1201 }
1202
1203
1204
1205
1206
1207
1208 private void prefetchRegionCache(final TableName tableName,
1209 final byte[] row) {
1210
1211
1212 MetaScannerVisitor visitor = new MetaScannerVisitorBase() {
1213 public boolean processRow(Result result) throws IOException {
1214 try {
1215 HRegionInfo regionInfo = MetaScanner.getHRegionInfo(result);
1216 if (regionInfo == null) {
1217 return true;
1218 }
1219
1220
1221 if (!regionInfo.getTable().equals(tableName)) {
1222 return false;
1223 }
1224 if (regionInfo.isOffline()) {
1225
1226 return true;
1227 }
1228
1229 ServerName serverName = HRegionInfo.getServerName(result);
1230 if (serverName == null) {
1231 return true;
1232 }
1233
1234 long seqNum = HRegionInfo.getSeqNumDuringOpen(result);
1235 HRegionLocation loc = new HRegionLocation(regionInfo, serverName, seqNum);
1236
1237 cacheLocation(tableName, null, loc);
1238 return true;
1239 } catch (RuntimeException e) {
1240 throw new IOException(e);
1241 }
1242 }
1243 };
1244 try {
1245
1246 MetaScanner.metaScan(this, visitor, tableName, row,
1247 this.prefetchRegionLimit, TableName.META_TABLE_NAME);
1248 } catch (IOException e) {
1249 if (ExceptionUtil.isInterrupt(e)) {
1250 Thread.currentThread().interrupt();
1251 } else {
1252 LOG.warn("Encountered problems when prefetch hbase:meta table: ", e);
1253 }
1254 }
1255 }
1256
1257 private RegionLocations locateMeta(final TableName tableName,
1258 boolean useCache, int replicaId) throws IOException {
1259
1260
1261
1262 byte[] metaCacheKey = HConstants.EMPTY_START_ROW;
1263 RegionLocations locations = null;
1264 if (useCache) {
1265 locations = getCachedLocation(tableName, metaCacheKey);
1266 if (locations != null && locations.getRegionLocation(replicaId) != null) {
1267 return locations;
1268 }
1269 }
1270
1271
1272 synchronized (metaRegionLock) {
1273
1274
1275 if (useCache) {
1276 locations = getCachedLocation(tableName, metaCacheKey);
1277 if (locations != null && locations.getRegionLocation(replicaId) != null) {
1278 return locations;
1279 }
1280 }
1281
1282
1283 locations = this.registry.getMetaRegionLocation();
1284 if (locations != null) {
1285 cacheLocation(tableName, locations);
1286 }
1287 }
1288 return locations;
1289 }
1290
1291
1292
1293
1294
1295 private RegionLocations locateRegionInMeta(TableName tableName, byte[] row,
1296 boolean useCache, boolean retry, int replicaId) throws IOException {
1297
1298
1299
1300 if (useCache) {
1301 RegionLocations locations = getCachedLocation(tableName, row);
1302 if (locations != null && locations.getRegionLocation(replicaId) != null) {
1303 return locations;
1304 }
1305 }
1306
1307
1308
1309
1310 byte[] metaKey = HRegionInfo.createRegionName(tableName, row, HConstants.NINES, false);
1311
1312 Scan s = null;
1313 if (useMetaReplicas) {
1314
1315
1316
1317 s = new Scan();
1318 s.setReversed(true);
1319 s.setStartRow(metaKey);
1320 s.setSmall(true);
1321 s.setCaching(1);
1322 s.setConsistency(Consistency.TIMELINE);
1323 }
1324
1325 int localNumRetries = (retry ? numTries : 1);
1326
1327 for (int tries = 0; true; tries++) {
1328 if (tries >= localNumRetries) {
1329 throw new NoServerForRegionException("Unable to find region for "
1330 + Bytes.toStringBinary(row) + " in " + tableName +
1331 " after " + localNumRetries + " tries.");
1332 }
1333 if (useCache) {
1334 RegionLocations locations = getCachedLocation(tableName, row);
1335 if (locations != null && locations.getRegionLocation(replicaId) != null) {
1336 return locations;
1337 }
1338 } else {
1339
1340
1341 metaCache.clearCache(tableName, row);
1342 }
1343
1344 RegionLocations metaLocation = null;
1345
1346 try {
1347 Result regionInfoRow = null;
1348 if (s != null) {
1349 ReversedClientScanner rcs = null;
1350 try {
1351 rcs = new ClientSmallReversedScanner(conf, s, TableName.META_TABLE_NAME, this,
1352 rpcCallerFactory, rpcControllerFactory, getMetaLookupPool(), 0);
1353 regionInfoRow = rcs.next();
1354 } finally {
1355 if (rcs != null) {
1356 rcs.close();
1357 }
1358 }
1359 }
1360
1361
1362
1363 if (regionInfoRow == null) {
1364
1365 metaLocation = locateRegion(TableName.META_TABLE_NAME, metaKey, false, false);
1366
1367 if (metaLocation == null) continue;
1368
1369 ClientService.BlockingInterface service =
1370 getClient(metaLocation.getDefaultRegionLocation().getServerName());
1371
1372
1373
1374
1375 if (useCache) {
1376 if (TableName.META_TABLE_NAME.equals(tableName) && getRegionCachePrefetch(tableName)) {
1377
1378
1379 RegionLocations locations = getCachedLocation(tableName, row);
1380 if (locations != null && locations.getRegionLocation(replicaId) != null) {
1381 return locations;
1382 }
1383
1384
1385 prefetchRegionCache(tableName, row);
1386 }
1387 RegionLocations locations = getCachedLocation(tableName, row);
1388 if (locations != null && locations.getRegionLocation(replicaId) != null) {
1389 return locations;
1390 }
1391 } else {
1392
1393
1394 metaCache.clearCache(tableName, row);
1395 }
1396
1397
1398 regionInfoRow = ProtobufUtil.getRowOrBefore(service,
1399 metaLocation.getDefaultRegionLocation().getRegionInfo().getRegionName(),
1400 metaKey, HConstants.CATALOG_FAMILY);
1401 }
1402
1403 if (regionInfoRow == null) {
1404 throw new TableNotFoundException(tableName);
1405 }
1406
1407
1408 RegionLocations locations = MetaTableAccessor.getRegionLocations(regionInfoRow);
1409 if (locations == null || locations.getRegionLocation(replicaId) == null) {
1410 throw new IOException("HRegionInfo was null in " +
1411 tableName + ", row=" + regionInfoRow);
1412 }
1413 HRegionInfo regionInfo = locations.getRegionLocation(replicaId).getRegionInfo();
1414 if (regionInfo == null) {
1415 throw new IOException("HRegionInfo was null or empty in " +
1416 TableName.META_TABLE_NAME + ", row=" + regionInfoRow);
1417 }
1418
1419
1420 if (!regionInfo.getTable().equals(tableName)) {
1421 throw new TableNotFoundException(
1422 "Table '" + tableName + "' was not found, got: " +
1423 regionInfo.getTable() + ".");
1424 }
1425 if (regionInfo.isSplit()) {
1426 throw new RegionOfflineException("the only available region for" +
1427 " the required row is a split parent," +
1428 " the daughters should be online soon: " +
1429 regionInfo.getRegionNameAsString());
1430 }
1431 if (regionInfo.isOffline()) {
1432 throw new RegionOfflineException("the region is offline, could" +
1433 " be caused by a disable table call: " +
1434 regionInfo.getRegionNameAsString());
1435 }
1436
1437 ServerName serverName = locations.getRegionLocation(replicaId).getServerName();
1438 if (serverName == null) {
1439 throw new NoServerForRegionException("No server address listed " +
1440 "in " + TableName.META_TABLE_NAME + " for region " +
1441 regionInfo.getRegionNameAsString() + " containing row " +
1442 Bytes.toStringBinary(row));
1443 }
1444
1445 if (isDeadServer(serverName)){
1446 throw new RegionServerStoppedException("hbase:meta says the region "+
1447 regionInfo.getRegionNameAsString()+" is managed by the server " + serverName +
1448 ", but it is dead.");
1449 }
1450
1451 cacheLocation(tableName, locations);
1452 return locations;
1453 } catch (TableNotFoundException e) {
1454
1455
1456
1457 throw e;
1458 } catch (IOException e) {
1459 ExceptionUtil.rethrowIfInterrupt(e);
1460
1461 if (e instanceof RemoteException) {
1462 e = ((RemoteException)e).unwrapRemoteException();
1463 }
1464 if (tries < localNumRetries - 1) {
1465 if (LOG.isDebugEnabled()) {
1466 LOG.debug("locateRegionInMeta parentTable=" +
1467 TableName.META_TABLE_NAME + ", metaLocation=" + metaLocation +
1468 ", attempt=" + tries + " of " +
1469 localNumRetries + " failed; retrying after sleep of " +
1470 ConnectionUtils.getPauseTime(this.pause, tries) + " because: " + e.getMessage());
1471 }
1472 } else {
1473 throw e;
1474 }
1475
1476 if(!(e instanceof RegionOfflineException ||
1477 e instanceof NoServerForRegionException)) {
1478 relocateRegion(TableName.META_TABLE_NAME, metaKey, replicaId);
1479 }
1480 }
1481 try{
1482 Thread.sleep(ConnectionUtils.getPauseTime(this.pause, tries));
1483 } catch (InterruptedException e) {
1484 throw new InterruptedIOException("Giving up trying to location region in " +
1485 "meta: thread is interrupted.");
1486 }
1487 }
1488 }
1489
1490
1491
1492
1493
1494
1495 private void cacheLocation(final TableName tableName, final RegionLocations location) {
1496 metaCache.cacheLocation(tableName, location);
1497 }
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507 RegionLocations getCachedLocation(final TableName tableName,
1508 final byte [] row) {
1509 return metaCache.getCachedLocation(tableName, row);
1510 }
1511
1512 public void clearRegionCache(final TableName tableName, byte[] row) {
1513 metaCache.clearCache(tableName, row);
1514 }
1515
1516
1517
1518
1519 @Override
1520 public void clearCaches(final ServerName serverName) {
1521 metaCache.clearCache(serverName);
1522 }
1523
1524 @Override
1525 public void clearRegionCache() {
1526 metaCache.clearCache();
1527 }
1528
1529 @Override
1530 public void clearRegionCache(final TableName tableName) {
1531 metaCache.clearCache(tableName);
1532 }
1533
1534 @Override
1535 public void clearRegionCache(final byte[] tableName) {
1536 clearRegionCache(TableName.valueOf(tableName));
1537 }
1538
1539
1540
1541
1542
1543
1544
1545 private void cacheLocation(final TableName tableName, final ServerName source,
1546 final HRegionLocation location) {
1547 metaCache.cacheLocation(tableName, source, location);
1548 }
1549
1550
1551 private final ConcurrentHashMap<String, Object> stubs =
1552 new ConcurrentHashMap<String, Object>();
1553
1554 private final ConcurrentHashMap<String, String> connectionLock =
1555 new ConcurrentHashMap<String, String>();
1556
1557
1558
1559
1560 static class MasterServiceState {
1561 HConnection connection;
1562 MasterService.BlockingInterface stub;
1563 int userCount;
1564
1565 MasterServiceState (final HConnection connection) {
1566 super();
1567 this.connection = connection;
1568 }
1569
1570 @Override
1571 public String toString() {
1572 return "MasterService";
1573 }
1574
1575 Object getStub() {
1576 return this.stub;
1577 }
1578
1579 void clearStub() {
1580 this.stub = null;
1581 }
1582
1583 boolean isMasterRunning() throws ServiceException {
1584 IsMasterRunningResponse response =
1585 this.stub.isMasterRunning(null, RequestConverter.buildIsMasterRunningRequest());
1586 return response != null? response.getIsMasterRunning(): false;
1587 }
1588 }
1589
1590
1591
1592
1593
1594
1595 abstract class StubMaker {
1596
1597
1598
1599 protected abstract String getServiceName();
1600
1601
1602
1603
1604
1605 protected abstract Object makeStub(final BlockingRpcChannel channel);
1606
1607
1608
1609
1610
1611 protected abstract void isMasterRunning() throws ServiceException;
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621 private Object makeStubNoRetries() throws IOException, KeeperException, ServiceException {
1622 ZooKeeperKeepAliveConnection zkw;
1623 try {
1624 zkw = getKeepAliveZooKeeperWatcher();
1625 } catch (IOException e) {
1626 ExceptionUtil.rethrowIfInterrupt(e);
1627 throw new ZooKeeperConnectionException("Can't connect to ZooKeeper", e);
1628 }
1629 try {
1630 checkIfBaseNodeAvailable(zkw);
1631 ServerName sn = MasterAddressTracker.getMasterAddress(zkw);
1632 if (sn == null) {
1633 String msg = "ZooKeeper available but no active master location found";
1634 LOG.info(msg);
1635 throw new MasterNotRunningException(msg);
1636 }
1637 if (isDeadServer(sn)) {
1638 throw new MasterNotRunningException(sn + " is dead.");
1639 }
1640
1641 String key = getStubKey(getServiceName(),
1642 sn.getHostname(), sn.getPort(), hostnamesCanChange);
1643 connectionLock.putIfAbsent(key, key);
1644 Object stub = null;
1645 synchronized (connectionLock.get(key)) {
1646 stub = stubs.get(key);
1647 if (stub == null) {
1648 BlockingRpcChannel channel = rpcClient.createBlockingRpcChannel(sn, user, rpcTimeout);
1649 stub = makeStub(channel);
1650 isMasterRunning();
1651 stubs.put(key, stub);
1652 }
1653 }
1654 return stub;
1655 } finally {
1656 zkw.close();
1657 }
1658 }
1659
1660
1661
1662
1663
1664
1665 Object makeStub() throws IOException {
1666
1667
1668 synchronized (masterAndZKLock) {
1669 Exception exceptionCaught = null;
1670 if (!closed) {
1671 try {
1672 return makeStubNoRetries();
1673 } catch (IOException e) {
1674 exceptionCaught = e;
1675 } catch (KeeperException e) {
1676 exceptionCaught = e;
1677 } catch (ServiceException e) {
1678 exceptionCaught = e;
1679 }
1680
1681 throw new MasterNotRunningException(exceptionCaught);
1682 } else {
1683 throw new DoNotRetryIOException("Connection was closed while trying to get master");
1684 }
1685 }
1686 }
1687 }
1688
1689
1690
1691
1692 class MasterServiceStubMaker extends StubMaker {
1693 private MasterService.BlockingInterface stub;
1694 @Override
1695 protected String getServiceName() {
1696 return MasterService.getDescriptor().getName();
1697 }
1698
1699 @Override
1700 MasterService.BlockingInterface makeStub() throws IOException {
1701 return (MasterService.BlockingInterface)super.makeStub();
1702 }
1703
1704 @Override
1705 protected Object makeStub(BlockingRpcChannel channel) {
1706 this.stub = MasterService.newBlockingStub(channel);
1707 return this.stub;
1708 }
1709
1710 @Override
1711 protected void isMasterRunning() throws ServiceException {
1712 this.stub.isMasterRunning(null, RequestConverter.buildIsMasterRunningRequest());
1713 }
1714 }
1715
1716 @Override
1717 public AdminService.BlockingInterface getAdmin(final ServerName serverName)
1718 throws IOException {
1719 return getAdmin(serverName, false);
1720 }
1721
1722 @Override
1723
1724 public AdminService.BlockingInterface getAdmin(final ServerName serverName,
1725 final boolean master)
1726 throws IOException {
1727 if (isDeadServer(serverName)) {
1728 throw new RegionServerStoppedException(serverName + " is dead.");
1729 }
1730 String key = getStubKey(AdminService.BlockingInterface.class.getName(),
1731 serverName.getHostname(), serverName.getPort(), this.hostnamesCanChange);
1732 this.connectionLock.putIfAbsent(key, key);
1733 AdminService.BlockingInterface stub = null;
1734 synchronized (this.connectionLock.get(key)) {
1735 stub = (AdminService.BlockingInterface)this.stubs.get(key);
1736 if (stub == null) {
1737 BlockingRpcChannel channel =
1738 this.rpcClient.createBlockingRpcChannel(serverName, user, rpcTimeout);
1739 stub = AdminService.newBlockingStub(channel);
1740 this.stubs.put(key, stub);
1741 }
1742 }
1743 return stub;
1744 }
1745
1746 @Override
1747 public ClientService.BlockingInterface getClient(final ServerName sn)
1748 throws IOException {
1749 if (isDeadServer(sn)) {
1750 throw new RegionServerStoppedException(sn + " is dead.");
1751 }
1752 String key = getStubKey(ClientService.BlockingInterface.class.getName(), sn.getHostname(),
1753 sn.getPort(), this.hostnamesCanChange);
1754 this.connectionLock.putIfAbsent(key, key);
1755 ClientService.BlockingInterface stub = null;
1756 synchronized (this.connectionLock.get(key)) {
1757 stub = (ClientService.BlockingInterface)this.stubs.get(key);
1758 if (stub == null) {
1759 BlockingRpcChannel channel =
1760 this.rpcClient.createBlockingRpcChannel(sn, user, rpcTimeout);
1761 stub = ClientService.newBlockingStub(channel);
1762
1763
1764 this.stubs.put(key, stub);
1765 }
1766 }
1767 return stub;
1768 }
1769
1770 static String getStubKey(final String serviceName,
1771 final String rsHostname,
1772 int port,
1773 boolean resolveHostnames) {
1774
1775
1776
1777
1778
1779
1780 String address = rsHostname;
1781 if (resolveHostnames) {
1782 InetAddress i = new InetSocketAddress(rsHostname, port).getAddress();
1783 if (i != null) {
1784 address = i.getHostAddress() + "-" + rsHostname;
1785 }
1786 }
1787 return serviceName + "@" + address + ":" + port;
1788 }
1789
1790 private ZooKeeperKeepAliveConnection keepAliveZookeeper;
1791 private AtomicInteger keepAliveZookeeperUserCount = new AtomicInteger(0);
1792 private boolean canCloseZKW = true;
1793
1794
1795 private static final long keepAlive = 5 * 60 * 1000;
1796
1797
1798
1799
1800
1801 ZooKeeperKeepAliveConnection getKeepAliveZooKeeperWatcher()
1802 throws IOException {
1803 synchronized (masterAndZKLock) {
1804 if (keepAliveZookeeper == null) {
1805 if (this.closed) {
1806 throw new IOException(toString() + " closed");
1807 }
1808
1809
1810 keepAliveZookeeper = new ZooKeeperKeepAliveConnection(conf, this.toString(), this);
1811 }
1812 keepAliveZookeeperUserCount.addAndGet(1);
1813 keepZooKeeperWatcherAliveUntil = Long.MAX_VALUE;
1814 return keepAliveZookeeper;
1815 }
1816 }
1817
1818 void releaseZooKeeperWatcher(final ZooKeeperWatcher zkw) {
1819 if (zkw == null){
1820 return;
1821 }
1822 if (keepAliveZookeeperUserCount.addAndGet(-1) <= 0 ){
1823 keepZooKeeperWatcherAliveUntil = System.currentTimeMillis() + keepAlive;
1824 }
1825 }
1826
1827 private void closeZooKeeperWatcher() {
1828 synchronized (masterAndZKLock) {
1829 if (keepAliveZookeeper != null) {
1830 LOG.info("Closing zookeeper sessionid=0x" +
1831 Long.toHexString(
1832 keepAliveZookeeper.getRecoverableZooKeeper().getSessionId()));
1833 keepAliveZookeeper.internalClose();
1834 keepAliveZookeeper = null;
1835 }
1836 keepAliveZookeeperUserCount.set(0);
1837 }
1838 }
1839
1840 final MasterServiceState masterServiceState = new MasterServiceState(this);
1841
1842 @Override
1843 public MasterService.BlockingInterface getMaster() throws MasterNotRunningException {
1844 return getKeepAliveMasterService();
1845 }
1846
1847 private void resetMasterServiceState(final MasterServiceState mss) {
1848 mss.userCount++;
1849 }
1850
1851 @Override
1852 public MasterKeepAliveConnection getKeepAliveMasterService()
1853 throws MasterNotRunningException {
1854 synchronized (masterAndZKLock) {
1855 if (!isKeepAliveMasterConnectedAndRunning(this.masterServiceState)) {
1856 MasterServiceStubMaker stubMaker = new MasterServiceStubMaker();
1857 try {
1858 this.masterServiceState.stub = stubMaker.makeStub();
1859 } catch (MasterNotRunningException ex) {
1860 throw ex;
1861 } catch (IOException e) {
1862
1863 throw new MasterNotRunningException(e);
1864 }
1865 }
1866 resetMasterServiceState(this.masterServiceState);
1867 }
1868
1869 final MasterService.BlockingInterface stub = this.masterServiceState.stub;
1870 return new MasterKeepAliveConnection() {
1871 MasterServiceState mss = masterServiceState;
1872 @Override
1873 public MasterProtos.AbortProcedureResponse abortProcedure(
1874 RpcController controller,
1875 MasterProtos.AbortProcedureRequest request) throws ServiceException {
1876 return stub.abortProcedure(controller, request);
1877 }
1878 @Override
1879 public MasterProtos.ListProceduresResponse listProcedures(
1880 RpcController controller,
1881 MasterProtos.ListProceduresRequest request) throws ServiceException {
1882 return stub.listProcedures(controller, request);
1883 }
1884 @Override
1885 public AddColumnResponse addColumn(RpcController controller, AddColumnRequest request)
1886 throws ServiceException {
1887 return stub.addColumn(controller, request);
1888 }
1889
1890 @Override
1891 public DeleteColumnResponse deleteColumn(RpcController controller,
1892 DeleteColumnRequest request)
1893 throws ServiceException {
1894 return stub.deleteColumn(controller, request);
1895 }
1896
1897 @Override
1898 public ModifyColumnResponse modifyColumn(RpcController controller,
1899 ModifyColumnRequest request)
1900 throws ServiceException {
1901 return stub.modifyColumn(controller, request);
1902 }
1903
1904 @Override
1905 public MoveRegionResponse moveRegion(RpcController controller,
1906 MoveRegionRequest request) throws ServiceException {
1907 return stub.moveRegion(controller, request);
1908 }
1909
1910 @Override
1911 public DispatchMergingRegionsResponse dispatchMergingRegions(
1912 RpcController controller, DispatchMergingRegionsRequest request)
1913 throws ServiceException {
1914 return stub.dispatchMergingRegions(controller, request);
1915 }
1916
1917 @Override
1918 public AssignRegionResponse assignRegion(RpcController controller,
1919 AssignRegionRequest request) throws ServiceException {
1920 return stub.assignRegion(controller, request);
1921 }
1922
1923 @Override
1924 public UnassignRegionResponse unassignRegion(RpcController controller,
1925 UnassignRegionRequest request) throws ServiceException {
1926 return stub.unassignRegion(controller, request);
1927 }
1928
1929 @Override
1930 public OfflineRegionResponse offlineRegion(RpcController controller,
1931 OfflineRegionRequest request) throws ServiceException {
1932 return stub.offlineRegion(controller, request);
1933 }
1934
1935 @Override
1936 public DeleteTableResponse deleteTable(RpcController controller,
1937 DeleteTableRequest request) throws ServiceException {
1938 return stub.deleteTable(controller, request);
1939 }
1940
1941 @Override
1942 public TruncateTableResponse truncateTable(RpcController controller,
1943 TruncateTableRequest request) throws ServiceException {
1944 return stub.truncateTable(controller, request);
1945 }
1946
1947 @Override
1948 public EnableTableResponse enableTable(RpcController controller,
1949 EnableTableRequest request) throws ServiceException {
1950 return stub.enableTable(controller, request);
1951 }
1952
1953 @Override
1954 public DisableTableResponse disableTable(RpcController controller,
1955 DisableTableRequest request) throws ServiceException {
1956 return stub.disableTable(controller, request);
1957 }
1958
1959 @Override
1960 public ModifyTableResponse modifyTable(RpcController controller,
1961 ModifyTableRequest request) throws ServiceException {
1962 return stub.modifyTable(controller, request);
1963 }
1964
1965 @Override
1966 public CreateTableResponse createTable(RpcController controller,
1967 CreateTableRequest request) throws ServiceException {
1968 return stub.createTable(controller, request);
1969 }
1970
1971 @Override
1972 public ShutdownResponse shutdown(RpcController controller,
1973 ShutdownRequest request) throws ServiceException {
1974 return stub.shutdown(controller, request);
1975 }
1976
1977 @Override
1978 public StopMasterResponse stopMaster(RpcController controller,
1979 StopMasterRequest request) throws ServiceException {
1980 return stub.stopMaster(controller, request);
1981 }
1982
1983 @Override
1984 public BalanceResponse balance(RpcController controller,
1985 BalanceRequest request) throws ServiceException {
1986 return stub.balance(controller, request);
1987 }
1988
1989 @Override
1990 public SetBalancerRunningResponse setBalancerRunning(
1991 RpcController controller, SetBalancerRunningRequest request)
1992 throws ServiceException {
1993 return stub.setBalancerRunning(controller, request);
1994 }
1995
1996 @Override
1997 public NormalizeResponse normalize(RpcController controller,
1998 NormalizeRequest request) throws ServiceException {
1999 return stub.normalize(controller, request);
2000 }
2001
2002 @Override
2003 public SetNormalizerRunningResponse setNormalizerRunning(
2004 RpcController controller, SetNormalizerRunningRequest request)
2005 throws ServiceException {
2006 return stub.setNormalizerRunning(controller, request);
2007 }
2008
2009 @Override
2010 public RunCatalogScanResponse runCatalogScan(RpcController controller,
2011 RunCatalogScanRequest request) throws ServiceException {
2012 return stub.runCatalogScan(controller, request);
2013 }
2014
2015 @Override
2016 public EnableCatalogJanitorResponse enableCatalogJanitor(
2017 RpcController controller, EnableCatalogJanitorRequest request)
2018 throws ServiceException {
2019 return stub.enableCatalogJanitor(controller, request);
2020 }
2021
2022 @Override
2023 public IsCatalogJanitorEnabledResponse isCatalogJanitorEnabled(
2024 RpcController controller, IsCatalogJanitorEnabledRequest request)
2025 throws ServiceException {
2026 return stub.isCatalogJanitorEnabled(controller, request);
2027 }
2028
2029 @Override
2030 public CoprocessorServiceResponse execMasterService(
2031 RpcController controller, CoprocessorServiceRequest request)
2032 throws ServiceException {
2033 return stub.execMasterService(controller, request);
2034 }
2035
2036 @Override
2037 public SnapshotResponse snapshot(RpcController controller,
2038 SnapshotRequest request) throws ServiceException {
2039 return stub.snapshot(controller, request);
2040 }
2041
2042 @Override
2043 public GetCompletedSnapshotsResponse getCompletedSnapshots(
2044 RpcController controller, GetCompletedSnapshotsRequest request)
2045 throws ServiceException {
2046 return stub.getCompletedSnapshots(controller, request);
2047 }
2048
2049 @Override
2050 public DeleteSnapshotResponse deleteSnapshot(RpcController controller,
2051 DeleteSnapshotRequest request) throws ServiceException {
2052 return stub.deleteSnapshot(controller, request);
2053 }
2054
2055 @Override
2056 public IsSnapshotDoneResponse isSnapshotDone(RpcController controller,
2057 IsSnapshotDoneRequest request) throws ServiceException {
2058 return stub.isSnapshotDone(controller, request);
2059 }
2060
2061 @Override
2062 public RestoreSnapshotResponse restoreSnapshot(
2063 RpcController controller, RestoreSnapshotRequest request)
2064 throws ServiceException {
2065 return stub.restoreSnapshot(controller, request);
2066 }
2067
2068 @Override
2069 public IsRestoreSnapshotDoneResponse isRestoreSnapshotDone(
2070 RpcController controller, IsRestoreSnapshotDoneRequest request)
2071 throws ServiceException {
2072 return stub.isRestoreSnapshotDone(controller, request);
2073 }
2074
2075 @Override
2076 public ExecProcedureResponse execProcedure(
2077 RpcController controller, ExecProcedureRequest request)
2078 throws ServiceException {
2079 return stub.execProcedure(controller, request);
2080 }
2081
2082 @Override
2083 public ExecProcedureResponse execProcedureWithRet(
2084 RpcController controller, ExecProcedureRequest request)
2085 throws ServiceException {
2086 return stub.execProcedureWithRet(controller, request);
2087 }
2088
2089 @Override
2090 public IsProcedureDoneResponse isProcedureDone(RpcController controller,
2091 IsProcedureDoneRequest request) throws ServiceException {
2092 return stub.isProcedureDone(controller, request);
2093 }
2094
2095 @Override
2096 public GetProcedureResultResponse getProcedureResult(RpcController controller,
2097 GetProcedureResultRequest request) throws ServiceException {
2098 return stub.getProcedureResult(controller, request);
2099 }
2100
2101 @Override
2102 public IsMasterRunningResponse isMasterRunning(
2103 RpcController controller, IsMasterRunningRequest request)
2104 throws ServiceException {
2105 return stub.isMasterRunning(controller, request);
2106 }
2107
2108 @Override
2109 public ModifyNamespaceResponse modifyNamespace(RpcController controller,
2110 ModifyNamespaceRequest request)
2111 throws ServiceException {
2112 return stub.modifyNamespace(controller, request);
2113 }
2114
2115 @Override
2116 public CreateNamespaceResponse createNamespace(
2117 RpcController controller, CreateNamespaceRequest request) throws ServiceException {
2118 return stub.createNamespace(controller, request);
2119 }
2120
2121 @Override
2122 public DeleteNamespaceResponse deleteNamespace(
2123 RpcController controller, DeleteNamespaceRequest request) throws ServiceException {
2124 return stub.deleteNamespace(controller, request);
2125 }
2126
2127 @Override
2128 public GetNamespaceDescriptorResponse getNamespaceDescriptor(RpcController controller,
2129 GetNamespaceDescriptorRequest request) throws ServiceException {
2130 return stub.getNamespaceDescriptor(controller, request);
2131 }
2132
2133 @Override
2134 public ListNamespaceDescriptorsResponse listNamespaceDescriptors(RpcController controller,
2135 ListNamespaceDescriptorsRequest request) throws ServiceException {
2136 return stub.listNamespaceDescriptors(controller, request);
2137 }
2138
2139 @Override
2140 public ListTableDescriptorsByNamespaceResponse listTableDescriptorsByNamespace(
2141 RpcController controller, ListTableDescriptorsByNamespaceRequest request)
2142 throws ServiceException {
2143 return stub.listTableDescriptorsByNamespace(controller, request);
2144 }
2145
2146 @Override
2147 public ListTableNamesByNamespaceResponse listTableNamesByNamespace(
2148 RpcController controller, ListTableNamesByNamespaceRequest request)
2149 throws ServiceException {
2150 return stub.listTableNamesByNamespace(controller, request);
2151 }
2152
2153 @Override
2154 public void close() {
2155 release(this.mss);
2156 }
2157
2158 @Override
2159 public GetSchemaAlterStatusResponse getSchemaAlterStatus(
2160 RpcController controller, GetSchemaAlterStatusRequest request)
2161 throws ServiceException {
2162 return stub.getSchemaAlterStatus(controller, request);
2163 }
2164
2165 @Override
2166 public GetTableDescriptorsResponse getTableDescriptors(
2167 RpcController controller, GetTableDescriptorsRequest request)
2168 throws ServiceException {
2169 return stub.getTableDescriptors(controller, request);
2170 }
2171
2172 @Override
2173 public GetTableNamesResponse getTableNames(
2174 RpcController controller, GetTableNamesRequest request)
2175 throws ServiceException {
2176 return stub.getTableNames(controller, request);
2177 }
2178
2179 @Override
2180 public GetClusterStatusResponse getClusterStatus(
2181 RpcController controller, GetClusterStatusRequest request)
2182 throws ServiceException {
2183 return stub.getClusterStatus(controller, request);
2184 }
2185
2186 @Override
2187 public SetQuotaResponse setQuota(RpcController controller, SetQuotaRequest request)
2188 throws ServiceException {
2189 return stub.setQuota(controller, request);
2190 }
2191
2192 @Override
2193 public MajorCompactionTimestampResponse getLastMajorCompactionTimestamp(
2194 RpcController controller, MajorCompactionTimestampRequest request)
2195 throws ServiceException {
2196 return stub.getLastMajorCompactionTimestamp(controller, request);
2197 }
2198
2199 @Override
2200 public MajorCompactionTimestampResponse getLastMajorCompactionTimestampForRegion(
2201 RpcController controller, MajorCompactionTimestampForRegionRequest request)
2202 throws ServiceException {
2203 return stub.getLastMajorCompactionTimestampForRegion(controller, request);
2204 }
2205
2206 @Override
2207 public IsBalancerEnabledResponse isBalancerEnabled(RpcController controller,
2208 IsBalancerEnabledRequest request) throws ServiceException {
2209 return stub.isBalancerEnabled(controller, request);
2210 }
2211
2212 @Override
2213 public IsNormalizerEnabledResponse isNormalizerEnabled(RpcController controller,
2214 IsNormalizerEnabledRequest request) throws ServiceException {
2215 return stub.isNormalizerEnabled(controller, request);
2216 }
2217
2218 @Override
2219 public SecurityCapabilitiesResponse getSecurityCapabilities(RpcController controller,
2220 SecurityCapabilitiesRequest request) throws ServiceException {
2221 return stub.getSecurityCapabilities(controller, request);
2222 }
2223 };
2224 }
2225
2226
2227 private static void release(MasterServiceState mss) {
2228 if (mss != null && mss.connection != null) {
2229 ((HConnectionImplementation)mss.connection).releaseMaster(mss);
2230 }
2231 }
2232
2233 private boolean isKeepAliveMasterConnectedAndRunning(MasterServiceState mss) {
2234 if (mss.getStub() == null){
2235 return false;
2236 }
2237 try {
2238 return mss.isMasterRunning();
2239 } catch (UndeclaredThrowableException e) {
2240
2241
2242 LOG.info("Master connection is not running anymore", e.getUndeclaredThrowable());
2243 return false;
2244 } catch (ServiceException se) {
2245 LOG.warn("Checking master connection", se);
2246 return false;
2247 }
2248 }
2249
2250 void releaseMaster(MasterServiceState mss) {
2251 if (mss.getStub() == null) return;
2252 synchronized (masterAndZKLock) {
2253 --mss.userCount;
2254 }
2255 }
2256
2257 private void closeMasterService(MasterServiceState mss) {
2258 if (mss.getStub() != null) {
2259 LOG.info("Closing master protocol: " + mss);
2260 mss.clearStub();
2261 }
2262 mss.userCount = 0;
2263 }
2264
2265
2266
2267
2268
2269 private void closeMaster() {
2270 synchronized (masterAndZKLock) {
2271 closeMasterService(masterServiceState);
2272 }
2273 }
2274
2275 void updateCachedLocation(HRegionInfo hri, ServerName source,
2276 ServerName serverName, long seqNum) {
2277 HRegionLocation newHrl = new HRegionLocation(hri, serverName, seqNum);
2278 cacheLocation(hri.getTable(), source, newHrl);
2279 }
2280
2281 @Override
2282 public void deleteCachedRegionLocation(final HRegionLocation location) {
2283 metaCache.clearCache(location);
2284 }
2285
2286 @Override
2287 public void updateCachedLocations(final TableName tableName, byte[] rowkey,
2288 final Object exception, final HRegionLocation source) {
2289 assert source != null;
2290 updateCachedLocations(tableName, source.getRegionInfo().getRegionName()
2291 , rowkey, exception, source.getServerName());
2292 }
2293
2294
2295
2296
2297
2298
2299
2300
2301
2302 @Override
2303 public void updateCachedLocations(final TableName tableName, byte[] regionName, byte[] rowkey,
2304 final Object exception, final ServerName source) {
2305 if (rowkey == null || tableName == null) {
2306 LOG.warn("Coding error, see method javadoc. row=" + (rowkey == null ? "null" : rowkey) +
2307 ", tableName=" + (tableName == null ? "null" : tableName));
2308 return;
2309 }
2310
2311 if (source == null) {
2312
2313 return;
2314 }
2315
2316 if (regionName == null) {
2317
2318 metaCache.clearCache(tableName, rowkey, source);
2319 return;
2320 }
2321
2322
2323 final RegionLocations oldLocations = getCachedLocation(tableName, rowkey);
2324 HRegionLocation oldLocation = null;
2325 if (oldLocations != null) {
2326 oldLocation = oldLocations.getRegionLocationByRegionName(regionName);
2327 }
2328 if (oldLocation == null || !source.equals(oldLocation.getServerName())) {
2329
2330
2331 return;
2332 }
2333
2334 HRegionInfo regionInfo = oldLocation.getRegionInfo();
2335 Throwable cause = ClientExceptionsUtil.findException(exception);
2336 if (cause != null) {
2337 if (!ClientExceptionsUtil.isMetaClearingException(cause)) {
2338
2339 return;
2340 }
2341
2342 if (cause instanceof RegionMovedException) {
2343 RegionMovedException rme = (RegionMovedException) cause;
2344 if (LOG.isTraceEnabled()) {
2345 LOG.trace("Region " + regionInfo.getRegionNameAsString() + " moved to " +
2346 rme.getHostname() + ":" + rme.getPort() +
2347 " according to " + source.getHostAndPort());
2348 }
2349
2350
2351 updateCachedLocation(
2352 regionInfo, source, rme.getServerName(), rme.getLocationSeqNum());
2353 return;
2354 }
2355 }
2356
2357
2358
2359 metaCache.clearCache(regionInfo);
2360 }
2361
2362 @Override
2363 public void updateCachedLocations(final byte[] tableName, byte[] rowkey,
2364 final Object exception, final HRegionLocation source) {
2365 updateCachedLocations(TableName.valueOf(tableName), rowkey, exception, source);
2366 }
2367
2368 @Override
2369 @Deprecated
2370 public void processBatch(List<? extends Row> list,
2371 final TableName tableName,
2372 ExecutorService pool,
2373 Object[] results) throws IOException, InterruptedException {
2374
2375
2376
2377 if (results.length != list.size()) {
2378 throw new IllegalArgumentException(
2379 "argument results must be the same size as argument list");
2380 }
2381 processBatchCallback(list, tableName, pool, results, null);
2382 }
2383
2384 @Override
2385 @Deprecated
2386 public void processBatch(List<? extends Row> list,
2387 final byte[] tableName,
2388 ExecutorService pool,
2389 Object[] results) throws IOException, InterruptedException {
2390 processBatch(list, TableName.valueOf(tableName), pool, results);
2391 }
2392
2393
2394
2395
2396
2397
2398
2399
2400 @Override
2401 @Deprecated
2402 public <R> void processBatchCallback(
2403 List<? extends Row> list,
2404 TableName tableName,
2405 ExecutorService pool,
2406 Object[] results,
2407 Batch.Callback<R> callback)
2408 throws IOException, InterruptedException {
2409
2410 AsyncRequestFuture ars = this.asyncProcess.submitAll(
2411 pool, tableName, list, callback, results);
2412 ars.waitUntilDone();
2413 if (ars.hasError()) {
2414 throw ars.getErrors();
2415 }
2416 }
2417
2418 @Override
2419 @Deprecated
2420 public <R> void processBatchCallback(
2421 List<? extends Row> list,
2422 byte[] tableName,
2423 ExecutorService pool,
2424 Object[] results,
2425 Batch.Callback<R> callback)
2426 throws IOException, InterruptedException {
2427 processBatchCallback(list, TableName.valueOf(tableName), pool, results, callback);
2428 }
2429
2430
2431 protected AsyncProcess createAsyncProcess(Configuration conf) {
2432
2433 return new AsyncProcess(this, conf, this.batchPool,
2434 RpcRetryingCallerFactory.instantiate(conf, this.getStatisticsTracker()), false,
2435 RpcControllerFactory.instantiate(conf));
2436 }
2437
2438 @Override
2439 public AsyncProcess getAsyncProcess() {
2440 return asyncProcess;
2441 }
2442
2443 @Override
2444 public ServerStatisticTracker getStatisticsTracker() {
2445 return this.stats;
2446 }
2447
2448 @Override
2449 public ClientBackoffPolicy getBackoffPolicy() {
2450 return this.backoffPolicy;
2451 }
2452
2453
2454
2455
2456
2457 @VisibleForTesting
2458 int getNumberOfCachedRegionLocations(final TableName tableName) {
2459 return metaCache.getNumberOfCachedRegionLocations(tableName);
2460 }
2461
2462 @Override
2463 @Deprecated
2464 public void setRegionCachePrefetch(final TableName tableName, final boolean enable) {
2465 if (!enable) {
2466 regionCachePrefetchDisabledTables.add(Bytes.mapKey(tableName.getName()));
2467 }
2468 else {
2469 regionCachePrefetchDisabledTables.remove(Bytes.mapKey(tableName.getName()));
2470 }
2471 }
2472
2473 @Override
2474 @Deprecated
2475 public void setRegionCachePrefetch(final byte[] tableName,
2476 final boolean enable) {
2477 setRegionCachePrefetch(TableName.valueOf(tableName), enable);
2478 }
2479
2480 @Override
2481 @Deprecated
2482 public boolean getRegionCachePrefetch(TableName tableName) {
2483 return usePrefetch &&
2484 !regionCachePrefetchDisabledTables.contains(Bytes.mapKey(tableName.getName()));
2485 }
2486
2487 @Override
2488 @Deprecated
2489 public boolean getRegionCachePrefetch(byte[] tableName) {
2490 return getRegionCachePrefetch(TableName.valueOf(tableName));
2491 }
2492
2493 @Override
2494 public void abort(final String msg, Throwable t) {
2495 if (t instanceof KeeperException.SessionExpiredException
2496 && keepAliveZookeeper != null) {
2497 synchronized (masterAndZKLock) {
2498 if (keepAliveZookeeper != null) {
2499 LOG.warn("This client just lost it's session with ZooKeeper," +
2500 " closing it." +
2501 " It will be recreated next time someone needs it", t);
2502 closeZooKeeperWatcher();
2503 }
2504 }
2505 } else {
2506 if (t != null) {
2507 LOG.fatal(msg, t);
2508 } else {
2509 LOG.fatal(msg);
2510 }
2511 this.aborted = true;
2512 close();
2513 this.closed = true;
2514 }
2515 }
2516
2517 @Override
2518 public boolean isClosed() {
2519 return this.closed;
2520 }
2521
2522 @Override
2523 public boolean isAborted(){
2524 return this.aborted;
2525 }
2526
2527 @Override
2528 public int getCurrentNrHRS() throws IOException {
2529 return this.registry.getCurrentNrHRS();
2530 }
2531
2532
2533
2534
2535 void incCount() {
2536 ++refCount;
2537 }
2538
2539
2540
2541
2542 void decCount() {
2543 if (refCount > 0) {
2544 --refCount;
2545 }
2546 }
2547
2548
2549
2550
2551
2552
2553 boolean isZeroReference() {
2554 return refCount == 0;
2555 }
2556
2557 void internalClose() {
2558 if (this.closed) {
2559 return;
2560 }
2561 closeMaster();
2562 shutdownPools();
2563 if (this.metrics != null) {
2564 this.metrics.shutdown();
2565 }
2566 this.closed = true;
2567 closeZooKeeperWatcher();
2568 this.stubs.clear();
2569 if (clusterStatusListener != null) {
2570 clusterStatusListener.close();
2571 }
2572 if (rpcClient != null) {
2573 rpcClient.close();
2574 }
2575 }
2576
2577 @Override
2578 public void close() {
2579 if (managed) {
2580 if (aborted) {
2581 ConnectionManager.deleteStaleConnection(this);
2582 } else {
2583 ConnectionManager.deleteConnection(this, false);
2584 }
2585 } else {
2586 internalClose();
2587 }
2588 }
2589
2590
2591
2592
2593
2594
2595
2596
2597
2598
2599
2600
2601 @Override
2602 protected void finalize() throws Throwable {
2603 super.finalize();
2604
2605 refCount = 1;
2606 close();
2607 }
2608
2609
2610
2611
2612 @Deprecated
2613 @Override
2614 public HTableDescriptor[] listTables() throws IOException {
2615 MasterKeepAliveConnection master = getKeepAliveMasterService();
2616 try {
2617 GetTableDescriptorsRequest req =
2618 RequestConverter.buildGetTableDescriptorsRequest((List<TableName>)null);
2619 return ProtobufUtil.getHTableDescriptorArray(master.getTableDescriptors(null, req));
2620 } catch (ServiceException se) {
2621 throw ProtobufUtil.getRemoteException(se);
2622 } finally {
2623 master.close();
2624 }
2625 }
2626
2627
2628
2629
2630 @Deprecated
2631 @Override
2632 public String[] getTableNames() throws IOException {
2633 TableName[] tableNames = listTableNames();
2634 String result[] = new String[tableNames.length];
2635 for (int i = 0; i < tableNames.length; i++) {
2636 result[i] = tableNames[i].getNameAsString();
2637 }
2638 return result;
2639 }
2640
2641
2642
2643
2644 @Deprecated
2645 @Override
2646 public TableName[] listTableNames() throws IOException {
2647 MasterKeepAliveConnection master = getKeepAliveMasterService();
2648 try {
2649 return ProtobufUtil.getTableNameArray(master.getTableNames(null,
2650 GetTableNamesRequest.newBuilder().build())
2651 .getTableNamesList());
2652 } catch (ServiceException se) {
2653 throw ProtobufUtil.getRemoteException(se);
2654 } finally {
2655 master.close();
2656 }
2657 }
2658
2659
2660
2661
2662 @Deprecated
2663 @Override
2664 public HTableDescriptor[] getHTableDescriptorsByTableName(
2665 List<TableName> tableNames) throws IOException {
2666 if (tableNames == null || tableNames.isEmpty()) return new HTableDescriptor[0];
2667 MasterKeepAliveConnection master = getKeepAliveMasterService();
2668 try {
2669 GetTableDescriptorsRequest req =
2670 RequestConverter.buildGetTableDescriptorsRequest(tableNames);
2671 return ProtobufUtil.getHTableDescriptorArray(master.getTableDescriptors(null, req));
2672 } catch (ServiceException se) {
2673 throw ProtobufUtil.getRemoteException(se);
2674 } finally {
2675 master.close();
2676 }
2677 }
2678
2679
2680
2681
2682 @Deprecated
2683 @Override
2684 public HTableDescriptor[] getHTableDescriptors(
2685 List<String> names) throws IOException {
2686 List<TableName> tableNames = new ArrayList<TableName>(names.size());
2687 for(String name : names) {
2688 tableNames.add(TableName.valueOf(name));
2689 }
2690
2691 return getHTableDescriptorsByTableName(tableNames);
2692 }
2693
2694 @Override
2695 public NonceGenerator getNonceGenerator() {
2696 return this.nonceGenerator;
2697 }
2698
2699
2700
2701
2702
2703
2704
2705
2706 @Deprecated
2707 @Override
2708 public HTableDescriptor getHTableDescriptor(final TableName tableName)
2709 throws IOException {
2710 if (tableName == null) return null;
2711 MasterKeepAliveConnection master = getKeepAliveMasterService();
2712 GetTableDescriptorsResponse htds;
2713 try {
2714 GetTableDescriptorsRequest req =
2715 RequestConverter.buildGetTableDescriptorsRequest(tableName);
2716 htds = master.getTableDescriptors(null, req);
2717 } catch (ServiceException se) {
2718 throw ProtobufUtil.getRemoteException(se);
2719 } finally {
2720 master.close();
2721 }
2722 if (!htds.getTableSchemaList().isEmpty()) {
2723 return HTableDescriptor.convert(htds.getTableSchemaList().get(0));
2724 }
2725 throw new TableNotFoundException(tableName.getNameAsString());
2726 }
2727
2728
2729
2730
2731 @Deprecated
2732 @Override
2733 public HTableDescriptor getHTableDescriptor(final byte[] tableName)
2734 throws IOException {
2735 return getHTableDescriptor(TableName.valueOf(tableName));
2736 }
2737
2738 @Override
2739 public RpcRetryingCallerFactory getNewRpcRetryingCallerFactory(Configuration conf) {
2740 return RpcRetryingCallerFactory
2741 .instantiate(conf, this.interceptor, this.getStatisticsTracker());
2742 }
2743
2744 @Override
2745 public boolean isManaged() {
2746 return managed;
2747 }
2748
2749 @Override
2750 public boolean hasCellBlockSupport() {
2751 return this.rpcClient.hasCellBlockSupport();
2752 }
2753 }
2754
2755
2756
2757
2758 static class ServerErrorTracker {
2759
2760 private final ConcurrentMap<ServerName, ServerErrors> errorsByServer =
2761 new ConcurrentHashMap<ServerName, ServerErrors>();
2762 private final long canRetryUntil;
2763 private final int maxRetries;
2764 private final long startTrackingTime;
2765
2766 public ServerErrorTracker(long timeout, int maxRetries) {
2767 this.maxRetries = maxRetries;
2768 this.canRetryUntil = EnvironmentEdgeManager.currentTime() + timeout;
2769 this.startTrackingTime = new Date().getTime();
2770 }
2771
2772
2773
2774
2775 boolean canRetryMore(int numRetry) {
2776
2777 return numRetry < maxRetries || (maxRetries > 1 &&
2778 EnvironmentEdgeManager.currentTime() < this.canRetryUntil);
2779 }
2780
2781
2782
2783
2784
2785
2786
2787
2788 long calculateBackoffTime(ServerName server, long basePause) {
2789 long result;
2790 ServerErrors errorStats = errorsByServer.get(server);
2791 if (errorStats != null) {
2792 result = ConnectionUtils.getPauseTime(basePause, errorStats.retries.get());
2793 } else {
2794 result = 0;
2795 }
2796 return result;
2797 }
2798
2799
2800
2801
2802
2803
2804 void reportServerError(ServerName server) {
2805 ServerErrors errors = errorsByServer.get(server);
2806 if (errors != null) {
2807 errors.addError();
2808 } else {
2809 errors = errorsByServer.putIfAbsent(server, new ServerErrors());
2810 if (errors != null){
2811 errors.addError();
2812 }
2813 }
2814 }
2815
2816 long getStartTrackingTime() {
2817 return startTrackingTime;
2818 }
2819
2820
2821
2822
2823 private static class ServerErrors {
2824 public final AtomicInteger retries = new AtomicInteger(0);
2825
2826 public void addError() {
2827 retries.incrementAndGet();
2828 }
2829 }
2830 }
2831 }