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