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 volatile RegionLocations metaLocations = null;
1235 private volatile long lastMetaLookupTime = EnvironmentEdgeManager.currentTime();
1236
1237 private final static long META_LOOKUP_CACHE_INTERVAL = 10000;
1238
1239 private RegionLocations locateMeta(final TableName tableName,
1240 boolean useCache, int replicaId) throws IOException {
1241
1242
1243
1244 if (useCache) {
1245 long now = EnvironmentEdgeManager.currentTime();
1246 if (now - lastMetaLookupTime < META_LOOKUP_CACHE_INTERVAL) {
1247 if (metaLocations != null &&
1248 metaLocations.getRegionLocation(replicaId) != null) {
1249 return metaLocations;
1250 }
1251 } else {
1252 useCache = false;
1253 }
1254 }
1255
1256 synchronized (metaRegionLock) {
1257
1258
1259 if (useCache) {
1260 if (metaLocations != null &&
1261 metaLocations.getRegionLocation(replicaId) != null) {
1262 return metaLocations;
1263 }
1264 }
1265
1266 metaLocations = this.registry.getMetaRegionLocation();
1267 lastMetaLookupTime = EnvironmentEdgeManager.currentTime();
1268 if (metaLocations != null &&
1269 metaLocations.getRegionLocation(replicaId) != null) {
1270 return metaLocations;
1271 }
1272 return null;
1273 }
1274 }
1275
1276
1277
1278
1279
1280 private RegionLocations locateRegionInMeta(TableName tableName, byte[] row,
1281 boolean useCache, boolean retry, int replicaId) throws IOException {
1282
1283
1284
1285 if (useCache) {
1286 RegionLocations locations = getCachedLocation(tableName, row);
1287 if (locations != null && locations.getRegionLocation(replicaId) != null) {
1288 return locations;
1289 }
1290 }
1291
1292
1293
1294
1295 byte[] metaKey = HRegionInfo.createRegionName(tableName, row, HConstants.NINES, false);
1296
1297 Scan s = new Scan();
1298 s.setReversed(true);
1299 s.withStartRow(metaKey);
1300
1301 if (this.useMetaReplicas) {
1302 s.setConsistency(Consistency.TIMELINE);
1303 }
1304
1305 int localNumRetries = (retry ? numTries : 1);
1306
1307 for (int tries = 0; true; tries++) {
1308 if (tries >= localNumRetries) {
1309 throw new NoServerForRegionException("Unable to find region for " +
1310 Bytes.toStringBinary(row) + " in " + tableName + " after " + tries + " tries.");
1311 }
1312 if (useCache) {
1313 RegionLocations locations = getCachedLocation(tableName, row);
1314 if (locations != null && locations.getRegionLocation(replicaId) != null) {
1315 return locations;
1316 }
1317 } else {
1318
1319
1320
1321 metaCache.clearCache(tableName, row, replicaId);
1322 }
1323
1324
1325 long pauseBase = this.pause;
1326 userRegionLock.lock();
1327 try {
1328 if (useCache) {
1329 RegionLocations locations = getCachedLocation(tableName, row);
1330 if (locations != null && locations.getRegionLocation(replicaId) != null) {
1331 return locations;
1332 }
1333 }
1334 Result regionInfoRow = null;
1335 s.resetMvccReadPoint();
1336 s.setOneRowLimit();
1337 try (ReversedClientScanner rcs =
1338 new ReversedClientScanner(conf, s, TableName.META_TABLE_NAME, this, rpcCallerFactory,
1339 rpcControllerFactory, getMetaLookupPool(),
1340 metaReplicaCallTimeoutScanInMicroSecond)) {
1341 regionInfoRow = rcs.next();
1342 }
1343
1344 if (regionInfoRow == null) {
1345 throw new TableNotFoundException(tableName);
1346 }
1347
1348
1349 RegionLocations locations = MetaTableAccessor.getRegionLocations(regionInfoRow);
1350 if (locations == null || locations.getRegionLocation(replicaId) == null) {
1351 throw new IOException("HRegionInfo was null in " +
1352 tableName + ", row=" + regionInfoRow);
1353 }
1354 HRegionInfo regionInfo = locations.getRegionLocation(replicaId).getRegionInfo();
1355 if (regionInfo == null) {
1356 throw new IOException("HRegionInfo was null or empty in " +
1357 TableName.META_TABLE_NAME + ", row=" + regionInfoRow);
1358 }
1359
1360
1361 if (!regionInfo.getTable().equals(tableName)) {
1362 throw new TableNotFoundException(
1363 "Table '" + tableName + "' was not found, got: " +
1364 regionInfo.getTable() + ".");
1365 }
1366 if (regionInfo.isSplit()) {
1367 throw new RegionOfflineException(
1368 "the only available region for the required row is a split parent," +
1369 " the daughters should be online soon: " + regionInfo.getRegionNameAsString());
1370 }
1371 if (regionInfo.isOffline()) {
1372 throw new RegionOfflineException("the region is offline, could" +
1373 " be caused by a disable table call: " + regionInfo.getRegionNameAsString());
1374 }
1375
1376 ServerName serverName = locations.getRegionLocation(replicaId).getServerName();
1377 if (serverName == null) {
1378 throw new NoServerForRegionException("No server address listed in " +
1379 TableName.META_TABLE_NAME + " for region " + regionInfo.getRegionNameAsString() +
1380 " containing row " + Bytes.toStringBinary(row));
1381 }
1382
1383 if (isDeadServer(serverName)){
1384 throw new RegionServerStoppedException("hbase:meta says the region "+
1385 regionInfo.getRegionNameAsString()+" is managed by the server " + serverName +
1386 ", but it is dead.");
1387 }
1388
1389 cacheLocation(tableName, locations);
1390 return locations;
1391 } catch (TableNotFoundException e) {
1392
1393
1394
1395 throw e;
1396 } catch (IOException e) {
1397 ExceptionUtil.rethrowIfInterrupt(e);
1398
1399 if (e instanceof RemoteException) {
1400 e = ((RemoteException)e).unwrapRemoteException();
1401 }
1402 if (e instanceof CallQueueTooBigException) {
1403
1404 pauseBase = this.pauseForCQTBE;
1405 }
1406 if (tries < localNumRetries - 1) {
1407 if (LOG.isDebugEnabled()) {
1408 LOG.debug("locateRegionInMeta parentTable=" + TableName.META_TABLE_NAME +
1409 ", metaLocation=" + ", attempt=" + tries + " of " + localNumRetries +
1410 " failed; retrying after sleep of " +
1411 ConnectionUtils.getPauseTime(pauseBase, tries) + " because: " + e.getMessage());
1412 }
1413 } else {
1414 throw e;
1415 }
1416
1417 if(!(e instanceof RegionOfflineException ||
1418 e instanceof NoServerForRegionException)) {
1419 relocateRegion(TableName.META_TABLE_NAME, metaKey, replicaId);
1420 }
1421 } finally {
1422 userRegionLock.unlock();
1423 }
1424 try{
1425 Thread.sleep(ConnectionUtils.getPauseTime(pauseBase, tries));
1426 } catch (InterruptedException e) {
1427 throw new InterruptedIOException("Giving up trying to location region in " +
1428 "meta: thread is interrupted.");
1429 }
1430 }
1431 }
1432
1433
1434
1435
1436
1437
1438 @Override
1439 public void cacheLocation(final TableName tableName, final RegionLocations location) {
1440 metaCache.cacheLocation(tableName, location);
1441 }
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451 RegionLocations getCachedLocation(final TableName tableName,
1452 final byte [] row) {
1453 return metaCache.getCachedLocation(tableName, row);
1454 }
1455
1456 public void clearRegionCache(final TableName tableName, byte[] row) {
1457 metaCache.clearCache(tableName, row);
1458 }
1459
1460
1461
1462
1463 @Override
1464 public void clearCaches(final ServerName serverName) {
1465 metaCache.clearCache(serverName);
1466 }
1467
1468 @Override
1469 public void clearRegionCache() {
1470 metaCache.clearCache();
1471 }
1472
1473 @Override
1474 public void clearRegionCache(final TableName tableName) {
1475 metaCache.clearCache(tableName);
1476 }
1477
1478 @Override
1479 public void clearRegionCache(final byte[] tableName) {
1480 clearRegionCache(TableName.valueOf(tableName));
1481 }
1482
1483
1484
1485
1486
1487
1488
1489 private void cacheLocation(final TableName tableName, final ServerName source,
1490 final HRegionLocation location) {
1491 metaCache.cacheLocation(tableName, source, location);
1492 }
1493
1494
1495 private final ConcurrentHashMap<String, Object> stubs =
1496 new ConcurrentHashMap<String, Object>();
1497
1498 private final ConcurrentHashMap<String, String> connectionLock =
1499 new ConcurrentHashMap<String, String>();
1500
1501
1502
1503
1504 static class MasterServiceState {
1505 HConnection connection;
1506 MasterService.BlockingInterface stub;
1507 int userCount;
1508
1509 MasterServiceState (final HConnection connection) {
1510 super();
1511 this.connection = connection;
1512 }
1513
1514 @Override
1515 public String toString() {
1516 return "MasterService";
1517 }
1518
1519 Object getStub() {
1520 return this.stub;
1521 }
1522
1523 void clearStub() {
1524 this.stub = null;
1525 }
1526
1527 boolean isMasterRunning() throws ServiceException {
1528 IsMasterRunningResponse response =
1529 this.stub.isMasterRunning(null, RequestConverter.buildIsMasterRunningRequest());
1530 return response != null? response.getIsMasterRunning(): false;
1531 }
1532 }
1533
1534
1535
1536
1537
1538
1539 abstract class StubMaker {
1540
1541
1542
1543 protected abstract String getServiceName();
1544
1545
1546
1547
1548
1549 protected abstract Object makeStub(final BlockingRpcChannel channel);
1550
1551
1552
1553
1554
1555 protected abstract void isMasterRunning() throws ServiceException;
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565 private Object makeStubNoRetries() throws IOException, KeeperException, ServiceException {
1566 ZooKeeperKeepAliveConnection zkw;
1567 try {
1568 zkw = getKeepAliveZooKeeperWatcher();
1569 } catch (IOException e) {
1570 ExceptionUtil.rethrowIfInterrupt(e);
1571 throw new ZooKeeperConnectionException("Can't connect to ZooKeeper", e);
1572 }
1573 try {
1574 checkIfBaseNodeAvailable(zkw);
1575 ServerName sn = MasterAddressTracker.getMasterAddress(zkw);
1576 if (sn == null) {
1577 String msg = "ZooKeeper available but no active master location found";
1578 LOG.info(msg);
1579 throw new MasterNotRunningException(msg);
1580 }
1581 if (isDeadServer(sn)) {
1582 throw new MasterNotRunningException(sn + " is dead.");
1583 }
1584
1585 String key = getStubKey(getServiceName(),
1586 sn.getHostname(), sn.getPort(), hostnamesCanChange);
1587 connectionLock.putIfAbsent(key, key);
1588 Object stub = null;
1589 synchronized (connectionLock.get(key)) {
1590 stub = stubs.get(key);
1591 if (stub == null) {
1592 BlockingRpcChannel channel = rpcClient.createBlockingRpcChannel(sn, user, rpcTimeout);
1593 stub = makeStub(channel);
1594 isMasterRunning();
1595 stubs.put(key, stub);
1596 }
1597 }
1598 return stub;
1599 } finally {
1600 zkw.close();
1601 }
1602 }
1603
1604
1605
1606
1607
1608
1609 Object makeStub() throws IOException {
1610
1611
1612 synchronized (masterAndZKLock) {
1613 Exception exceptionCaught = null;
1614 if (!closed) {
1615 try {
1616 return makeStubNoRetries();
1617 } catch (IOException e) {
1618 exceptionCaught = e;
1619 } catch (KeeperException e) {
1620 exceptionCaught = e;
1621 } catch (ServiceException e) {
1622 exceptionCaught = e;
1623 }
1624
1625 throw new MasterNotRunningException(exceptionCaught);
1626 } else {
1627 throw new DoNotRetryIOException("Connection was closed while trying to get master");
1628 }
1629 }
1630 }
1631 }
1632
1633
1634
1635
1636 class MasterServiceStubMaker extends StubMaker {
1637 private MasterService.BlockingInterface stub;
1638 @Override
1639 protected String getServiceName() {
1640 return MasterService.getDescriptor().getName();
1641 }
1642
1643 @Override
1644 MasterService.BlockingInterface makeStub() throws IOException {
1645 return (MasterService.BlockingInterface)super.makeStub();
1646 }
1647
1648 @Override
1649 protected Object makeStub(BlockingRpcChannel channel) {
1650 this.stub = MasterService.newBlockingStub(channel);
1651 return this.stub;
1652 }
1653
1654 @Override
1655 protected void isMasterRunning() throws ServiceException {
1656 this.stub.isMasterRunning(null, RequestConverter.buildIsMasterRunningRequest());
1657 }
1658 }
1659
1660 @Override
1661 public AdminService.BlockingInterface getAdmin(final ServerName serverName)
1662 throws IOException {
1663 return getAdmin(serverName, false);
1664 }
1665
1666 @Override
1667
1668 public AdminService.BlockingInterface getAdmin(final ServerName serverName,
1669 final boolean master)
1670 throws IOException {
1671 if (isDeadServer(serverName)) {
1672 throw new RegionServerStoppedException(serverName + " is dead.");
1673 }
1674 String key = getStubKey(AdminService.BlockingInterface.class.getName(),
1675 serverName.getHostname(), serverName.getPort(), this.hostnamesCanChange);
1676 this.connectionLock.putIfAbsent(key, key);
1677 AdminService.BlockingInterface stub = null;
1678 synchronized (this.connectionLock.get(key)) {
1679 stub = (AdminService.BlockingInterface)this.stubs.get(key);
1680 if (stub == null) {
1681 BlockingRpcChannel channel =
1682 this.rpcClient.createBlockingRpcChannel(serverName, user, rpcTimeout);
1683 stub = AdminService.newBlockingStub(channel);
1684 this.stubs.put(key, stub);
1685 }
1686 }
1687 return stub;
1688 }
1689
1690 @Override
1691 public ClientService.BlockingInterface getClient(final ServerName sn)
1692 throws IOException {
1693 if (isDeadServer(sn)) {
1694 throw new RegionServerStoppedException(sn + " is dead.");
1695 }
1696 String key = getStubKey(ClientService.BlockingInterface.class.getName(), sn.getHostname(),
1697 sn.getPort(), this.hostnamesCanChange);
1698 this.connectionLock.putIfAbsent(key, key);
1699 ClientService.BlockingInterface stub = null;
1700 synchronized (this.connectionLock.get(key)) {
1701 stub = (ClientService.BlockingInterface)this.stubs.get(key);
1702 if (stub == null) {
1703 BlockingRpcChannel channel =
1704 this.rpcClient.createBlockingRpcChannel(sn, user, rpcTimeout);
1705 stub = ClientService.newBlockingStub(channel);
1706
1707
1708 this.stubs.put(key, stub);
1709 }
1710 }
1711 return stub;
1712 }
1713
1714 static String getStubKey(final String serviceName,
1715 final String rsHostname,
1716 int port,
1717 boolean resolveHostnames) {
1718
1719
1720
1721
1722
1723
1724 String address = rsHostname;
1725 if (resolveHostnames) {
1726 InetAddress i = new InetSocketAddress(rsHostname, port).getAddress();
1727 if (i != null) {
1728 address = i.getHostAddress() + "-" + rsHostname;
1729 }
1730 }
1731 return serviceName + "@" + address + ":" + port;
1732 }
1733
1734 private ZooKeeperKeepAliveConnection keepAliveZookeeper;
1735 private AtomicInteger keepAliveZookeeperUserCount = new AtomicInteger(0);
1736 private boolean canCloseZKW = true;
1737
1738
1739 private static final long keepAlive = 5 * 60 * 1000;
1740
1741
1742
1743
1744
1745 ZooKeeperKeepAliveConnection getKeepAliveZooKeeperWatcher()
1746 throws IOException {
1747 synchronized (masterAndZKLock) {
1748 if (keepAliveZookeeper == null) {
1749 if (this.closed) {
1750 throw new IOException(toString() + " closed");
1751 }
1752
1753
1754 keepAliveZookeeper = new ZooKeeperKeepAliveConnection(conf, this.toString(), this);
1755 }
1756 keepAliveZookeeperUserCount.addAndGet(1);
1757 keepZooKeeperWatcherAliveUntil = Long.MAX_VALUE;
1758 return keepAliveZookeeper;
1759 }
1760 }
1761
1762 void releaseZooKeeperWatcher(final ZooKeeperWatcher zkw) {
1763 if (zkw == null){
1764 return;
1765 }
1766 if (keepAliveZookeeperUserCount.addAndGet(-1) <= 0 ){
1767 keepZooKeeperWatcherAliveUntil = System.currentTimeMillis() + keepAlive;
1768 }
1769 }
1770
1771 private void closeZooKeeperWatcher() {
1772 synchronized (masterAndZKLock) {
1773 if (keepAliveZookeeper != null) {
1774 LOG.info("Closing zookeeper sessionid=0x" +
1775 Long.toHexString(
1776 keepAliveZookeeper.getRecoverableZooKeeper().getSessionId()));
1777 keepAliveZookeeper.internalClose();
1778 keepAliveZookeeper = null;
1779 }
1780 keepAliveZookeeperUserCount.set(0);
1781 }
1782 }
1783
1784 final MasterServiceState masterServiceState = new MasterServiceState(this);
1785
1786 @Override
1787 public MasterService.BlockingInterface getMaster() throws MasterNotRunningException {
1788 return getKeepAliveMasterService();
1789 }
1790
1791 private void resetMasterServiceState(final MasterServiceState mss) {
1792 mss.userCount++;
1793 }
1794
1795 @Override
1796 public MasterKeepAliveConnection getKeepAliveMasterService()
1797 throws MasterNotRunningException {
1798 synchronized (masterAndZKLock) {
1799 if (!isKeepAliveMasterConnectedAndRunning(this.masterServiceState)) {
1800 MasterServiceStubMaker stubMaker = new MasterServiceStubMaker();
1801 try {
1802 this.masterServiceState.stub = stubMaker.makeStub();
1803 } catch (MasterNotRunningException ex) {
1804 throw ex;
1805 } catch (IOException e) {
1806
1807 throw new MasterNotRunningException(e);
1808 }
1809 }
1810 resetMasterServiceState(this.masterServiceState);
1811 }
1812
1813 final MasterService.BlockingInterface stub = this.masterServiceState.stub;
1814 return new MasterKeepAliveConnection() {
1815 MasterServiceState mss = masterServiceState;
1816 @Override
1817 public MasterProtos.AbortProcedureResponse abortProcedure(
1818 RpcController controller,
1819 MasterProtos.AbortProcedureRequest request) throws ServiceException {
1820 return stub.abortProcedure(controller, request);
1821 }
1822 @Override
1823 public MasterProtos.ListProceduresResponse listProcedures(
1824 RpcController controller,
1825 MasterProtos.ListProceduresRequest request) throws ServiceException {
1826 return stub.listProcedures(controller, request);
1827 }
1828
1829 @Override
1830 public MasterProtos.ClearDeadServersResponse clearDeadServers(
1831 RpcController controller,
1832 MasterProtos.ClearDeadServersRequest request) throws ServiceException {
1833 return stub.clearDeadServers(controller, request);
1834 }
1835
1836 @Override
1837 public AddColumnResponse addColumn(RpcController controller, AddColumnRequest request)
1838 throws ServiceException {
1839 return stub.addColumn(controller, request);
1840 }
1841
1842 @Override
1843 public DeleteColumnResponse deleteColumn(RpcController controller,
1844 DeleteColumnRequest request)
1845 throws ServiceException {
1846 return stub.deleteColumn(controller, request);
1847 }
1848
1849 @Override
1850 public ModifyColumnResponse modifyColumn(RpcController controller,
1851 ModifyColumnRequest request)
1852 throws ServiceException {
1853 return stub.modifyColumn(controller, request);
1854 }
1855
1856 @Override
1857 public MoveRegionResponse moveRegion(RpcController controller,
1858 MoveRegionRequest request) throws ServiceException {
1859 return stub.moveRegion(controller, request);
1860 }
1861
1862 @Override
1863 public DispatchMergingRegionsResponse dispatchMergingRegions(
1864 RpcController controller, DispatchMergingRegionsRequest request)
1865 throws ServiceException {
1866 return stub.dispatchMergingRegions(controller, request);
1867 }
1868
1869 @Override
1870 public AssignRegionResponse assignRegion(RpcController controller,
1871 AssignRegionRequest request) throws ServiceException {
1872 return stub.assignRegion(controller, request);
1873 }
1874
1875 @Override
1876 public UnassignRegionResponse unassignRegion(RpcController controller,
1877 UnassignRegionRequest request) throws ServiceException {
1878 return stub.unassignRegion(controller, request);
1879 }
1880
1881 @Override
1882 public OfflineRegionResponse offlineRegion(RpcController controller,
1883 OfflineRegionRequest request) throws ServiceException {
1884 return stub.offlineRegion(controller, request);
1885 }
1886
1887 @Override
1888 public DeleteTableResponse deleteTable(RpcController controller,
1889 DeleteTableRequest request) throws ServiceException {
1890 return stub.deleteTable(controller, request);
1891 }
1892
1893 @Override
1894 public TruncateTableResponse truncateTable(RpcController controller,
1895 TruncateTableRequest request) throws ServiceException {
1896 return stub.truncateTable(controller, request);
1897 }
1898
1899 @Override
1900 public EnableTableResponse enableTable(RpcController controller,
1901 EnableTableRequest request) throws ServiceException {
1902 return stub.enableTable(controller, request);
1903 }
1904
1905 @Override
1906 public DisableTableResponse disableTable(RpcController controller,
1907 DisableTableRequest request) throws ServiceException {
1908 return stub.disableTable(controller, request);
1909 }
1910
1911 @Override
1912 public ModifyTableResponse modifyTable(RpcController controller,
1913 ModifyTableRequest request) throws ServiceException {
1914 return stub.modifyTable(controller, request);
1915 }
1916
1917 @Override
1918 public CreateTableResponse createTable(RpcController controller,
1919 CreateTableRequest request) throws ServiceException {
1920 return stub.createTable(controller, request);
1921 }
1922
1923 @Override
1924 public ShutdownResponse shutdown(RpcController controller,
1925 ShutdownRequest request) throws ServiceException {
1926 return stub.shutdown(controller, request);
1927 }
1928
1929 @Override
1930 public StopMasterResponse stopMaster(RpcController controller,
1931 StopMasterRequest request) throws ServiceException {
1932 return stub.stopMaster(controller, request);
1933 }
1934
1935 @Override
1936 public MasterProtos.IsInMaintenanceModeResponse isMasterInMaintenanceMode(
1937 final RpcController controller,
1938 final MasterProtos.IsInMaintenanceModeRequest request) throws ServiceException {
1939 return stub.isMasterInMaintenanceMode(controller, request);
1940 }
1941
1942 @Override
1943 public BalanceResponse balance(RpcController controller,
1944 BalanceRequest request) throws ServiceException {
1945 return stub.balance(controller, request);
1946 }
1947
1948 @Override
1949 public SetBalancerRunningResponse setBalancerRunning(
1950 RpcController controller, SetBalancerRunningRequest request)
1951 throws ServiceException {
1952 return stub.setBalancerRunning(controller, request);
1953 }
1954
1955 @Override
1956 public NormalizeResponse normalize(RpcController controller,
1957 NormalizeRequest request) throws ServiceException {
1958 return stub.normalize(controller, request);
1959 }
1960
1961 @Override
1962 public SetNormalizerRunningResponse setNormalizerRunning(
1963 RpcController controller, SetNormalizerRunningRequest request)
1964 throws ServiceException {
1965 return stub.setNormalizerRunning(controller, request);
1966 }
1967
1968 @Override
1969 public RunCatalogScanResponse runCatalogScan(RpcController controller,
1970 RunCatalogScanRequest request) throws ServiceException {
1971 return stub.runCatalogScan(controller, request);
1972 }
1973
1974 @Override
1975 public EnableCatalogJanitorResponse enableCatalogJanitor(
1976 RpcController controller, EnableCatalogJanitorRequest request)
1977 throws ServiceException {
1978 return stub.enableCatalogJanitor(controller, request);
1979 }
1980
1981 @Override
1982 public IsCatalogJanitorEnabledResponse isCatalogJanitorEnabled(
1983 RpcController controller, IsCatalogJanitorEnabledRequest request)
1984 throws ServiceException {
1985 return stub.isCatalogJanitorEnabled(controller, request);
1986 }
1987
1988 @Override
1989 public RunCleanerChoreResponse runCleanerChore(RpcController controller,
1990 RunCleanerChoreRequest request) throws ServiceException {
1991 return stub.runCleanerChore(controller, request);
1992 }
1993
1994 @Override
1995 public SetCleanerChoreRunningResponse setCleanerChoreRunning(RpcController controller,
1996 SetCleanerChoreRunningRequest request) throws ServiceException {
1997 return stub.setCleanerChoreRunning(controller, request);
1998 }
1999
2000 @Override
2001 public IsCleanerChoreEnabledResponse isCleanerChoreEnabled(
2002 RpcController controller, IsCleanerChoreEnabledRequest request)
2003 throws ServiceException {
2004 return stub.isCleanerChoreEnabled(controller, request);
2005 }
2006
2007 @Override
2008 public CoprocessorServiceResponse execMasterService(
2009 RpcController controller, CoprocessorServiceRequest request)
2010 throws ServiceException {
2011 return stub.execMasterService(controller, request);
2012 }
2013
2014 @Override
2015 public SnapshotResponse snapshot(RpcController controller,
2016 SnapshotRequest request) throws ServiceException {
2017 return stub.snapshot(controller, request);
2018 }
2019
2020 @Override
2021 public GetCompletedSnapshotsResponse getCompletedSnapshots(
2022 RpcController controller, GetCompletedSnapshotsRequest request)
2023 throws ServiceException {
2024 return stub.getCompletedSnapshots(controller, request);
2025 }
2026
2027 @Override
2028 public DeleteSnapshotResponse deleteSnapshot(RpcController controller,
2029 DeleteSnapshotRequest request) throws ServiceException {
2030 return stub.deleteSnapshot(controller, request);
2031 }
2032
2033 @Override
2034 public IsSnapshotDoneResponse isSnapshotDone(RpcController controller,
2035 IsSnapshotDoneRequest request) throws ServiceException {
2036 return stub.isSnapshotDone(controller, request);
2037 }
2038
2039 @Override
2040 public RestoreSnapshotResponse restoreSnapshot(
2041 RpcController controller, RestoreSnapshotRequest request)
2042 throws ServiceException {
2043 return stub.restoreSnapshot(controller, request);
2044 }
2045
2046 @Override
2047 public IsRestoreSnapshotDoneResponse isRestoreSnapshotDone(
2048 RpcController controller, IsRestoreSnapshotDoneRequest request)
2049 throws ServiceException {
2050 return stub.isRestoreSnapshotDone(controller, request);
2051 }
2052
2053 @Override
2054 public ExecProcedureResponse execProcedure(
2055 RpcController controller, ExecProcedureRequest request)
2056 throws ServiceException {
2057 return stub.execProcedure(controller, request);
2058 }
2059
2060 @Override
2061 public ExecProcedureResponse execProcedureWithRet(
2062 RpcController controller, ExecProcedureRequest request)
2063 throws ServiceException {
2064 return stub.execProcedureWithRet(controller, request);
2065 }
2066
2067 @Override
2068 public IsProcedureDoneResponse isProcedureDone(RpcController controller,
2069 IsProcedureDoneRequest request) throws ServiceException {
2070 return stub.isProcedureDone(controller, request);
2071 }
2072
2073 @Override
2074 public GetProcedureResultResponse getProcedureResult(RpcController controller,
2075 GetProcedureResultRequest request) throws ServiceException {
2076 return stub.getProcedureResult(controller, request);
2077 }
2078
2079 @Override
2080 public IsMasterRunningResponse isMasterRunning(
2081 RpcController controller, IsMasterRunningRequest request)
2082 throws ServiceException {
2083 return stub.isMasterRunning(controller, request);
2084 }
2085
2086 @Override
2087 public ModifyNamespaceResponse modifyNamespace(RpcController controller,
2088 ModifyNamespaceRequest request)
2089 throws ServiceException {
2090 return stub.modifyNamespace(controller, request);
2091 }
2092
2093 @Override
2094 public CreateNamespaceResponse createNamespace(
2095 RpcController controller, CreateNamespaceRequest request) throws ServiceException {
2096 return stub.createNamespace(controller, request);
2097 }
2098
2099 @Override
2100 public DeleteNamespaceResponse deleteNamespace(
2101 RpcController controller, DeleteNamespaceRequest request) throws ServiceException {
2102 return stub.deleteNamespace(controller, request);
2103 }
2104
2105 @Override
2106 public GetNamespaceDescriptorResponse getNamespaceDescriptor(RpcController controller,
2107 GetNamespaceDescriptorRequest request) throws ServiceException {
2108 return stub.getNamespaceDescriptor(controller, request);
2109 }
2110
2111 @Override
2112 public ListNamespaceDescriptorsResponse listNamespaceDescriptors(RpcController controller,
2113 ListNamespaceDescriptorsRequest request) throws ServiceException {
2114 return stub.listNamespaceDescriptors(controller, request);
2115 }
2116
2117 @Override
2118 public ListTableDescriptorsByNamespaceResponse listTableDescriptorsByNamespace(
2119 RpcController controller, ListTableDescriptorsByNamespaceRequest request)
2120 throws ServiceException {
2121 return stub.listTableDescriptorsByNamespace(controller, request);
2122 }
2123
2124 @Override
2125 public ListTableNamesByNamespaceResponse listTableNamesByNamespace(
2126 RpcController controller, ListTableNamesByNamespaceRequest request)
2127 throws ServiceException {
2128 return stub.listTableNamesByNamespace(controller, request);
2129 }
2130
2131 @Override
2132 public void close() {
2133 release(this.mss);
2134 }
2135
2136 @Override
2137 public GetSchemaAlterStatusResponse getSchemaAlterStatus(
2138 RpcController controller, GetSchemaAlterStatusRequest request)
2139 throws ServiceException {
2140 return stub.getSchemaAlterStatus(controller, request);
2141 }
2142
2143 @Override
2144 public GetTableDescriptorsResponse getTableDescriptors(
2145 RpcController controller, GetTableDescriptorsRequest request)
2146 throws ServiceException {
2147 return stub.getTableDescriptors(controller, request);
2148 }
2149
2150 @Override
2151 public GetTableNamesResponse getTableNames(
2152 RpcController controller, GetTableNamesRequest request)
2153 throws ServiceException {
2154 return stub.getTableNames(controller, request);
2155 }
2156
2157 @Override
2158 public GetClusterStatusResponse getClusterStatus(
2159 RpcController controller, GetClusterStatusRequest request)
2160 throws ServiceException {
2161 return stub.getClusterStatus(controller, request);
2162 }
2163
2164 @Override
2165 public SetQuotaResponse setQuota(RpcController controller, SetQuotaRequest request)
2166 throws ServiceException {
2167 return stub.setQuota(controller, request);
2168 }
2169
2170 @Override
2171 public MajorCompactionTimestampResponse getLastMajorCompactionTimestamp(
2172 RpcController controller, MajorCompactionTimestampRequest request)
2173 throws ServiceException {
2174 return stub.getLastMajorCompactionTimestamp(controller, request);
2175 }
2176
2177 @Override
2178 public MajorCompactionTimestampResponse getLastMajorCompactionTimestampForRegion(
2179 RpcController controller, MajorCompactionTimestampForRegionRequest request)
2180 throws ServiceException {
2181 return stub.getLastMajorCompactionTimestampForRegion(controller, request);
2182 }
2183
2184 @Override
2185 public IsBalancerEnabledResponse isBalancerEnabled(RpcController controller,
2186 IsBalancerEnabledRequest request) throws ServiceException {
2187 return stub.isBalancerEnabled(controller, request);
2188 }
2189
2190 @Override
2191 public MasterProtos.SetSplitOrMergeEnabledResponse setSplitOrMergeEnabled(
2192 RpcController controller,
2193 MasterProtos.SetSplitOrMergeEnabledRequest request) throws ServiceException {
2194 return stub.setSplitOrMergeEnabled(controller, request);
2195 }
2196
2197 @Override
2198 public MasterProtos.IsSplitOrMergeEnabledResponse isSplitOrMergeEnabled(
2199 RpcController controller,
2200 MasterProtos.IsSplitOrMergeEnabledRequest request) throws ServiceException {
2201 return stub.isSplitOrMergeEnabled(controller, request);
2202 }
2203
2204 @Override
2205 public IsNormalizerEnabledResponse isNormalizerEnabled(RpcController controller,
2206 IsNormalizerEnabledRequest request) throws ServiceException {
2207 return stub.isNormalizerEnabled(controller, request);
2208 }
2209
2210 @Override
2211 public SecurityCapabilitiesResponse getSecurityCapabilities(RpcController controller,
2212 SecurityCapabilitiesRequest request) throws ServiceException {
2213 return stub.getSecurityCapabilities(controller, request);
2214 }
2215 };
2216 }
2217
2218
2219 private static void release(MasterServiceState mss) {
2220 if (mss != null && mss.connection != null) {
2221 ((HConnectionImplementation)mss.connection).releaseMaster(mss);
2222 }
2223 }
2224
2225 private boolean isKeepAliveMasterConnectedAndRunning(MasterServiceState mss) {
2226 if (mss.getStub() == null){
2227 return false;
2228 }
2229 try {
2230 return mss.isMasterRunning();
2231 } catch (UndeclaredThrowableException e) {
2232
2233
2234 LOG.info("Master connection is not running anymore", e.getUndeclaredThrowable());
2235 return false;
2236 } catch (ServiceException se) {
2237 LOG.warn("Checking master connection", se);
2238 return false;
2239 }
2240 }
2241
2242 void releaseMaster(MasterServiceState mss) {
2243 if (mss.getStub() == null) return;
2244 synchronized (masterAndZKLock) {
2245 --mss.userCount;
2246 }
2247 }
2248
2249 private void closeMasterService(MasterServiceState mss) {
2250 if (mss.getStub() != null) {
2251 LOG.info("Closing master protocol: " + mss);
2252 mss.clearStub();
2253 }
2254 mss.userCount = 0;
2255 }
2256
2257
2258
2259
2260
2261 private void closeMaster() {
2262 synchronized (masterAndZKLock) {
2263 closeMasterService(masterServiceState);
2264 }
2265 }
2266
2267 void updateCachedLocation(HRegionInfo hri, ServerName source,
2268 ServerName serverName, long seqNum) {
2269 HRegionLocation newHrl = new HRegionLocation(hri, serverName, seqNum);
2270 cacheLocation(hri.getTable(), source, newHrl);
2271 }
2272
2273 @Override
2274 public void deleteCachedRegionLocation(final HRegionLocation location) {
2275 metaCache.clearCache(location);
2276 }
2277
2278 @Override
2279 public void updateCachedLocations(final TableName tableName, byte[] rowkey,
2280 final Object exception, final HRegionLocation source) {
2281 assert source != null;
2282 updateCachedLocations(tableName, source.getRegionInfo().getRegionName()
2283 , rowkey, exception, source.getServerName());
2284 }
2285
2286
2287
2288
2289
2290
2291
2292
2293
2294 @Override
2295 public void updateCachedLocations(final TableName tableName, byte[] regionName, byte[] rowkey,
2296 final Object exception, final ServerName source) {
2297 if (rowkey == null || tableName == null) {
2298 LOG.warn("Coding error, see method javadoc. row=" + (rowkey == null ? "null" : rowkey) +
2299 ", tableName=" + (tableName == null ? "null" : tableName));
2300 return;
2301 }
2302
2303 if (source == null) {
2304
2305 return;
2306 }
2307
2308 if (regionName == null) {
2309
2310 if (metrics != null) {
2311 metrics.incrCacheDroppingExceptions(exception);
2312 }
2313 metaCache.clearCache(tableName, rowkey, source);
2314 return;
2315 }
2316
2317
2318 final RegionLocations oldLocations = getCachedLocation(tableName, rowkey);
2319 HRegionLocation oldLocation = null;
2320 if (oldLocations != null) {
2321 oldLocation = oldLocations.getRegionLocationByRegionName(regionName);
2322 }
2323 if (oldLocation == null || !source.equals(oldLocation.getServerName())) {
2324
2325
2326 return;
2327 }
2328
2329 HRegionInfo regionInfo = oldLocation.getRegionInfo();
2330 Throwable cause = ClientExceptionsUtil.findException(exception);
2331 if (cause != null) {
2332 if (!ClientExceptionsUtil.isMetaClearingException(cause)) {
2333
2334 return;
2335 }
2336
2337 if (cause instanceof RegionMovedException) {
2338 RegionMovedException rme = (RegionMovedException) cause;
2339 if (LOG.isTraceEnabled()) {
2340 LOG.trace("Region " + regionInfo.getRegionNameAsString() + " moved to " +
2341 rme.getHostname() + ":" + rme.getPort() +
2342 " according to " + source.getHostAndPort());
2343 }
2344
2345
2346 updateCachedLocation(
2347 regionInfo, source, rme.getServerName(), rme.getLocationSeqNum());
2348 return;
2349 }
2350 }
2351
2352 if (metrics != null) {
2353 metrics.incrCacheDroppingExceptions(exception);
2354 }
2355
2356
2357
2358 metaCache.clearCache(regionInfo);
2359 }
2360
2361 @Override
2362 public void updateCachedLocations(final byte[] tableName, byte[] rowkey,
2363 final Object exception, final HRegionLocation source) {
2364 updateCachedLocations(TableName.valueOf(tableName), rowkey, exception, source);
2365 }
2366
2367 @Override
2368 @Deprecated
2369 public void processBatch(List<? extends Row> list,
2370 final TableName tableName,
2371 ExecutorService pool,
2372 Object[] results) throws IOException, InterruptedException {
2373
2374
2375
2376 if (results.length != list.size()) {
2377 throw new IllegalArgumentException(
2378 "argument results must be the same size as argument list");
2379 }
2380 processBatchCallback(list, tableName, pool, results, null);
2381 }
2382
2383 @Override
2384 @Deprecated
2385 public void processBatch(List<? extends Row> list,
2386 final byte[] tableName,
2387 ExecutorService pool,
2388 Object[] results) throws IOException, InterruptedException {
2389 processBatch(list, TableName.valueOf(tableName), pool, results);
2390 }
2391
2392
2393
2394
2395
2396
2397
2398
2399 @Override
2400 @Deprecated
2401 public <R> void processBatchCallback(
2402 List<? extends Row> list,
2403 TableName tableName,
2404 ExecutorService pool,
2405 Object[] results,
2406 Batch.Callback<R> callback)
2407 throws IOException, InterruptedException {
2408
2409 AsyncRequestFuture ars = this.asyncProcess.submitAll(
2410 pool, tableName, list, callback, results);
2411 ars.waitUntilDone();
2412 if (ars.hasError()) {
2413 throw ars.getErrors();
2414 }
2415 }
2416
2417 @Override
2418 @Deprecated
2419 public <R> void processBatchCallback(
2420 List<? extends Row> list,
2421 byte[] tableName,
2422 ExecutorService pool,
2423 Object[] results,
2424 Batch.Callback<R> callback)
2425 throws IOException, InterruptedException {
2426 processBatchCallback(list, TableName.valueOf(tableName), pool, results, callback);
2427 }
2428
2429
2430 protected AsyncProcess createAsyncProcess(Configuration conf) {
2431
2432 return new AsyncProcess(this, conf, batchPool, rpcCallerFactory, false, rpcControllerFactory,
2433 writeRpcTimeout);
2434 }
2435
2436 @Override
2437 public AsyncProcess getAsyncProcess() {
2438 return asyncProcess;
2439 }
2440
2441 @Override
2442 public ServerStatisticTracker getStatisticsTracker() {
2443 return this.stats;
2444 }
2445
2446 @Override
2447 public ClientBackoffPolicy getBackoffPolicy() {
2448 return this.backoffPolicy;
2449 }
2450
2451
2452
2453
2454
2455 @VisibleForTesting
2456 int getNumberOfCachedRegionLocations(final TableName tableName) {
2457 return metaCache.getNumberOfCachedRegionLocations(tableName);
2458 }
2459
2460 @Override
2461 @Deprecated
2462 public void setRegionCachePrefetch(final TableName tableName, final boolean enable) {
2463 }
2464
2465 @Override
2466 @Deprecated
2467 public void setRegionCachePrefetch(final byte[] tableName,
2468 final boolean enable) {
2469 }
2470
2471 @Override
2472 @Deprecated
2473 public boolean getRegionCachePrefetch(TableName tableName) {
2474 return false;
2475 }
2476
2477 @Override
2478 @Deprecated
2479 public boolean getRegionCachePrefetch(byte[] tableName) {
2480 return false;
2481 }
2482
2483 @Override
2484 public void abort(final String msg, Throwable t) {
2485 if (t instanceof KeeperException.SessionExpiredException
2486 && keepAliveZookeeper != null) {
2487 synchronized (masterAndZKLock) {
2488 if (keepAliveZookeeper != null) {
2489 LOG.warn("This client just lost it's session with ZooKeeper," +
2490 " closing it." +
2491 " It will be recreated next time someone needs it", t);
2492 closeZooKeeperWatcher();
2493 }
2494 }
2495 } else {
2496 if (t != null) {
2497 LOG.fatal(msg, t);
2498 } else {
2499 LOG.fatal(msg);
2500 }
2501 this.aborted = true;
2502 close();
2503 this.closed = true;
2504 }
2505 }
2506
2507 @Override
2508 public boolean isClosed() {
2509 return this.closed;
2510 }
2511
2512 @Override
2513 public boolean isAborted(){
2514 return this.aborted;
2515 }
2516
2517 @Override
2518 public int getCurrentNrHRS() throws IOException {
2519 return this.registry.getCurrentNrHRS();
2520 }
2521
2522
2523
2524
2525 void incCount() {
2526 ++refCount;
2527 }
2528
2529
2530
2531
2532 void decCount() {
2533 if (refCount > 0) {
2534 --refCount;
2535 }
2536 }
2537
2538
2539
2540
2541
2542
2543 boolean isZeroReference() {
2544 return refCount == 0;
2545 }
2546
2547 void internalClose() {
2548 if (this.closed) {
2549 return;
2550 }
2551 closeMaster();
2552 shutdownPools();
2553 if (this.metrics != null) {
2554 this.metrics.shutdown();
2555 }
2556 this.closed = true;
2557 closeZooKeeperWatcher();
2558 this.stubs.clear();
2559 if (clusterStatusListener != null) {
2560 clusterStatusListener.close();
2561 }
2562 if (rpcClient != null) {
2563 rpcClient.close();
2564 }
2565 }
2566
2567 @Override
2568 public void close() {
2569 if (managed) {
2570 if (aborted) {
2571 ConnectionManager.deleteStaleConnection(this);
2572 } else {
2573 ConnectionManager.deleteConnection(this, false);
2574 }
2575 } else {
2576 internalClose();
2577 }
2578 }
2579
2580
2581
2582
2583
2584
2585
2586
2587
2588
2589
2590
2591 @Override
2592 protected void finalize() throws Throwable {
2593 super.finalize();
2594
2595 refCount = 1;
2596 close();
2597 }
2598
2599
2600
2601
2602 @Deprecated
2603 @Override
2604 public HTableDescriptor[] listTables() throws IOException {
2605 MasterKeepAliveConnection master = getKeepAliveMasterService();
2606 try {
2607 GetTableDescriptorsRequest req =
2608 RequestConverter.buildGetTableDescriptorsRequest((List<TableName>)null);
2609 return ProtobufUtil.getHTableDescriptorArray(master.getTableDescriptors(null, req));
2610 } catch (ServiceException se) {
2611 throw ProtobufUtil.getRemoteException(se);
2612 } finally {
2613 master.close();
2614 }
2615 }
2616
2617
2618
2619
2620 @Deprecated
2621 @Override
2622 public String[] getTableNames() throws IOException {
2623 TableName[] tableNames = listTableNames();
2624 String result[] = new String[tableNames.length];
2625 for (int i = 0; i < tableNames.length; i++) {
2626 result[i] = tableNames[i].getNameAsString();
2627 }
2628 return result;
2629 }
2630
2631
2632
2633
2634 @Deprecated
2635 @Override
2636 public TableName[] listTableNames() throws IOException {
2637 MasterKeepAliveConnection master = getKeepAliveMasterService();
2638 try {
2639 return ProtobufUtil.getTableNameArray(master.getTableNames(null,
2640 GetTableNamesRequest.newBuilder().build())
2641 .getTableNamesList());
2642 } catch (ServiceException se) {
2643 throw ProtobufUtil.getRemoteException(se);
2644 } finally {
2645 master.close();
2646 }
2647 }
2648
2649
2650
2651
2652 @Deprecated
2653 @Override
2654 public HTableDescriptor[] getHTableDescriptorsByTableName(
2655 List<TableName> tableNames) throws IOException {
2656 if (tableNames == null || tableNames.isEmpty()) return new HTableDescriptor[0];
2657 MasterKeepAliveConnection master = getKeepAliveMasterService();
2658 try {
2659 GetTableDescriptorsRequest req =
2660 RequestConverter.buildGetTableDescriptorsRequest(tableNames);
2661 return ProtobufUtil.getHTableDescriptorArray(master.getTableDescriptors(null, req));
2662 } catch (ServiceException se) {
2663 throw ProtobufUtil.getRemoteException(se);
2664 } finally {
2665 master.close();
2666 }
2667 }
2668
2669
2670
2671
2672 @Deprecated
2673 @Override
2674 public HTableDescriptor[] getHTableDescriptors(
2675 List<String> names) throws IOException {
2676 List<TableName> tableNames = new ArrayList<TableName>(names.size());
2677 for(String name : names) {
2678 tableNames.add(TableName.valueOf(name));
2679 }
2680
2681 return getHTableDescriptorsByTableName(tableNames);
2682 }
2683
2684 @Override
2685 public NonceGenerator getNonceGenerator() {
2686 return this.nonceGenerator;
2687 }
2688
2689
2690
2691
2692
2693
2694
2695
2696 @Deprecated
2697 @Override
2698 public HTableDescriptor getHTableDescriptor(final TableName tableName)
2699 throws IOException {
2700 if (tableName == null) return null;
2701 MasterKeepAliveConnection master = getKeepAliveMasterService();
2702 GetTableDescriptorsResponse htds;
2703 try {
2704 GetTableDescriptorsRequest req =
2705 RequestConverter.buildGetTableDescriptorsRequest(tableName);
2706 htds = master.getTableDescriptors(null, req);
2707 } catch (ServiceException se) {
2708 throw ProtobufUtil.getRemoteException(se);
2709 } finally {
2710 master.close();
2711 }
2712 if (!htds.getTableSchemaList().isEmpty()) {
2713 return HTableDescriptor.convert(htds.getTableSchemaList().get(0));
2714 }
2715 throw new TableNotFoundException(tableName.getNameAsString());
2716 }
2717
2718
2719
2720
2721 @Deprecated
2722 @Override
2723 public HTableDescriptor getHTableDescriptor(final byte[] tableName)
2724 throws IOException {
2725 return getHTableDescriptor(TableName.valueOf(tableName));
2726 }
2727
2728 @Override
2729 public RpcRetryingCallerFactory getNewRpcRetryingCallerFactory(Configuration conf) {
2730 return RpcRetryingCallerFactory
2731 .instantiate(conf, this.interceptor, this.getStatisticsTracker());
2732 }
2733
2734 @Override
2735 public boolean isManaged() {
2736 return managed;
2737 }
2738
2739 @Override
2740 public boolean hasCellBlockSupport() {
2741 return this.rpcClient.hasCellBlockSupport();
2742 }
2743
2744 @Override
2745 public ConnectionConfiguration getConnectionConfiguration() {
2746 return this.connectionConfig;
2747 }
2748
2749 @Override
2750 public RpcRetryingCallerFactory getRpcRetryingCallerFactory() {
2751 return this.rpcCallerFactory;
2752 }
2753
2754 @Override
2755 public RpcControllerFactory getRpcControllerFactory() {
2756 return this.rpcControllerFactory;
2757 }
2758 }
2759
2760
2761
2762
2763 static class ServerErrorTracker {
2764
2765 private final ConcurrentMap<ServerName, ServerErrors> errorsByServer =
2766 new ConcurrentHashMap<ServerName, ServerErrors>();
2767 private final long canRetryUntil;
2768 private final int maxRetries;
2769 private final long startTrackingTime;
2770
2771 public ServerErrorTracker(long timeout, int maxRetries) {
2772 this.maxRetries = maxRetries;
2773 this.canRetryUntil = EnvironmentEdgeManager.currentTime() + timeout;
2774 this.startTrackingTime = new Date().getTime();
2775 }
2776
2777
2778
2779
2780 boolean canRetryMore(int numRetry) {
2781
2782 return numRetry < maxRetries || (maxRetries > 1 &&
2783 EnvironmentEdgeManager.currentTime() < this.canRetryUntil);
2784 }
2785
2786
2787
2788
2789
2790
2791
2792
2793 long calculateBackoffTime(ServerName server, long basePause) {
2794 long result;
2795 ServerErrors errorStats = errorsByServer.get(server);
2796 if (errorStats != null) {
2797 result = ConnectionUtils.getPauseTime(basePause, errorStats.retries.get());
2798 } else {
2799 result = 0;
2800 }
2801 return result;
2802 }
2803
2804
2805
2806
2807
2808
2809 void reportServerError(ServerName server) {
2810 ServerErrors errors = errorsByServer.get(server);
2811 if (errors != null) {
2812 errors.addError();
2813 } else {
2814 errors = errorsByServer.putIfAbsent(server, new ServerErrors());
2815 if (errors != null){
2816 errors.addError();
2817 }
2818 }
2819 }
2820
2821 long getStartTrackingTime() {
2822 return startTrackingTime;
2823 }
2824
2825
2826
2827
2828 private static class ServerErrors {
2829 public final AtomicInteger retries = new AtomicInteger(0);
2830
2831 public void addError() {
2832 retries.incrementAndGet();
2833 }
2834 }
2835 }
2836 }