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