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