3 * Licensed to the Apache Software Foundation (ASF) under one
4 * or more contributor license agreements. See the NOTICE file
5 * distributed with this work for additional information
6 * regarding copyright ownership. The ASF licenses this file
7 * to you under the Apache License, Version 2.0 (the
8 * "License"); you may not use this file except in compliance
9 * with the License. You may obtain a copy of the License at
11 * http://www.apache.org/licenses/LICENSE-2.0
13 * Unless required by applicable law or agreed to in writing, software
14 * distributed under the License is distributed on an "AS IS" BASIS,
15 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 * See the License for the specific language governing permissions and
17 * limitations under the License.
19 package org
.apache
.hadoop
.hbase
.zookeeper
;
21 import java
.io
.BufferedReader
;
22 import java
.io
.BufferedWriter
;
23 import java
.io
.IOException
;
24 import java
.io
.InputStreamReader
;
25 import java
.io
.OutputStreamWriter
;
26 import java
.io
.PrintWriter
;
27 import java
.net
.InetSocketAddress
;
28 import java
.net
.Socket
;
29 import java
.net
.UnknownHostException
;
30 import java
.nio
.charset
.StandardCharsets
;
31 import java
.util
.ArrayList
;
32 import java
.util
.Arrays
;
33 import java
.util
.Collections
;
34 import java
.util
.Deque
;
35 import java
.util
.HashMap
;
36 import java
.util
.Iterator
;
37 import java
.util
.LinkedList
;
38 import java
.util
.List
;
40 import java
.util
.stream
.Collectors
;
41 import javax
.security
.auth
.login
.AppConfigurationEntry
;
42 import javax
.security
.auth
.login
.AppConfigurationEntry
.LoginModuleControlFlag
;
43 import org
.apache
.commons
.lang3
.StringUtils
;
44 import org
.apache
.hadoop
.conf
.Configuration
;
45 import org
.apache
.hadoop
.hbase
.AuthUtil
;
46 import org
.apache
.hadoop
.hbase
.HConstants
;
47 import org
.apache
.hadoop
.hbase
.exceptions
.DeserializationException
;
48 import org
.apache
.hadoop
.hbase
.security
.Superusers
;
49 import org
.apache
.hadoop
.hbase
.util
.Bytes
;
50 import org
.apache
.hadoop
.hbase
.util
.Threads
;
51 import org
.apache
.hadoop
.hbase
.zookeeper
.ZKUtil
.ZKUtilOp
.CreateAndFailSilent
;
52 import org
.apache
.hadoop
.hbase
.zookeeper
.ZKUtil
.ZKUtilOp
.DeleteNodeFailSilent
;
53 import org
.apache
.hadoop
.hbase
.zookeeper
.ZKUtil
.ZKUtilOp
.SetData
;
54 import org
.apache
.hadoop
.security
.SecurityUtil
;
55 import org
.apache
.hadoop
.security
.UserGroupInformation
;
56 import org
.apache
.hadoop
.security
.authentication
.util
.KerberosUtil
;
57 import org
.apache
.yetus
.audience
.InterfaceAudience
;
58 import org
.apache
.zookeeper
.AsyncCallback
;
59 import org
.apache
.zookeeper
.CreateMode
;
60 import org
.apache
.zookeeper
.KeeperException
;
61 import org
.apache
.zookeeper
.KeeperException
.NoNodeException
;
62 import org
.apache
.zookeeper
.Op
;
63 import org
.apache
.zookeeper
.Watcher
;
64 import org
.apache
.zookeeper
.ZooDefs
.Ids
;
65 import org
.apache
.zookeeper
.ZooDefs
.Perms
;
66 import org
.apache
.zookeeper
.ZooKeeper
;
67 import org
.apache
.zookeeper
.client
.ZooKeeperSaslClient
;
68 import org
.apache
.zookeeper
.data
.ACL
;
69 import org
.apache
.zookeeper
.data
.Id
;
70 import org
.apache
.zookeeper
.data
.Stat
;
71 import org
.apache
.zookeeper
.proto
.CreateRequest
;
72 import org
.apache
.zookeeper
.proto
.DeleteRequest
;
73 import org
.apache
.zookeeper
.proto
.SetDataRequest
;
74 import org
.apache
.zookeeper
.server
.ZooKeeperSaslServer
;
75 import org
.slf4j
.Logger
;
76 import org
.slf4j
.LoggerFactory
;
78 import org
.apache
.hbase
.thirdparty
.com
.google
.protobuf
.InvalidProtocolBufferException
;
80 import org
.apache
.hadoop
.hbase
.shaded
.protobuf
.ProtobufUtil
;
81 import org
.apache
.hadoop
.hbase
.shaded
.protobuf
.generated
.ReplicationProtos
;
84 * Internal HBase utility class for ZooKeeper.
86 * <p>Contains only static methods and constants.
88 * <p>Methods all throw {@link KeeperException} if there is an unexpected
89 * zookeeper exception, so callers of these methods must handle appropriately.
90 * If ZK is required for the operation, the server will need to be aborted.
92 @InterfaceAudience.Private
93 public final class ZKUtil
{
94 private static final Logger LOG
= LoggerFactory
.getLogger(ZKUtil
.class);
96 private static int zkDumpConnectionTimeOut
;
102 * Creates a new connection to ZooKeeper, pulling settings and ensemble config
103 * from the specified configuration object using methods from {@link ZKConfig}.
105 * Sets the connection status monitoring watcher to the specified watcher.
107 * @param conf configuration to pull ensemble and other settings from
108 * @param watcher watcher to monitor connection changes
109 * @return connection to zookeeper
110 * @throws IOException if unable to connect to zk or config problem
112 public static RecoverableZooKeeper
connect(Configuration conf
, Watcher watcher
)
114 String ensemble
= ZKConfig
.getZKQuorumServersString(conf
);
115 return connect(conf
, ensemble
, watcher
);
118 public static RecoverableZooKeeper
connect(Configuration conf
, String ensemble
,
121 return connect(conf
, ensemble
, watcher
, null);
124 public static RecoverableZooKeeper
connect(Configuration conf
, String ensemble
,
125 Watcher watcher
, final String identifier
)
127 if(ensemble
== null) {
128 throw new IOException("Unable to determine ZooKeeper ensemble");
130 int timeout
= conf
.getInt(HConstants
.ZK_SESSION_TIMEOUT
,
131 HConstants
.DEFAULT_ZK_SESSION_TIMEOUT
);
132 if (LOG
.isTraceEnabled()) {
133 LOG
.trace("{} opening connection to ZooKeeper ensemble={}", identifier
, ensemble
);
135 int retry
= conf
.getInt("zookeeper.recovery.retry", 3);
136 int retryIntervalMillis
=
137 conf
.getInt("zookeeper.recovery.retry.intervalmill", 1000);
138 int maxSleepTime
= conf
.getInt("zookeeper.recovery.retry.maxsleeptime", 60000);
139 zkDumpConnectionTimeOut
= conf
.getInt("zookeeper.dump.connection.timeout",
141 int multiMaxSize
= conf
.getInt("zookeeper.multi.max.size", 1024*1024);
142 return new RecoverableZooKeeper(ensemble
, timeout
, watcher
,
143 retry
, retryIntervalMillis
, maxSleepTime
, identifier
, multiMaxSize
);
147 * Log in the current zookeeper server process using the given configuration
148 * keys for the credential file and login principal.
150 * <p><strong>This is only applicable when running on secure hbase</strong>
151 * On regular HBase (without security features), this will safely be ignored.
154 * @param conf The configuration data to use
155 * @param keytabFileKey Property key used to configure the path to the credential file
156 * @param userNameKey Property key used to configure the login principal
157 * @param hostname Current hostname to use in any credentials
158 * @throws IOException underlying exception from SecurityUtil.login() call
160 public static void loginServer(Configuration conf
, String keytabFileKey
,
161 String userNameKey
, String hostname
) throws IOException
{
162 login(conf
, keytabFileKey
, userNameKey
, hostname
,
163 ZooKeeperSaslServer
.LOGIN_CONTEXT_NAME_KEY
,
164 JaasConfiguration
.SERVER_KEYTAB_KERBEROS_CONFIG_NAME
);
168 * Log in the current zookeeper client using the given configuration
169 * keys for the credential file and login principal.
171 * <p><strong>This is only applicable when running on secure hbase</strong>
172 * On regular HBase (without security features), this will safely be ignored.
175 * @param conf The configuration data to use
176 * @param keytabFileKey Property key used to configure the path to the credential file
177 * @param userNameKey Property key used to configure the login principal
178 * @param hostname Current hostname to use in any credentials
179 * @throws IOException underlying exception from SecurityUtil.login() call
181 public static void loginClient(Configuration conf
, String keytabFileKey
,
182 String userNameKey
, String hostname
) throws IOException
{
183 login(conf
, keytabFileKey
, userNameKey
, hostname
,
184 ZooKeeperSaslClient
.LOGIN_CONTEXT_NAME_KEY
,
185 JaasConfiguration
.CLIENT_KEYTAB_KERBEROS_CONFIG_NAME
);
189 * Log in the current process using the given configuration keys for the
190 * credential file and login principal.
192 * <p><strong>This is only applicable when running on secure hbase</strong>
193 * On regular HBase (without security features), this will safely be ignored.
196 * @param conf The configuration data to use
197 * @param keytabFileKey Property key used to configure the path to the credential file
198 * @param userNameKey Property key used to configure the login principal
199 * @param hostname Current hostname to use in any credentials
200 * @param loginContextProperty property name to expose the entry name
201 * @param loginContextName jaas entry name
202 * @throws IOException underlying exception from SecurityUtil.login() call
204 private static void login(Configuration conf
, String keytabFileKey
,
205 String userNameKey
, String hostname
,
206 String loginContextProperty
, String loginContextName
)
208 if (!isSecureZooKeeper(conf
)) {
212 // User has specified a jaas.conf, keep this one as the good one.
213 // HBASE_OPTS="-Djava.security.auth.login.config=jaas.conf"
214 if (System
.getProperty("java.security.auth.login.config") != null) {
218 // No keytab specified, no auth
219 String keytabFilename
= conf
.get(keytabFileKey
);
220 if (keytabFilename
== null) {
221 LOG
.warn("no keytab specified for: {}", keytabFileKey
);
225 String principalConfig
= conf
.get(userNameKey
, System
.getProperty("user.name"));
226 String principalName
= SecurityUtil
.getServerPrincipal(principalConfig
, hostname
);
228 // Initialize the "jaas.conf" for keyTab/principal,
229 // If keyTab is not specified use the Ticket Cache.
230 // and set the zookeeper login context name.
231 JaasConfiguration jaasConf
= new JaasConfiguration(loginContextName
,
232 principalName
, keytabFilename
);
233 javax
.security
.auth
.login
.Configuration
.setConfiguration(jaasConf
);
234 System
.setProperty(loginContextProperty
, loginContextName
);
238 * A JAAS configuration that defines the login modules that we want to use for login.
240 private static class JaasConfiguration
extends javax
.security
.auth
.login
.Configuration
{
241 private static final String SERVER_KEYTAB_KERBEROS_CONFIG_NAME
=
242 "zookeeper-server-keytab-kerberos";
243 private static final String CLIENT_KEYTAB_KERBEROS_CONFIG_NAME
=
244 "zookeeper-client-keytab-kerberos";
246 private static final Map
<String
, String
> BASIC_JAAS_OPTIONS
= new HashMap
<>();
248 String jaasEnvVar
= System
.getenv("HBASE_JAAS_DEBUG");
249 if ("true".equalsIgnoreCase(jaasEnvVar
)) {
250 BASIC_JAAS_OPTIONS
.put("debug", "true");
254 private static final Map
<String
,String
> KEYTAB_KERBEROS_OPTIONS
= new HashMap
<>();
256 KEYTAB_KERBEROS_OPTIONS
.put("doNotPrompt", "true");
257 KEYTAB_KERBEROS_OPTIONS
.put("storeKey", "true");
258 KEYTAB_KERBEROS_OPTIONS
.put("refreshKrb5Config", "true");
259 KEYTAB_KERBEROS_OPTIONS
.putAll(BASIC_JAAS_OPTIONS
);
262 private static final AppConfigurationEntry KEYTAB_KERBEROS_LOGIN
=
263 new AppConfigurationEntry(KerberosUtil
.getKrb5LoginModuleName(),
264 LoginModuleControlFlag
.REQUIRED
,
265 KEYTAB_KERBEROS_OPTIONS
);
267 private static final AppConfigurationEntry
[] KEYTAB_KERBEROS_CONF
=
268 new AppConfigurationEntry
[]{KEYTAB_KERBEROS_LOGIN
};
270 private javax
.security
.auth
.login
.Configuration baseConfig
;
271 private final String loginContextName
;
272 private final boolean useTicketCache
;
273 private final String keytabFile
;
274 private final String principal
;
276 public JaasConfiguration(String loginContextName
, String principal
, String keytabFile
) {
277 this(loginContextName
, principal
, keytabFile
, keytabFile
== null || keytabFile
.length() == 0);
280 private JaasConfiguration(String loginContextName
, String principal
,
281 String keytabFile
, boolean useTicketCache
) {
283 this.baseConfig
= javax
.security
.auth
.login
.Configuration
.getConfiguration();
284 } catch (SecurityException e
) {
285 this.baseConfig
= null;
287 this.loginContextName
= loginContextName
;
288 this.useTicketCache
= useTicketCache
;
289 this.keytabFile
= keytabFile
;
290 this.principal
= principal
;
291 LOG
.info("JaasConfiguration loginContextName={} principal={} useTicketCache={} keytabFile={}",
292 loginContextName
, principal
, useTicketCache
, keytabFile
);
296 public AppConfigurationEntry
[] getAppConfigurationEntry(String appName
) {
297 if (loginContextName
.equals(appName
)) {
298 if (!useTicketCache
) {
299 KEYTAB_KERBEROS_OPTIONS
.put("keyTab", keytabFile
);
300 KEYTAB_KERBEROS_OPTIONS
.put("useKeyTab", "true");
302 KEYTAB_KERBEROS_OPTIONS
.put("principal", principal
);
303 KEYTAB_KERBEROS_OPTIONS
.put("useTicketCache", useTicketCache ?
"true" : "false");
304 return KEYTAB_KERBEROS_CONF
;
307 if (baseConfig
!= null) {
308 return baseConfig
.getAppConfigurationEntry(appName
);
319 * Returns the full path of the immediate parent of the specified node.
320 * @param node path to get parent of
321 * @return parent of path, null if passed the root node or an invalid node
323 public static String
getParent(String node
) {
324 int idx
= node
.lastIndexOf(ZNodePaths
.ZNODE_PATH_SEPARATOR
);
325 return idx
<= 0 ?
null : node
.substring(0, idx
);
329 * Get the name of the current node from the specified fully-qualified path.
330 * @param path fully-qualified path
331 * @return name of the current node
333 public static String
getNodeName(String path
) {
334 return path
.substring(path
.lastIndexOf("/")+1);
338 // Existence checks and watches
342 * Watch the specified znode for delete/create/change events. The watcher is
343 * set whether or not the node exists. If the node already exists, the method
344 * returns true. If the node does not exist, the method returns false.
346 * @param zkw zk reference
347 * @param znode path of node to watch
348 * @return true if znode exists, false if does not exist or error
349 * @throws KeeperException if unexpected zookeeper exception
351 public static boolean watchAndCheckExists(ZKWatcher zkw
, String znode
)
352 throws KeeperException
{
354 Stat s
= zkw
.getRecoverableZooKeeper().exists(znode
, zkw
);
355 boolean exists
= s
!= null;
357 LOG
.debug(zkw
.prefix("Set watcher on existing znode=" + znode
));
359 LOG
.debug(zkw
.prefix("Set watcher on znode that does not yet exist, " + znode
));
362 } catch (KeeperException e
) {
363 LOG
.warn(zkw
.prefix("Unable to set watcher on znode " + znode
), e
);
364 zkw
.keeperException(e
);
366 } catch (InterruptedException e
) {
367 LOG
.warn(zkw
.prefix("Unable to set watcher on znode " + znode
), e
);
368 zkw
.interruptedException(e
);
374 * Watch the specified znode, but only if exists. Useful when watching
375 * for deletions. Uses .getData() (and handles NoNodeException) instead
376 * of .exists() to accomplish this, as .getData() will only set a watch if
378 * @param zkw zk reference
379 * @param znode path of node to watch
380 * @return true if the watch is set, false if node does not exists
381 * @throws KeeperException if unexpected zookeeper exception
383 public static boolean setWatchIfNodeExists(ZKWatcher zkw
, String znode
)
384 throws KeeperException
{
386 zkw
.getRecoverableZooKeeper().getData(znode
, true, null);
388 } catch (NoNodeException e
) {
390 } catch (InterruptedException e
) {
391 LOG
.warn(zkw
.prefix("Unable to set watcher on znode " + znode
), e
);
392 zkw
.interruptedException(e
);
398 * Check if the specified node exists. Sets no watches.
400 * @param zkw zk reference
401 * @param znode path of node to watch
402 * @return version of the node if it exists, -1 if does not exist
403 * @throws KeeperException if unexpected zookeeper exception
405 public static int checkExists(ZKWatcher zkw
, String znode
)
406 throws KeeperException
{
408 Stat s
= zkw
.getRecoverableZooKeeper().exists(znode
, null);
409 return s
!= null ? s
.getVersion() : -1;
410 } catch (KeeperException e
) {
411 LOG
.warn(zkw
.prefix("Unable to set watcher on znode (" + znode
+ ")"), e
);
412 zkw
.keeperException(e
);
414 } catch (InterruptedException e
) {
415 LOG
.warn(zkw
.prefix("Unable to set watcher on znode (" + znode
+ ")"), e
);
416 zkw
.interruptedException(e
);
426 * Lists the children znodes of the specified znode. Also sets a watch on
427 * the specified znode which will capture a NodeDeleted event on the specified
428 * znode as well as NodeChildrenChanged if any children of the specified znode
429 * are created or deleted.
431 * Returns null if the specified node does not exist. Otherwise returns a
432 * list of children of the specified node. If the node exists but it has no
433 * children, an empty list will be returned.
435 * @param zkw zk reference
436 * @param znode path of node to list and watch children of
437 * @return list of children of the specified node, an empty list if the node
438 * exists but has no children, and null if the node does not exist
439 * @throws KeeperException if unexpected zookeeper exception
441 public static List
<String
> listChildrenAndWatchForNewChildren(
442 ZKWatcher zkw
, String znode
)
443 throws KeeperException
{
445 return zkw
.getRecoverableZooKeeper().getChildren(znode
, zkw
);
446 } catch(KeeperException
.NoNodeException ke
) {
447 LOG
.debug(zkw
.prefix("Unable to list children of znode " + znode
+ " " +
448 "because node does not exist (not an error)"));
449 } catch (KeeperException e
) {
450 LOG
.warn(zkw
.prefix("Unable to list children of znode " + znode
+ " "), e
);
451 zkw
.keeperException(e
);
452 } catch (InterruptedException e
) {
453 LOG
.warn(zkw
.prefix("Unable to list children of znode " + znode
+ " "), e
);
454 zkw
.interruptedException(e
);
461 * List all the children of the specified znode, setting a watch for children
462 * changes and also setting a watch on every individual child in order to get
463 * the NodeCreated and NodeDeleted events.
464 * @param zkw zookeeper reference
465 * @param znode node to get children of and watch
466 * @return list of znode names, null if the node doesn't exist
467 * @throws KeeperException if a ZooKeeper operation fails
469 public static List
<String
> listChildrenAndWatchThem(ZKWatcher zkw
,
470 String znode
) throws KeeperException
{
471 List
<String
> children
= listChildrenAndWatchForNewChildren(zkw
, znode
);
472 if (children
== null) {
475 for (String child
: children
) {
476 watchAndCheckExists(zkw
, ZNodePaths
.joinZNode(znode
, child
));
482 * Lists the children of the specified znode without setting any watches.
484 * Sets no watches at all, this method is best effort.
486 * Returns an empty list if the node has no children. Returns null if the
487 * parent node itself does not exist.
489 * @param zkw zookeeper reference
490 * @param znode node to get children
491 * @return list of data of children of specified znode, empty if no children,
492 * null if parent does not exist
493 * @throws KeeperException if unexpected zookeeper exception
495 public static List
<String
> listChildrenNoWatch(ZKWatcher zkw
, String znode
)
496 throws KeeperException
{
497 List
<String
> children
= null;
499 // List the children without watching
500 children
= zkw
.getRecoverableZooKeeper().getChildren(znode
, null);
501 } catch(KeeperException
.NoNodeException nne
) {
503 } catch(InterruptedException ie
) {
504 zkw
.interruptedException(ie
);
510 * Simple class to hold a node path and node data.
514 public static class NodeAndData
{
516 private byte [] data
;
517 public NodeAndData(String node
, byte [] data
) {
521 public String
getNode() {
524 public byte [] getData() {
528 public String
toString() {
531 public boolean isEmpty() {
532 return (data
== null || data
.length
== 0);
537 * Checks if the specified znode has any children. Sets no watches.
539 * Returns true if the node exists and has children. Returns false if the
540 * node does not exist or if the node does not have any children.
542 * Used during master initialization to determine if the master is a
543 * failed-over-to master or the first master during initial cluster startup.
544 * If the directory for regionserver ephemeral nodes is empty then this is
545 * a cluster startup, if not then it is not cluster startup.
547 * @param zkw zk reference
548 * @param znode path of node to check for children of
549 * @return true if node has children, false if not or node does not exist
550 * @throws KeeperException if unexpected zookeeper exception
552 public static boolean nodeHasChildren(ZKWatcher zkw
, String znode
)
553 throws KeeperException
{
555 return !zkw
.getRecoverableZooKeeper().getChildren(znode
, null).isEmpty();
556 } catch(KeeperException
.NoNodeException ke
) {
557 LOG
.debug(zkw
.prefix("Unable to list children of znode " + znode
+
558 " because node does not exist (not an error)"));
560 } catch (KeeperException e
) {
561 LOG
.warn(zkw
.prefix("Unable to list children of znode " + znode
), e
);
562 zkw
.keeperException(e
);
564 } catch (InterruptedException e
) {
565 LOG
.warn(zkw
.prefix("Unable to list children of znode " + znode
), e
);
566 zkw
.interruptedException(e
);
572 * Get the number of children of the specified node.
574 * If the node does not exist or has no children, returns 0.
576 * Sets no watches at all.
578 * @param zkw zk reference
579 * @param znode path of node to count children of
580 * @return number of children of specified node, 0 if none or parent does not
582 * @throws KeeperException if unexpected zookeeper exception
584 public static int getNumberOfChildren(ZKWatcher zkw
, String znode
)
585 throws KeeperException
{
587 Stat stat
= zkw
.getRecoverableZooKeeper().exists(znode
, null);
588 return stat
== null ?
0 : stat
.getNumChildren();
589 } catch(KeeperException e
) {
590 LOG
.warn(zkw
.prefix("Unable to get children of node " + znode
));
591 zkw
.keeperException(e
);
592 } catch(InterruptedException e
) {
593 zkw
.interruptedException(e
);
603 * Get znode data. Does not set a watcher.
605 * @return ZNode data, null if the node does not exist or if there is an error.
607 public static byte [] getData(ZKWatcher zkw
, String znode
)
608 throws KeeperException
, InterruptedException
{
610 byte [] data
= zkw
.getRecoverableZooKeeper().getData(znode
, null, null);
611 logRetrievedMsg(zkw
, znode
, data
, false);
613 } catch (KeeperException
.NoNodeException e
) {
614 LOG
.debug(zkw
.prefix("Unable to get data of znode " + znode
+ " " +
615 "because node does not exist (not an error)"));
617 } catch (KeeperException e
) {
618 LOG
.warn(zkw
.prefix("Unable to get data of znode " + znode
), e
);
619 zkw
.keeperException(e
);
625 * Get the data at the specified znode and set a watch.
627 * Returns the data and sets a watch if the node exists. Returns null and no
628 * watch is set if the node does not exist or there is an exception.
630 * @param zkw zk reference
631 * @param znode path of node
632 * @return data of the specified znode, or null
633 * @throws KeeperException if unexpected zookeeper exception
635 public static byte[] getDataAndWatch(ZKWatcher zkw
, String znode
) throws KeeperException
{
636 return getDataInternal(zkw
, znode
, null, true, true);
640 * Get the data at the specified znode and set a watch.
641 * Returns the data and sets a watch if the node exists. Returns null and no
642 * watch is set if the node does not exist or there is an exception.
644 * @param zkw zk reference
645 * @param znode path of node
646 * @param throwOnInterrupt if false then just interrupt the thread, do not throw exception
647 * @return data of the specified znode, or null
648 * @throws KeeperException if unexpected zookeeper exception
650 public static byte[] getDataAndWatch(ZKWatcher zkw
, String znode
, boolean throwOnInterrupt
)
651 throws KeeperException
{
652 return getDataInternal(zkw
, znode
, null, true, throwOnInterrupt
);
656 * Get the data at the specified znode and set a watch.
658 * Returns the data and sets a watch if the node exists. Returns null and no
659 * watch is set if the node does not exist or there is an exception.
661 * @param zkw zk reference
662 * @param znode path of node
663 * @param stat object to populate the version of the znode
664 * @return data of the specified znode, or null
665 * @throws KeeperException if unexpected zookeeper exception
667 public static byte[] getDataAndWatch(ZKWatcher zkw
, String znode
,
668 Stat stat
) throws KeeperException
{
669 return getDataInternal(zkw
, znode
, stat
, true, true);
672 private static byte[] getDataInternal(ZKWatcher zkw
, String znode
, Stat stat
, boolean watcherSet
,
673 boolean throwOnInterrupt
)
674 throws KeeperException
{
676 byte [] data
= zkw
.getRecoverableZooKeeper().getData(znode
, zkw
, stat
);
677 logRetrievedMsg(zkw
, znode
, data
, watcherSet
);
679 } catch (KeeperException
.NoNodeException e
) {
680 // This log can get pretty annoying when we cycle on 100ms waits.
681 // Enable trace if you really want to see it.
682 LOG
.trace(zkw
.prefix("Unable to get data of znode " + znode
+ " " +
683 "because node does not exist (not an error)"));
685 } catch (KeeperException e
) {
686 LOG
.warn(zkw
.prefix("Unable to get data of znode " + znode
), e
);
687 zkw
.keeperException(e
);
689 } catch (InterruptedException e
) {
690 LOG
.warn(zkw
.prefix("Unable to get data of znode " + znode
), e
);
691 if (throwOnInterrupt
) {
692 zkw
.interruptedException(e
);
694 zkw
.interruptedExceptionNoThrow(e
, true);
701 * Get the data at the specified znode without setting a watch.
703 * Returns the data if the node exists. Returns null if the node does not
706 * Sets the stats of the node in the passed Stat object. Pass a null stat if
709 * @param zkw zk reference
710 * @param znode path of node
711 * @param stat node status to get if node exists
712 * @return data of the specified znode, or null if node does not exist
713 * @throws KeeperException if unexpected zookeeper exception
715 public static byte [] getDataNoWatch(ZKWatcher zkw
, String znode
,
717 throws KeeperException
{
719 byte [] data
= zkw
.getRecoverableZooKeeper().getData(znode
, null, stat
);
720 logRetrievedMsg(zkw
, znode
, data
, false);
722 } catch (KeeperException
.NoNodeException e
) {
723 LOG
.debug(zkw
.prefix("Unable to get data of znode " + znode
+ " " +
724 "because node does not exist (not necessarily an error)"));
726 } catch (KeeperException e
) {
727 LOG
.warn(zkw
.prefix("Unable to get data of znode " + znode
), e
);
728 zkw
.keeperException(e
);
730 } catch (InterruptedException e
) {
731 LOG
.warn(zkw
.prefix("Unable to get data of znode " + znode
), e
);
732 zkw
.interruptedException(e
);
738 * Returns the date of child znodes of the specified znode. Also sets a watch on
739 * the specified znode which will capture a NodeDeleted event on the specified
740 * znode as well as NodeChildrenChanged if any children of the specified znode
741 * are created or deleted.
743 * Returns null if the specified node does not exist. Otherwise returns a
744 * list of children of the specified node. If the node exists but it has no
745 * children, an empty list will be returned.
747 * @param zkw zk reference
748 * @param baseNode path of node to list and watch children of
749 * @return list of data of children of the specified node, an empty list if the node
750 * exists but has no children, and null if the node does not exist
751 * @throws KeeperException if unexpected zookeeper exception
755 public static List
<NodeAndData
> getChildDataAndWatchForNewChildren(ZKWatcher zkw
, String baseNode
)
756 throws KeeperException
{
757 return getChildDataAndWatchForNewChildren(zkw
, baseNode
, true);
761 * Returns the date of child znodes of the specified znode. Also sets a watch on
762 * the specified znode which will capture a NodeDeleted event on the specified
763 * znode as well as NodeChildrenChanged if any children of the specified znode
764 * are created or deleted.
766 * Returns null if the specified node does not exist. Otherwise returns a
767 * list of children of the specified node. If the node exists but it has no
768 * children, an empty list will be returned.
770 * @param zkw zk reference
771 * @param baseNode path of node to list and watch children of
772 * @param throwOnInterrupt if true then just interrupt the thread, do not throw exception
773 * @return list of data of children of the specified node, an empty list if the node
774 * exists but has no children, and null if the node does not exist
775 * @throws KeeperException if unexpected zookeeper exception
779 public static List
<NodeAndData
> getChildDataAndWatchForNewChildren(
780 ZKWatcher zkw
, String baseNode
, boolean throwOnInterrupt
) throws KeeperException
{
782 ZKUtil
.listChildrenAndWatchForNewChildren(zkw
, baseNode
);
784 List
<NodeAndData
> newNodes
= new ArrayList
<>();
785 for (String node
: nodes
) {
786 if (Thread
.interrupted()) {
787 // Partial data should not be processed. Cancel processing by sending empty list.
788 return Collections
.emptyList();
790 String nodePath
= ZNodePaths
.joinZNode(baseNode
, node
);
791 byte[] data
= ZKUtil
.getDataAndWatch(zkw
, nodePath
, throwOnInterrupt
);
792 newNodes
.add(new NodeAndData(nodePath
, data
));
800 * Update the data of an existing node with the expected version to have the
803 * Throws an exception if there is a version mismatch or some other problem.
805 * Sets no watches under any conditions.
807 * @param zkw zk reference
808 * @param znode the path to the ZNode
809 * @param data the data to store in ZooKeeper
810 * @param expectedVersion the expected version
811 * @throws KeeperException if unexpected zookeeper exception
812 * @throws KeeperException.BadVersionException if version mismatch
816 public static void updateExistingNodeData(ZKWatcher zkw
, String znode
, byte[] data
,
817 int expectedVersion
) throws KeeperException
{
819 zkw
.getRecoverableZooKeeper().setData(znode
, data
, expectedVersion
);
820 } catch(InterruptedException ie
) {
821 zkw
.interruptedException(ie
);
830 * Sets the data of the existing znode to be the specified data. Ensures that
831 * the current data has the specified expected version.
833 * <p>If the node does not exist, a {@link NoNodeException} will be thrown.
835 * <p>If their is a version mismatch, method returns null.
837 * <p>No watches are set but setting data will trigger other watchers of this
840 * <p>If there is another problem, a KeeperException will be thrown.
842 * @param zkw zk reference
843 * @param znode path of node
844 * @param data data to set for node
845 * @param expectedVersion version expected when setting data
846 * @return true if data set, false if version mismatch
847 * @throws KeeperException if unexpected zookeeper exception
849 public static boolean setData(ZKWatcher zkw
, String znode
,
850 byte [] data
, int expectedVersion
)
851 throws KeeperException
, KeeperException
.NoNodeException
{
853 return zkw
.getRecoverableZooKeeper().setData(znode
, data
, expectedVersion
) != null;
854 } catch (InterruptedException e
) {
855 zkw
.interruptedException(e
);
861 * Set data into node creating node if it doesn't yet exist.
862 * Does not set watch.
864 * @param zkw zk reference
865 * @param znode path of node
866 * @param data data to set for node
867 * @throws KeeperException if a ZooKeeper operation fails
869 public static void createSetData(final ZKWatcher zkw
, final String znode
, final byte [] data
)
870 throws KeeperException
{
871 if (checkExists(zkw
, znode
) == -1) {
872 ZKUtil
.createWithParents(zkw
, znode
, data
);
874 ZKUtil
.setData(zkw
, znode
, data
);
879 * Sets the data of the existing znode to be the specified data. The node
880 * must exist but no checks are done on the existing data or version.
882 * <p>If the node does not exist, a {@link NoNodeException} will be thrown.
884 * <p>No watches are set but setting data will trigger other watchers of this
887 * <p>If there is another problem, a KeeperException will be thrown.
889 * @param zkw zk reference
890 * @param znode path of node
891 * @param data data to set for node
892 * @throws KeeperException if unexpected zookeeper exception
894 public static void setData(ZKWatcher zkw
, String znode
, byte [] data
)
895 throws KeeperException
, KeeperException
.NoNodeException
{
896 setData(zkw
, (SetData
)ZKUtilOp
.setData(znode
, data
));
899 private static void setData(ZKWatcher zkw
, SetData setData
)
900 throws KeeperException
, KeeperException
.NoNodeException
{
901 SetDataRequest sd
= (SetDataRequest
)toZooKeeperOp(zkw
, setData
).toRequestRecord();
902 setData(zkw
, sd
.getPath(), sd
.getData(), sd
.getVersion());
906 * Returns whether or not secure authentication is enabled
907 * (whether <code>hbase.security.authentication</code> is set to
908 * <code>kerberos</code>.
910 public static boolean isSecureZooKeeper(Configuration conf
) {
911 // Detection for embedded HBase client with jaas configuration
912 // defined for third party programs.
914 javax
.security
.auth
.login
.Configuration testConfig
=
915 javax
.security
.auth
.login
.Configuration
.getConfiguration();
916 if (testConfig
.getAppConfigurationEntry("Client") == null
917 && testConfig
.getAppConfigurationEntry(
918 JaasConfiguration
.CLIENT_KEYTAB_KERBEROS_CONFIG_NAME
) == null
919 && testConfig
.getAppConfigurationEntry(
920 JaasConfiguration
.SERVER_KEYTAB_KERBEROS_CONFIG_NAME
) == null
921 && conf
.get(HConstants
.ZK_CLIENT_KERBEROS_PRINCIPAL
) == null
922 && conf
.get(HConstants
.ZK_SERVER_KERBEROS_PRINCIPAL
) == null) {
926 } catch(Exception e
) {
927 // No Jaas configuration defined.
931 // Master & RSs uses hbase.zookeeper.client.*
932 return "kerberos".equalsIgnoreCase(conf
.get("hbase.security.authentication"));
935 private static ArrayList
<ACL
> createACL(ZKWatcher zkw
, String node
) {
936 return createACL(zkw
, node
, isSecureZooKeeper(zkw
.getConfiguration()));
939 public static ArrayList
<ACL
> createACL(ZKWatcher zkw
, String node
,
940 boolean isSecureZooKeeper
) {
941 if (!node
.startsWith(zkw
.getZNodePaths().baseZNode
)) {
942 return Ids
.OPEN_ACL_UNSAFE
;
944 if (isSecureZooKeeper
) {
945 ArrayList
<ACL
> acls
= new ArrayList
<>();
946 // add permission to hbase supper user
947 String
[] superUsers
= zkw
.getConfiguration().getStrings(Superusers
.SUPERUSER_CONF_KEY
);
948 String hbaseUser
= null;
950 hbaseUser
= UserGroupInformation
.getCurrentUser().getShortUserName();
951 } catch (IOException e
) {
952 LOG
.debug("Could not acquire current User.", e
);
954 if (superUsers
!= null) {
955 List
<String
> groups
= new ArrayList
<>();
956 for (String user
: superUsers
) {
957 if (AuthUtil
.isGroupPrincipal(user
)) {
958 // TODO: Set node ACL for groups when ZK supports this feature
961 if(!user
.equals(hbaseUser
)) {
962 acls
.add(new ACL(Perms
.ALL
, new Id("sasl", user
)));
966 if (!groups
.isEmpty()) {
967 LOG
.warn("Znode ACL setting for group {} is skipped, ZooKeeper doesn't support this " +
968 "feature presently.", groups
);
971 // Certain znodes are accessed directly by the client,
972 // so they must be readable by non-authenticated clients
973 if (zkw
.getZNodePaths().isClientReadable(node
)) {
974 acls
.addAll(Ids
.CREATOR_ALL_ACL
);
975 acls
.addAll(Ids
.READ_ACL_UNSAFE
);
977 acls
.addAll(Ids
.CREATOR_ALL_ACL
);
981 return Ids
.OPEN_ACL_UNSAFE
;
991 * Set the specified znode to be an ephemeral node carrying the specified
994 * If the node is created successfully, a watcher is also set on the node.
996 * If the node is not created successfully because it already exists, this
997 * method will also set a watcher on the node.
999 * If there is another problem, a KeeperException will be thrown.
1001 * @param zkw zk reference
1002 * @param znode path of node
1003 * @param data data of node
1004 * @return true if node created, false if not, watch set in both cases
1005 * @throws KeeperException if unexpected zookeeper exception
1007 public static boolean createEphemeralNodeAndWatch(ZKWatcher zkw
,
1008 String znode
, byte [] data
)
1009 throws KeeperException
{
1012 zkw
.getRecoverableZooKeeper().create(znode
, data
, createACL(zkw
, znode
),
1013 CreateMode
.EPHEMERAL
);
1014 } catch (KeeperException
.NodeExistsException nee
) {
1016 } catch (InterruptedException e
) {
1017 LOG
.info("Interrupted", e
);
1018 Thread
.currentThread().interrupt();
1020 if(!watchAndCheckExists(zkw
, znode
)) {
1021 // It did exist but now it doesn't, try again
1022 return createEphemeralNodeAndWatch(zkw
, znode
, data
);
1028 * Creates the specified znode to be a persistent node carrying the specified
1031 * Returns true if the node was successfully created, false if the node
1034 * If the node is created successfully, a watcher is also set on the node.
1036 * If the node is not created successfully because it already exists, this
1037 * method will also set a watcher on the node but return false.
1039 * If there is another problem, a KeeperException will be thrown.
1041 * @param zkw zk reference
1042 * @param znode path of node
1043 * @param data data of node
1044 * @return true if node created, false if not, watch set in both cases
1045 * @throws KeeperException if unexpected zookeeper exception
1047 public static boolean createNodeIfNotExistsAndWatch(
1048 ZKWatcher zkw
, String znode
, byte [] data
)
1049 throws KeeperException
{
1052 zkw
.getRecoverableZooKeeper().create(znode
, data
, createACL(zkw
, znode
),
1053 CreateMode
.PERSISTENT
);
1054 } catch (KeeperException
.NodeExistsException nee
) {
1056 } catch (InterruptedException e
) {
1057 zkw
.interruptedException(e
);
1061 zkw
.getRecoverableZooKeeper().exists(znode
, zkw
);
1062 } catch (InterruptedException e
) {
1063 zkw
.interruptedException(e
);
1070 * Creates the specified znode with the specified data but does not watch it.
1072 * Returns the znode of the newly created node
1074 * If there is another problem, a KeeperException will be thrown.
1076 * @param zkw zk reference
1077 * @param znode path of node
1078 * @param data data of node
1079 * @param createMode specifying whether the node to be created is ephemeral and/or sequential
1080 * @return true name of the newly created znode or null
1081 * @throws KeeperException if unexpected zookeeper exception
1083 public static String
createNodeIfNotExistsNoWatch(ZKWatcher zkw
, String znode
, byte[] data
,
1084 CreateMode createMode
) throws KeeperException
{
1085 String createdZNode
= null;
1087 createdZNode
= zkw
.getRecoverableZooKeeper().create(znode
, data
,
1088 createACL(zkw
, znode
), createMode
);
1089 } catch (KeeperException
.NodeExistsException nee
) {
1091 } catch (InterruptedException e
) {
1092 zkw
.interruptedException(e
);
1095 return createdZNode
;
1099 * Creates the specified node with the specified data and watches it.
1101 * <p>Throws an exception if the node already exists.
1103 * <p>The node created is persistent and open access.
1105 * <p>Returns the version number of the created node if successful.
1107 * @param zkw zk reference
1108 * @param znode path of node to create
1109 * @param data data of node to create
1110 * @return version of node created
1111 * @throws KeeperException if unexpected zookeeper exception
1112 * @throws KeeperException.NodeExistsException if node already exists
1114 public static int createAndWatch(ZKWatcher zkw
,
1115 String znode
, byte [] data
)
1116 throws KeeperException
, KeeperException
.NodeExistsException
{
1118 zkw
.getRecoverableZooKeeper().create(znode
, data
, createACL(zkw
, znode
),
1119 CreateMode
.PERSISTENT
);
1120 Stat stat
= zkw
.getRecoverableZooKeeper().exists(znode
, zkw
);
1122 // Likely a race condition. Someone deleted the znode.
1123 throw KeeperException
.create(KeeperException
.Code
.SYSTEMERROR
,
1124 "ZK.exists returned null (i.e.: znode does not exist) for znode=" + znode
);
1127 return stat
.getVersion();
1128 } catch (InterruptedException e
) {
1129 zkw
.interruptedException(e
);
1135 * Async creates the specified node with the specified data.
1137 * <p>Throws an exception if the node already exists.
1139 * <p>The node created is persistent and open access.
1141 * @param zkw zk reference
1142 * @param znode path of node to create
1143 * @param data data of node to create
1144 * @param cb the callback to use for the creation
1145 * @param ctx the context to use for the creation
1147 public static void asyncCreate(ZKWatcher zkw
,
1148 String znode
, byte [] data
, final AsyncCallback
.StringCallback cb
,
1150 zkw
.getRecoverableZooKeeper().getZooKeeper().create(znode
, data
,
1151 createACL(zkw
, znode
), CreateMode
.PERSISTENT
, cb
, ctx
);
1155 * Creates the specified node, iff the node does not exist. Does not set a
1156 * watch and fails silently if the node already exists.
1158 * The node created is persistent and open access.
1160 * @param zkw zk reference
1161 * @param znode path of node
1162 * @throws KeeperException if unexpected zookeeper exception
1164 public static void createAndFailSilent(ZKWatcher zkw
,
1165 String znode
) throws KeeperException
{
1166 createAndFailSilent(zkw
, znode
, new byte[0]);
1170 * Creates the specified node containing specified data, iff the node does not exist. Does
1171 * not set a watch and fails silently if the node already exists.
1173 * The node created is persistent and open access.
1175 * @param zkw zk reference
1176 * @param znode path of node
1177 * @param data a byte array data to store in the znode
1178 * @throws KeeperException if unexpected zookeeper exception
1180 public static void createAndFailSilent(ZKWatcher zkw
,
1181 String znode
, byte[] data
)
1182 throws KeeperException
{
1183 createAndFailSilent(zkw
,
1184 (CreateAndFailSilent
)ZKUtilOp
.createAndFailSilent(znode
, data
));
1187 private static void createAndFailSilent(ZKWatcher zkw
, CreateAndFailSilent cafs
)
1188 throws KeeperException
{
1189 CreateRequest create
= (CreateRequest
)toZooKeeperOp(zkw
, cafs
).toRequestRecord();
1190 String znode
= create
.getPath();
1192 RecoverableZooKeeper zk
= zkw
.getRecoverableZooKeeper();
1193 if (zk
.exists(znode
, false) == null) {
1194 zk
.create(znode
, create
.getData(), create
.getAcl(), CreateMode
.fromFlag(create
.getFlags()));
1196 } catch(KeeperException
.NodeExistsException nee
) {
1197 } catch(KeeperException
.NoAuthException nee
){
1199 if (null == zkw
.getRecoverableZooKeeper().exists(znode
, false)) {
1200 // If we failed to create the file and it does not already exist.
1203 } catch (InterruptedException ie
) {
1204 zkw
.interruptedException(ie
);
1206 } catch(InterruptedException ie
) {
1207 zkw
.interruptedException(ie
);
1212 * Creates the specified node and all parent nodes required for it to exist.
1214 * No watches are set and no errors are thrown if the node already exists.
1216 * The nodes created are persistent and open access.
1218 * @param zkw zk reference
1219 * @param znode path of node
1220 * @throws KeeperException if unexpected zookeeper exception
1222 public static void createWithParents(ZKWatcher zkw
, String znode
)
1223 throws KeeperException
{
1224 createWithParents(zkw
, znode
, new byte[0]);
1228 * Creates the specified node and all parent nodes required for it to exist. The creation of
1229 * parent znodes is not atomic with the leafe znode creation but the data is written atomically
1230 * when the leaf node is created.
1232 * No watches are set and no errors are thrown if the node already exists.
1234 * The nodes created are persistent and open access.
1236 * @param zkw zk reference
1237 * @param znode path of node
1238 * @throws KeeperException if unexpected zookeeper exception
1240 public static void createWithParents(ZKWatcher zkw
, String znode
, byte[] data
)
1241 throws KeeperException
{
1246 zkw
.getRecoverableZooKeeper().create(znode
, data
, createACL(zkw
, znode
),
1247 CreateMode
.PERSISTENT
);
1248 } catch(KeeperException
.NodeExistsException nee
) {
1250 } catch(KeeperException
.NoNodeException nne
) {
1251 createWithParents(zkw
, getParent(znode
));
1252 createWithParents(zkw
, znode
, data
);
1253 } catch(InterruptedException ie
) {
1254 zkw
.interruptedException(ie
);
1263 * Delete the specified node. Sets no watches. Throws all exceptions.
1265 public static void deleteNode(ZKWatcher zkw
, String node
)
1266 throws KeeperException
{
1267 deleteNode(zkw
, node
, -1);
1271 * Delete the specified node with the specified version. Sets no watches.
1272 * Throws all exceptions.
1274 public static boolean deleteNode(ZKWatcher zkw
, String node
,
1276 throws KeeperException
{
1278 zkw
.getRecoverableZooKeeper().delete(node
, version
);
1280 } catch(KeeperException
.BadVersionException bve
) {
1282 } catch(InterruptedException ie
) {
1283 zkw
.interruptedException(ie
);
1289 * Deletes the specified node. Fails silent if the node does not exist.
1291 * @param zkw reference to the {@link ZKWatcher} which also contains configuration and operation
1292 * @param node the node to delete
1293 * @throws KeeperException if a ZooKeeper operation fails
1295 public static void deleteNodeFailSilent(ZKWatcher zkw
, String node
)
1296 throws KeeperException
{
1297 deleteNodeFailSilent(zkw
,
1298 (DeleteNodeFailSilent
)ZKUtilOp
.deleteNodeFailSilent(node
));
1301 private static void deleteNodeFailSilent(ZKWatcher zkw
,
1302 DeleteNodeFailSilent dnfs
) throws KeeperException
{
1303 DeleteRequest delete
= (DeleteRequest
)toZooKeeperOp(zkw
, dnfs
).toRequestRecord();
1305 zkw
.getRecoverableZooKeeper().delete(delete
.getPath(), delete
.getVersion());
1306 } catch(KeeperException
.NoNodeException nne
) {
1307 } catch(InterruptedException ie
) {
1308 zkw
.interruptedException(ie
);
1314 * Delete the specified node and all of it's children.
1316 * If the node does not exist, just returns.
1318 * Sets no watches. Throws all exceptions besides dealing with deletion of
1321 public static void deleteNodeRecursively(ZKWatcher zkw
, String node
)
1322 throws KeeperException
{
1323 deleteNodeRecursivelyMultiOrSequential(zkw
, true, node
);
1327 * Delete all the children of the specified node but not the node itself.
1329 * Sets no watches. Throws all exceptions besides dealing with deletion of
1332 * @throws KeeperException if a ZooKeeper operation fails
1334 public static void deleteChildrenRecursively(ZKWatcher zkw
, String node
)
1335 throws KeeperException
{
1336 deleteChildrenRecursivelyMultiOrSequential(zkw
, true, node
);
1340 * Delete all the children of the specified node but not the node itself. This
1341 * will first traverse the znode tree for listing the children and then delete
1342 * these znodes using multi-update api or sequential based on the specified
1345 * Sets no watches. Throws all exceptions besides dealing with deletion of
1348 * If the following is true:
1350 * <li>runSequentialOnMultiFailure is true
1352 * on calling multi, we get a ZooKeeper exception that can be handled by a
1353 * sequential call(*), we retry the operations one-by-one (sequentially).
1357 * @param runSequentialOnMultiFailure
1358 * - if true when we get a ZooKeeper exception that could retry the
1359 * operations one-by-one (sequentially)
1361 * - path of the parent node(s)
1362 * @throws KeeperException.NotEmptyException
1363 * if node has children while deleting
1364 * @throws KeeperException
1365 * if unexpected ZooKeeper exception
1366 * @throws IllegalArgumentException
1367 * if an invalid path is specified
1369 public static void deleteChildrenRecursivelyMultiOrSequential(
1370 ZKWatcher zkw
, boolean runSequentialOnMultiFailure
,
1371 String
... pathRoots
) throws KeeperException
{
1372 if (pathRoots
== null || pathRoots
.length
<= 0) {
1373 LOG
.warn("Given path is not valid!");
1376 List
<ZKUtilOp
> ops
= new ArrayList
<>();
1377 for (String eachRoot
: pathRoots
) {
1378 List
<String
> children
= listChildrenBFSNoWatch(zkw
, eachRoot
);
1379 // Delete the leaves first and eventually get rid of the root
1380 for (int i
= children
.size() - 1; i
>= 0; --i
) {
1381 ops
.add(ZKUtilOp
.deleteNodeFailSilent(children
.get(i
)));
1384 submitBatchedMultiOrSequential(zkw
, runSequentialOnMultiFailure
, ops
);
1388 * Delete the specified node and its children. This traverse the
1389 * znode tree for listing the children and then delete
1390 * these znodes including the parent using multi-update api or
1391 * sequential based on the specified configurations.
1393 * Sets no watches. Throws all exceptions besides dealing with deletion of
1396 * If the following is true:
1398 * <li>runSequentialOnMultiFailure is true
1400 * on calling multi, we get a ZooKeeper exception that can be handled by a
1401 * sequential call(*), we retry the operations one-by-one (sequentially).
1405 * @param runSequentialOnMultiFailure
1406 * - if true when we get a ZooKeeper exception that could retry the
1407 * operations one-by-one (sequentially)
1409 * - path of the parent node(s)
1410 * @throws KeeperException.NotEmptyException
1411 * if node has children while deleting
1412 * @throws KeeperException
1413 * if unexpected ZooKeeper exception
1414 * @throws IllegalArgumentException
1415 * if an invalid path is specified
1417 public static void deleteNodeRecursivelyMultiOrSequential(ZKWatcher zkw
,
1418 boolean runSequentialOnMultiFailure
, String
... pathRoots
) throws KeeperException
{
1419 if (pathRoots
== null || pathRoots
.length
<= 0) {
1420 LOG
.warn("Given path is not valid!");
1423 List
<ZKUtilOp
> ops
= new ArrayList
<>();
1424 for (String eachRoot
: pathRoots
) {
1425 // ZooKeeper Watches are one time triggers; When children of parent nodes are deleted
1426 // recursively, must set another watch, get notified of delete node
1427 List
<String
> children
= listChildrenBFSAndWatchThem(zkw
, eachRoot
);
1428 // Delete the leaves first and eventually get rid of the root
1429 for (int i
= children
.size() - 1; i
>= 0; --i
) {
1430 ops
.add(ZKUtilOp
.deleteNodeFailSilent(children
.get(i
)));
1433 if (zkw
.getRecoverableZooKeeper().exists(eachRoot
, zkw
) != null) {
1434 ops
.add(ZKUtilOp
.deleteNodeFailSilent(eachRoot
));
1436 } catch (InterruptedException e
) {
1437 zkw
.interruptedException(e
);
1440 submitBatchedMultiOrSequential(zkw
, runSequentialOnMultiFailure
, ops
);
1444 * Chunks the provided {@code ops} when their approximate size exceeds the the configured limit.
1445 * Take caution that this can ONLY be used for operations where atomicity is not important,
1446 * e.g. deletions. It must not be used when atomicity of the operations is critical.
1448 * @param zkw reference to the {@link ZKWatcher} which contains configuration and constants
1449 * @param runSequentialOnMultiFailure if true when we get a ZooKeeper exception that could
1450 * retry the operations one-by-one (sequentially)
1451 * @param ops list of ZKUtilOp {@link ZKUtilOp} to partition while submitting batched multi
1453 * @throws KeeperException unexpected ZooKeeper Exception / Zookeeper unreachable
1455 private static void submitBatchedMultiOrSequential(ZKWatcher zkw
,
1456 boolean runSequentialOnMultiFailure
, List
<ZKUtilOp
> ops
) throws KeeperException
{
1457 // at least one element should exist
1458 if (ops
.isEmpty()) {
1461 final int maxMultiSize
= zkw
.getRecoverableZooKeeper().getMaxMultiSizeLimit();
1462 // Batch up the items to over smashing through jute.maxbuffer with too many Ops.
1463 final List
<List
<ZKUtilOp
>> batchedOps
= partitionOps(ops
, maxMultiSize
);
1464 // Would use forEach() but have to handle KeeperException
1465 for (List
<ZKUtilOp
> batch
: batchedOps
) {
1466 multiOrSequential(zkw
, batch
, runSequentialOnMultiFailure
);
1471 * Partition the list of {@code ops} by size (using {@link #estimateSize(ZKUtilOp)}).
1473 static List
<List
<ZKUtilOp
>> partitionOps(List
<ZKUtilOp
> ops
, int maxPartitionSize
) {
1474 List
<List
<ZKUtilOp
>> partitionedOps
= new ArrayList
<>();
1475 List
<ZKUtilOp
> currentPartition
= new ArrayList
<>();
1476 int currentPartitionSize
= 0;
1477 partitionedOps
.add(currentPartition
);
1478 Iterator
<ZKUtilOp
> iter
= ops
.iterator();
1479 while (iter
.hasNext()) {
1480 ZKUtilOp currentOp
= iter
.next();
1481 int currentOpSize
= estimateSize(currentOp
);
1483 // Roll a new partition if necessary
1484 // If the current partition is empty, put the element in there anyways.
1485 // We can roll a new partition if we get another element
1486 if (!currentPartition
.isEmpty() && currentOpSize
+ currentPartitionSize
> maxPartitionSize
) {
1487 currentPartition
= new ArrayList
<>();
1488 partitionedOps
.add(currentPartition
);
1489 currentPartitionSize
= 0;
1492 // Add the current op to the partition
1493 currentPartition
.add(currentOp
);
1494 // And record its size
1495 currentPartitionSize
+= currentOpSize
;
1497 return partitionedOps
;
1500 static int estimateSize(ZKUtilOp op
) {
1501 return Bytes
.toBytes(op
.getPath()).length
;
1505 * BFS Traversal of all the children under path, with the entries in the list,
1506 * in the same order as that of the traversal. Lists all the children without
1507 * setting any watches.
1513 * @return list of children znodes under the path
1514 * @throws KeeperException
1515 * if unexpected ZooKeeper exception
1517 private static List
<String
> listChildrenBFSNoWatch(ZKWatcher zkw
,
1518 final String znode
) throws KeeperException
{
1519 Deque
<String
> queue
= new LinkedList
<>();
1520 List
<String
> tree
= new ArrayList
<>();
1523 String node
= queue
.pollFirst();
1527 List
<String
> children
= listChildrenNoWatch(zkw
, node
);
1528 if (children
== null) {
1531 for (final String child
: children
) {
1532 final String childPath
= node
+ "/" + child
;
1533 queue
.add(childPath
);
1534 tree
.add(childPath
);
1541 * BFS Traversal of all the children under path, with the entries in the list,
1542 * in the same order as that of the traversal.
1543 * Lists all the children and set watches on to them.
1549 * @return list of children znodes under the path
1550 * @throws KeeperException
1551 * if unexpected ZooKeeper exception
1553 private static List
<String
> listChildrenBFSAndWatchThem(ZKWatcher zkw
, final String znode
)
1554 throws KeeperException
{
1555 Deque
<String
> queue
= new LinkedList
<>();
1556 List
<String
> tree
= new ArrayList
<>();
1559 String node
= queue
.pollFirst();
1563 List
<String
> children
= listChildrenAndWatchThem(zkw
, node
);
1564 if (children
== null) {
1567 for (final String child
: children
) {
1568 final String childPath
= node
+ "/" + child
;
1569 queue
.add(childPath
);
1570 tree
.add(childPath
);
1577 * Represents an action taken by ZKUtil, e.g. createAndFailSilent.
1578 * These actions are higher-level than ZKOp actions, which represent
1579 * individual actions in the ZooKeeper API, like create.
1581 public abstract static class ZKUtilOp
{
1582 private String path
;
1584 @Override public String
toString() {
1585 return this.getClass().getSimpleName() + ", path=" + this.path
;
1588 private ZKUtilOp(String path
) {
1593 * @return a createAndFailSilent ZKUtilOp
1595 public static ZKUtilOp
createAndFailSilent(String path
, byte[] data
) {
1596 return new CreateAndFailSilent(path
, data
);
1600 * @return a deleteNodeFailSilent ZKUtilOP
1602 public static ZKUtilOp
deleteNodeFailSilent(String path
) {
1603 return new DeleteNodeFailSilent(path
);
1607 * @return a setData ZKUtilOp
1609 public static ZKUtilOp
setData(String path
, byte[] data
) {
1610 return new SetData(path
, data
);
1614 * @return a setData ZKUtilOp
1616 public static ZKUtilOp
setData(String path
, byte[] data
, int version
) {
1617 return new SetData(path
, data
, version
);
1621 * @return path to znode where the ZKOp will occur
1623 public String
getPath() {
1628 * ZKUtilOp representing createAndFailSilent in ZooKeeper
1629 * (attempt to create node, ignore error if already exists)
1631 public static final class CreateAndFailSilent
extends ZKUtilOp
{
1632 private byte [] data
;
1634 private CreateAndFailSilent(String path
, byte [] data
) {
1639 public byte[] getData() {
1644 public boolean equals(Object o
) {
1648 if (!(o
instanceof CreateAndFailSilent
)) {
1652 CreateAndFailSilent op
= (CreateAndFailSilent
) o
;
1653 return getPath().equals(op
.getPath()) && Arrays
.equals(data
, op
.data
);
1657 public int hashCode() {
1658 int ret
= 17 + getPath().hashCode() * 31;
1659 return ret
* 31 + Bytes
.hashCode(data
);
1664 * ZKUtilOp representing deleteNodeFailSilent in ZooKeeper
1665 * (attempt to delete node, ignore error if node doesn't exist)
1667 public static final class DeleteNodeFailSilent
extends ZKUtilOp
{
1668 private DeleteNodeFailSilent(String path
) {
1673 public boolean equals(Object o
) {
1677 if (!(o
instanceof DeleteNodeFailSilent
)) {
1681 return super.equals(o
);
1685 public int hashCode() {
1686 return getPath().hashCode();
1691 * ZKUtilOp representing setData in ZooKeeper
1693 public static final class SetData
extends ZKUtilOp
{
1694 private byte[] data
;
1695 private int version
= -1;
1697 private SetData(String path
, byte[] data
) {
1702 private SetData(String path
, byte[] data
, int version
) {
1705 this.version
= version
;
1708 public byte[] getData() {
1712 public int getVersion() {
1717 public boolean equals(Object o
) {
1721 if (!(o
instanceof SetData
)) {
1725 SetData op
= (SetData
) o
;
1726 return getPath().equals(op
.getPath()) && Arrays
.equals(data
, op
.data
)
1727 && getVersion() == op
.getVersion();
1731 public int hashCode() {
1732 int ret
= getPath().hashCode();
1733 ret
= ret
* 31 + Bytes
.hashCode(data
);
1734 return ret
* 31 + Integer
.hashCode(version
);
1740 * Convert from ZKUtilOp to ZKOp
1742 private static Op
toZooKeeperOp(ZKWatcher zkw
, ZKUtilOp op
) throws UnsupportedOperationException
{
1747 if (op
instanceof CreateAndFailSilent
) {
1748 CreateAndFailSilent cafs
= (CreateAndFailSilent
)op
;
1749 return Op
.create(cafs
.getPath(), cafs
.getData(), createACL(zkw
, cafs
.getPath()),
1750 CreateMode
.PERSISTENT
);
1751 } else if (op
instanceof DeleteNodeFailSilent
) {
1752 DeleteNodeFailSilent dnfs
= (DeleteNodeFailSilent
)op
;
1753 return Op
.delete(dnfs
.getPath(), -1);
1754 } else if (op
instanceof SetData
) {
1755 SetData sd
= (SetData
) op
;
1756 return Op
.setData(sd
.getPath(), sd
.getData(), sd
.getVersion());
1758 throw new UnsupportedOperationException("Unexpected ZKUtilOp type: "
1759 + op
.getClass().getName());
1763 // Static boolean for warning about useMulti because ideally there will be one warning per
1764 // process instance. It is fine if two threads end up racing on this for a bit. We do not
1765 // need to use an atomic type for this, that is overkill. The goal of reducing the number of
1766 // warnings from many to one or a few at startup is still achieved.
1767 private static boolean useMultiWarn
= true;
1770 * Use ZooKeeper's multi-update functionality.
1772 * If all of the following are true:
1773 * - runSequentialOnMultiFailure is true
1774 * - on calling multi, we get a ZooKeeper exception that can be handled by a sequential call(*)
1776 * - we retry the operations one-by-one (sequentially)
1778 * Note *: an example is receiving a NodeExistsException from a "create" call. Without multi,
1779 * a user could call "createAndFailSilent" to ensure that a node exists if they don't care who
1780 * actually created the node (i.e. the NodeExistsException from ZooKeeper is caught).
1781 * This will cause all operations in the multi to fail, however, because
1782 * the NodeExistsException that zk.create throws will fail the multi transaction.
1783 * In this case, if the previous conditions hold, the commands are run sequentially, which should
1784 * result in the correct final state, but means that the operations will not run atomically.
1786 * @throws KeeperException if a ZooKeeper operation fails
1788 public static void multiOrSequential(ZKWatcher zkw
, List
<ZKUtilOp
> ops
,
1789 boolean runSequentialOnMultiFailure
) throws KeeperException
{
1793 if (useMultiWarn
) { // Only check and warn at first use
1794 if (zkw
.getConfiguration().get("hbase.zookeeper.useMulti") != null) {
1795 LOG
.warn("hbase.zookeeper.useMulti is deprecated. Default to true always.");
1797 useMultiWarn
= false;
1799 List
<Op
> zkOps
= new LinkedList
<>();
1800 for (ZKUtilOp op
: ops
) {
1801 zkOps
.add(toZooKeeperOp(zkw
, op
));
1804 zkw
.getRecoverableZooKeeper().multi(zkOps
);
1805 } catch (KeeperException ke
) {
1806 switch (ke
.code()) {
1812 // if we get an exception that could be solved by running sequentially
1813 // (and the client asked us to), then break out and run sequentially
1814 if (runSequentialOnMultiFailure
) {
1815 LOG
.info("multi exception: {}; running operations sequentially " +
1816 "(runSequentialOnMultiFailure=true); {}", ke
.toString(),
1817 ops
.stream().map(o
-> o
.toString()).collect(Collectors
.joining(",")));
1818 processSequentially(zkw
, ops
);
1824 } catch (InterruptedException ie
) {
1825 zkw
.interruptedException(ie
);
1829 private static void processSequentially(ZKWatcher zkw
, List
<ZKUtilOp
> ops
)
1830 throws KeeperException
, NoNodeException
{
1831 for (ZKUtilOp op
: ops
) {
1832 if (op
instanceof CreateAndFailSilent
) {
1833 createAndFailSilent(zkw
, (CreateAndFailSilent
) op
);
1834 } else if (op
instanceof DeleteNodeFailSilent
) {
1835 deleteNodeFailSilent(zkw
, (DeleteNodeFailSilent
) op
);
1836 } else if (op
instanceof SetData
) {
1837 setData(zkw
, (SetData
) op
);
1839 throw new UnsupportedOperationException("Unexpected ZKUtilOp type: "
1840 + op
.getClass().getName());
1846 // ZooKeeper cluster information
1849 /** @return String dump of everything in ZooKeeper. */
1850 public static String
dump(ZKWatcher zkw
) {
1851 StringBuilder sb
= new StringBuilder();
1853 sb
.append("HBase is rooted at ").append(zkw
.getZNodePaths().baseZNode
);
1854 sb
.append("\nActive master address: ");
1856 sb
.append("\n ").append(MasterAddressTracker
.getMasterAddress(zkw
));
1857 } catch (IOException e
) {
1858 sb
.append("<<FAILED LOOKUP: " + e
.getMessage() + ">>");
1860 sb
.append("\nBackup master addresses:");
1861 final List
<String
> backupMasterChildrenNoWatchList
= listChildrenNoWatch(zkw
,
1862 zkw
.getZNodePaths().backupMasterAddressesZNode
);
1863 if (backupMasterChildrenNoWatchList
!= null) {
1864 for (String child
: backupMasterChildrenNoWatchList
) {
1865 sb
.append("\n ").append(child
);
1868 sb
.append("\nRegion server holding hbase:meta:");
1869 sb
.append("\n ").append(MetaTableLocator
.getMetaRegionLocation(zkw
));
1870 int numMetaReplicas
= zkw
.getMetaReplicaNodes().size();
1871 for (int i
= 1; i
< numMetaReplicas
; i
++) {
1872 sb
.append("\n replica" + i
+ ": "
1873 + MetaTableLocator
.getMetaRegionLocation(zkw
, i
));
1875 sb
.append("\nRegion servers:");
1876 final List
<String
> rsChildrenNoWatchList
=
1877 listChildrenNoWatch(zkw
, zkw
.getZNodePaths().rsZNode
);
1878 if (rsChildrenNoWatchList
!= null) {
1879 for (String child
: rsChildrenNoWatchList
) {
1880 sb
.append("\n ").append(child
);
1884 getReplicationZnodesDump(zkw
, sb
);
1885 } catch (KeeperException ke
) {
1886 LOG
.warn("Couldn't get the replication znode dump", ke
);
1888 sb
.append("\nQuorum Server Statistics:");
1889 String
[] servers
= zkw
.getQuorum().split(",");
1890 for (String server
: servers
) {
1891 sb
.append("\n ").append(server
);
1893 String
[] stat
= getServerStats(server
, ZKUtil
.zkDumpConnectionTimeOut
);
1896 sb
.append("[Error] invalid quorum server: " + server
);
1900 for (String s
: stat
) {
1901 sb
.append("\n ").append(s
);
1903 } catch (Exception e
) {
1904 sb
.append("\n ERROR: ").append(e
.getMessage());
1907 } catch (KeeperException ke
) {
1908 sb
.append("\nFATAL ZooKeeper Exception!\n");
1909 sb
.append("\n" + ke
.getMessage());
1911 return sb
.toString();
1915 * Appends replication znodes to the passed StringBuilder.
1917 * @param zkw reference to the {@link ZKWatcher} which also contains configuration and operation
1918 * @param sb the {@link StringBuilder} to append to
1919 * @throws KeeperException if a ZooKeeper operation fails
1921 private static void getReplicationZnodesDump(ZKWatcher zkw
, StringBuilder sb
)
1922 throws KeeperException
{
1923 String replicationZnode
= zkw
.getZNodePaths().replicationZNode
;
1925 if (ZKUtil
.checkExists(zkw
, replicationZnode
) == -1) {
1929 // do a ls -r on this znode
1930 sb
.append("\n").append(replicationZnode
).append(": ");
1931 List
<String
> children
= ZKUtil
.listChildrenNoWatch(zkw
, replicationZnode
);
1932 if (children
!= null) {
1933 Collections
.sort(children
);
1934 for (String child
: children
) {
1935 String zNode
= ZNodePaths
.joinZNode(replicationZnode
, child
);
1936 if (zNode
.equals(zkw
.getZNodePaths().peersZNode
)) {
1937 appendPeersZnodes(zkw
, zNode
, sb
);
1938 } else if (zNode
.equals(zkw
.getZNodePaths().queuesZNode
)) {
1939 appendRSZnodes(zkw
, zNode
, sb
);
1940 } else if (zNode
.equals(zkw
.getZNodePaths().hfileRefsZNode
)) {
1941 appendHFileRefsZNodes(zkw
, zNode
, sb
);
1947 private static void appendHFileRefsZNodes(ZKWatcher zkw
, String hFileRefsZNode
,
1948 StringBuilder sb
) throws KeeperException
{
1949 sb
.append("\n").append(hFileRefsZNode
).append(": ");
1950 final List
<String
> hFileRefChildrenNoWatchList
=
1951 ZKUtil
.listChildrenNoWatch(zkw
, hFileRefsZNode
);
1952 if (hFileRefChildrenNoWatchList
!= null) {
1953 for (String peerIdZNode
: hFileRefChildrenNoWatchList
) {
1954 String zNodeToProcess
= ZNodePaths
.joinZNode(hFileRefsZNode
, peerIdZNode
);
1955 sb
.append("\n").append(zNodeToProcess
).append(": ");
1956 List
<String
> peerHFileRefsZNodes
= ZKUtil
.listChildrenNoWatch(zkw
, zNodeToProcess
);
1957 if (peerHFileRefsZNodes
!= null) {
1958 sb
.append(String
.join(", ", peerHFileRefsZNodes
));
1965 * Returns a string with replication znodes and position of the replication log
1966 * @param zkw reference to the {@link ZKWatcher} which also contains configuration and operation
1967 * @return aq string of replication znodes and log positions
1969 public static String
getReplicationZnodesDump(ZKWatcher zkw
) throws KeeperException
{
1970 StringBuilder sb
= new StringBuilder();
1971 getReplicationZnodesDump(zkw
, sb
);
1972 return sb
.toString();
1975 private static void appendRSZnodes(ZKWatcher zkw
, String znode
, StringBuilder sb
)
1976 throws KeeperException
{
1977 List
<String
> stack
= new LinkedList
<>();
1980 String znodeToProcess
= stack
.remove(stack
.size() - 1);
1981 sb
.append("\n").append(znodeToProcess
).append(": ");
1984 data
= ZKUtil
.getData(zkw
, znodeToProcess
);
1985 } catch (InterruptedException e
) {
1986 zkw
.interruptedException(e
);
1989 if (data
!= null && data
.length
> 0) { // log position
1992 position
= ZKUtil
.parseWALPositionFrom(ZKUtil
.getData(zkw
, znodeToProcess
));
1993 sb
.append(position
);
1994 } catch (DeserializationException ignored
) {
1995 } catch (InterruptedException e
) {
1996 zkw
.interruptedException(e
);
2000 for (String zNodeChild
: ZKUtil
.listChildrenNoWatch(zkw
, znodeToProcess
)) {
2001 stack
.add(ZNodePaths
.joinZNode(znodeToProcess
, zNodeChild
));
2003 } while (stack
.size() > 0);
2006 private static void appendPeersZnodes(ZKWatcher zkw
, String peersZnode
,
2007 StringBuilder sb
) throws KeeperException
{
2008 int pblen
= ProtobufUtil
.lengthOfPBMagic();
2009 sb
.append("\n").append(peersZnode
).append(": ");
2010 for (String peerIdZnode
: ZKUtil
.listChildrenNoWatch(zkw
, peersZnode
)) {
2011 String znodeToProcess
= ZNodePaths
.joinZNode(peersZnode
, peerIdZnode
);
2014 data
= ZKUtil
.getData(zkw
, znodeToProcess
);
2015 } catch (InterruptedException e
) {
2016 zkw
.interruptedException(e
);
2019 // parse the data of the above peer znode.
2021 ReplicationProtos
.ReplicationPeer
.Builder builder
=
2022 ReplicationProtos
.ReplicationPeer
.newBuilder();
2023 ProtobufUtil
.mergeFrom(builder
, data
, pblen
, data
.length
- pblen
);
2024 String clusterKey
= builder
.getClusterkey();
2025 sb
.append("\n").append(znodeToProcess
).append(": ").append(clusterKey
);
2026 // add the peer-state.
2027 appendPeerState(zkw
, znodeToProcess
, sb
);
2028 } catch (IOException ipbe
) {
2029 LOG
.warn("Got Exception while parsing peer: " + znodeToProcess
, ipbe
);
2034 private static void appendPeerState(ZKWatcher zkw
, String znodeToProcess
, StringBuilder sb
)
2035 throws KeeperException
, InvalidProtocolBufferException
{
2036 String peerState
= zkw
.getConfiguration().get("zookeeper.znode.replication.peers.state",
2038 int pblen
= ProtobufUtil
.lengthOfPBMagic();
2039 for (String child
: ZKUtil
.listChildrenNoWatch(zkw
, znodeToProcess
)) {
2040 if (!child
.equals(peerState
)) {
2044 String peerStateZnode
= ZNodePaths
.joinZNode(znodeToProcess
, child
);
2045 sb
.append("\n").append(peerStateZnode
).append(": ");
2046 byte[] peerStateData
;
2048 peerStateData
= ZKUtil
.getData(zkw
, peerStateZnode
);
2049 ReplicationProtos
.ReplicationState
.Builder builder
=
2050 ReplicationProtos
.ReplicationState
.newBuilder();
2051 ProtobufUtil
.mergeFrom(builder
, peerStateData
, pblen
, peerStateData
.length
- pblen
);
2052 sb
.append(builder
.getState().name());
2053 } catch (IOException ipbe
) {
2054 LOG
.warn("Got Exception while parsing peer: " + znodeToProcess
, ipbe
);
2055 } catch (InterruptedException e
) {
2056 zkw
.interruptedException(e
);
2063 * Gets the statistics from the given server.
2065 * @param server The server to get the statistics from.
2066 * @param timeout The socket timeout to use.
2067 * @return The array of response strings.
2068 * @throws IOException When the socket communication fails.
2070 private static String
[] getServerStats(String server
, int timeout
)
2071 throws IOException
{
2072 String
[] sp
= server
.split(":");
2073 if (sp
.length
== 0) {
2077 String host
= sp
[0];
2078 int port
= sp
.length
> 1 ? Integer
.parseInt(sp
[1])
2079 : HConstants
.DEFAULT_ZOOKEEPER_CLIENT_PORT
;
2081 try (Socket socket
= new Socket()) {
2082 InetSocketAddress sockAddr
= new InetSocketAddress(host
, port
);
2083 if (sockAddr
.isUnresolved()) {
2084 throw new UnknownHostException(host
+ " cannot be resolved");
2086 socket
.connect(sockAddr
, timeout
);
2087 socket
.setSoTimeout(timeout
);
2088 try (PrintWriter out
= new PrintWriter(new BufferedWriter(
2089 new OutputStreamWriter(socket
.getOutputStream(), StandardCharsets
.UTF_8
)), true);
2090 BufferedReader in
= new BufferedReader(
2091 new InputStreamReader(socket
.getInputStream(), StandardCharsets
.UTF_8
))) {
2092 out
.println("stat");
2094 ArrayList
<String
> res
= new ArrayList
<>();
2096 String line
= in
.readLine();
2103 return res
.toArray(new String
[res
.size()]);
2108 private static void logRetrievedMsg(final ZKWatcher zkw
,
2109 final String znode
, final byte [] data
, final boolean watcherSet
) {
2110 if (!LOG
.isTraceEnabled()) {
2114 LOG
.trace(zkw
.prefix("Retrieved " + ((data
== null)?
0: data
.length
) +
2115 " byte(s) of data from znode " + znode
+
2116 (watcherSet?
" and set watcher; ": "; data=") +
2117 (data
== null?
"null": data
.length
== 0?
"empty": (
2118 zkw
.getZNodePaths().isMetaZNodePath(znode
)?
2119 getServerNameOrEmptyString(data
):
2120 znode
.startsWith(zkw
.getZNodePaths().backupMasterAddressesZNode
)?
2121 getServerNameOrEmptyString(data
):
2122 StringUtils
.abbreviate(Bytes
.toStringBinary(data
), 32)))));
2125 private static String
getServerNameOrEmptyString(final byte [] data
) {
2127 return ProtobufUtil
.parseServerNameFrom(data
).toString();
2128 } catch (DeserializationException e
) {
2134 * Waits for HBase installation's base (parent) znode to become available.
2135 * @throws IOException on ZK errors
2137 public static void waitForBaseZNode(Configuration conf
) throws IOException
{
2138 LOG
.info("Waiting until the base znode is available");
2139 String parentZNode
= conf
.get(HConstants
.ZOOKEEPER_ZNODE_PARENT
,
2140 HConstants
.DEFAULT_ZOOKEEPER_ZNODE_PARENT
);
2141 ZooKeeper zk
= new ZooKeeper(ZKConfig
.getZKQuorumServersString(conf
),
2142 conf
.getInt(HConstants
.ZK_SESSION_TIMEOUT
,
2143 HConstants
.DEFAULT_ZK_SESSION_TIMEOUT
), EmptyWatcher
.instance
);
2145 final int maxTimeMs
= 10000;
2146 final int maxNumAttempts
= maxTimeMs
/ HConstants
.SOCKET_RETRY_WAIT_MS
;
2148 KeeperException keeperEx
= null;
2151 for (int attempt
= 0; attempt
< maxNumAttempts
; ++attempt
) {
2153 if (zk
.exists(parentZNode
, false) != null) {
2154 LOG
.info("Parent znode exists: {}", parentZNode
);
2158 } catch (KeeperException e
) {
2161 Threads
.sleepWithoutInterrupt(HConstants
.SOCKET_RETRY_WAIT_MS
);
2166 } catch (InterruptedException ex
) {
2167 Thread
.currentThread().interrupt();
2170 if (keeperEx
!= null) {
2171 throw new IOException(keeperEx
);
2176 * Convert a {@link DeserializationException} to a more palatable {@link KeeperException}.
2177 * Used when can't let a {@link DeserializationException} out w/o changing public API.
2178 * @param e Exception to convert
2179 * @return Converted exception
2181 public static KeeperException
convert(final DeserializationException e
) {
2182 KeeperException ke
= new KeeperException
.DataInconsistencyException();
2188 * Recursively print the current state of ZK (non-transactional)
2189 * @param root name of the root directory in zk to print
2191 public static void logZKTree(ZKWatcher zkw
, String root
) {
2192 if (!LOG
.isDebugEnabled()) {
2196 LOG
.debug("Current zk system:");
2197 String prefix
= "|-";
2198 LOG
.debug(prefix
+ root
);
2200 logZKTree(zkw
, root
, prefix
);
2201 } catch (KeeperException e
) {
2202 throw new RuntimeException(e
);
2207 * Helper method to print the current state of the ZK tree.
2208 * @see #logZKTree(ZKWatcher, String)
2209 * @throws KeeperException if an unexpected exception occurs
2211 private static void logZKTree(ZKWatcher zkw
, String root
, String prefix
)
2212 throws KeeperException
{
2213 List
<String
> children
= ZKUtil
.listChildrenNoWatch(zkw
, root
);
2215 if (children
== null) {
2219 for (String child
: children
) {
2220 LOG
.debug(prefix
+ child
);
2221 String node
= ZNodePaths
.joinZNode(root
.equals("/") ?
"" : root
, child
);
2222 logZKTree(zkw
, node
, prefix
+ "---");
2227 * @param position the position to serialize
2228 * @return Serialized protobuf of <code>position</code> with pb magic prefix prepended suitable
2229 * for use as content of an wal position in a replication queue.
2231 public static byte[] positionToByteArray(final long position
) {
2232 byte[] bytes
= ReplicationProtos
.ReplicationHLogPosition
.newBuilder().setPosition(position
)
2233 .build().toByteArray();
2234 return ProtobufUtil
.prependPBMagic(bytes
);
2238 * @param bytes - Content of a WAL position znode.
2239 * @return long - The current WAL position.
2240 * @throws DeserializationException if the WAL position cannot be parsed
2242 public static long parseWALPositionFrom(final byte[] bytes
) throws DeserializationException
{
2243 if (bytes
== null) {
2244 throw new DeserializationException("Unable to parse null WAL position.");
2246 if (ProtobufUtil
.isPBMagicPrefix(bytes
)) {
2247 int pblen
= ProtobufUtil
.lengthOfPBMagic();
2248 ReplicationProtos
.ReplicationHLogPosition
.Builder builder
=
2249 ReplicationProtos
.ReplicationHLogPosition
.newBuilder();
2250 ReplicationProtos
.ReplicationHLogPosition position
;
2252 ProtobufUtil
.mergeFrom(builder
, bytes
, pblen
, bytes
.length
- pblen
);
2253 position
= builder
.build();
2254 } catch (IOException e
) {
2255 throw new DeserializationException(e
);
2257 return position
.getPosition();
2259 if (bytes
.length
> 0) {
2260 return Bytes
.toLong(bytes
);