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
.IOException
;
23 import java
.io
.InputStreamReader
;
24 import java
.io
.PrintWriter
;
25 import java
.net
.InetSocketAddress
;
26 import java
.net
.Socket
;
27 import java
.util
.ArrayList
;
28 import java
.util
.Arrays
;
29 import java
.util
.Deque
;
30 import java
.util
.HashMap
;
31 import java
.util
.LinkedList
;
32 import java
.util
.List
;
35 import javax
.security
.auth
.login
.AppConfigurationEntry
;
36 import javax
.security
.auth
.login
.AppConfigurationEntry
.LoginModuleControlFlag
;
38 import org
.apache
.commons
.lang
.StringUtils
;
39 import org
.apache
.commons
.logging
.Log
;
40 import org
.apache
.commons
.logging
.LogFactory
;
41 import org
.apache
.hadoop
.conf
.Configuration
;
42 import org
.apache
.hadoop
.hbase
.AuthUtil
;
43 import org
.apache
.hadoop
.hbase
.HBaseConfiguration
;
44 import org
.apache
.hadoop
.hbase
.HConstants
;
45 import org
.apache
.hadoop
.hbase
.classification
.InterfaceAudience
;
46 import org
.apache
.hadoop
.hbase
.exceptions
.DeserializationException
;
47 import org
.apache
.hadoop
.hbase
.replication
.ReplicationStateZKBase
;
48 import org
.apache
.hadoop
.hbase
.security
.Superusers
;
49 import org
.apache
.hadoop
.hbase
.shaded
.com
.google
.protobuf
.InvalidProtocolBufferException
;
50 import org
.apache
.hadoop
.hbase
.shaded
.com
.google
.protobuf
.UnsafeByteOperations
;
51 import org
.apache
.hadoop
.hbase
.shaded
.protobuf
.ProtobufUtil
;
52 import org
.apache
.hadoop
.hbase
.shaded
.protobuf
.generated
.ClusterStatusProtos
;
53 import org
.apache
.hadoop
.hbase
.shaded
.protobuf
.generated
.ClusterStatusProtos
.RegionStoreSequenceIds
;
54 import org
.apache
.hadoop
.hbase
.shaded
.protobuf
.generated
.ReplicationProtos
;
55 import org
.apache
.hadoop
.hbase
.util
.Bytes
;
56 import org
.apache
.hadoop
.hbase
.util
.Threads
;
57 import org
.apache
.hadoop
.hbase
.zookeeper
.ZKUtil
.ZKUtilOp
.CreateAndFailSilent
;
58 import org
.apache
.hadoop
.hbase
.zookeeper
.ZKUtil
.ZKUtilOp
.DeleteNodeFailSilent
;
59 import org
.apache
.hadoop
.hbase
.zookeeper
.ZKUtil
.ZKUtilOp
.SetData
;
60 import org
.apache
.hadoop
.security
.SecurityUtil
;
61 import org
.apache
.hadoop
.security
.UserGroupInformation
;
62 import org
.apache
.hadoop
.security
.authentication
.util
.KerberosUtil
;
63 import org
.apache
.zookeeper
.AsyncCallback
;
64 import org
.apache
.zookeeper
.CreateMode
;
65 import org
.apache
.zookeeper
.KeeperException
;
66 import org
.apache
.zookeeper
.KeeperException
.NoNodeException
;
67 import org
.apache
.zookeeper
.Op
;
68 import org
.apache
.zookeeper
.Watcher
;
69 import org
.apache
.zookeeper
.ZooDefs
.Ids
;
70 import org
.apache
.zookeeper
.ZooDefs
.Perms
;
71 import org
.apache
.zookeeper
.ZooKeeper
;
72 import org
.apache
.zookeeper
.client
.ZooKeeperSaslClient
;
73 import org
.apache
.zookeeper
.data
.ACL
;
74 import org
.apache
.zookeeper
.data
.Id
;
75 import org
.apache
.zookeeper
.data
.Stat
;
76 import org
.apache
.zookeeper
.proto
.CreateRequest
;
77 import org
.apache
.zookeeper
.proto
.DeleteRequest
;
78 import org
.apache
.zookeeper
.proto
.SetDataRequest
;
79 import org
.apache
.zookeeper
.server
.ZooKeeperSaslServer
;
82 * Internal HBase utility class for ZooKeeper.
84 * <p>Contains only static methods and constants.
86 * <p>Methods all throw {@link KeeperException} if there is an unexpected
87 * zookeeper exception, so callers of these methods must handle appropriately.
88 * If ZK is required for the operation, the server will need to be aborted.
90 @InterfaceAudience.Private
92 private static final Log LOG
= LogFactory
.getLog(ZKUtil
.class);
94 // TODO: Replace this with ZooKeeper constant when ZOOKEEPER-277 is resolved.
95 public static final char ZNODE_PATH_SEPARATOR
= '/';
96 private static int zkDumpConnectionTimeOut
;
99 * Creates a new connection to ZooKeeper, pulling settings and ensemble config
100 * from the specified configuration object using methods from {@link ZKConfig}.
102 * Sets the connection status monitoring watcher to the specified watcher.
104 * @param conf configuration to pull ensemble and other settings from
105 * @param watcher watcher to monitor connection changes
106 * @return connection to zookeeper
107 * @throws IOException if unable to connect to zk or config problem
109 public static RecoverableZooKeeper
connect(Configuration conf
, Watcher watcher
)
111 String ensemble
= ZKConfig
.getZKQuorumServersString(conf
);
112 return connect(conf
, ensemble
, watcher
);
115 public static RecoverableZooKeeper
connect(Configuration conf
, String ensemble
,
118 return connect(conf
, ensemble
, watcher
, null);
121 public static RecoverableZooKeeper
connect(Configuration conf
, String ensemble
,
122 Watcher watcher
, final String identifier
)
124 if(ensemble
== null) {
125 throw new IOException("Unable to determine ZooKeeper ensemble");
127 int timeout
= conf
.getInt(HConstants
.ZK_SESSION_TIMEOUT
,
128 HConstants
.DEFAULT_ZK_SESSION_TIMEOUT
);
129 if (LOG
.isTraceEnabled()) {
130 LOG
.trace(identifier
+ " opening connection to ZooKeeper ensemble=" + ensemble
);
132 int retry
= conf
.getInt("zookeeper.recovery.retry", 3);
133 int retryIntervalMillis
=
134 conf
.getInt("zookeeper.recovery.retry.intervalmill", 1000);
135 int maxSleepTime
= conf
.getInt("zookeeper.recovery.retry.maxsleeptime", 60000);
136 zkDumpConnectionTimeOut
= conf
.getInt("zookeeper.dump.connection.timeout",
138 return new RecoverableZooKeeper(ensemble
, timeout
, watcher
,
139 retry
, retryIntervalMillis
, maxSleepTime
, identifier
);
143 * Log in the current zookeeper server process using the given configuration
144 * keys for the credential file and login principal.
146 * <p><strong>This is only applicable when running on secure hbase</strong>
147 * On regular HBase (without security features), this will safely be ignored.
150 * @param conf The configuration data to use
151 * @param keytabFileKey Property key used to configure the path to the credential file
152 * @param userNameKey Property key used to configure the login principal
153 * @param hostname Current hostname to use in any credentials
154 * @throws IOException underlying exception from SecurityUtil.login() call
156 public static void loginServer(Configuration conf
, String keytabFileKey
,
157 String userNameKey
, String hostname
) throws IOException
{
158 login(conf
, keytabFileKey
, userNameKey
, hostname
,
159 ZooKeeperSaslServer
.LOGIN_CONTEXT_NAME_KEY
,
160 JaasConfiguration
.SERVER_KEYTAB_KERBEROS_CONFIG_NAME
);
164 * Log in the current zookeeper client using the given configuration
165 * keys for the credential file and login principal.
167 * <p><strong>This is only applicable when running on secure hbase</strong>
168 * On regular HBase (without security features), this will safely be ignored.
171 * @param conf The configuration data to use
172 * @param keytabFileKey Property key used to configure the path to the credential file
173 * @param userNameKey Property key used to configure the login principal
174 * @param hostname Current hostname to use in any credentials
175 * @throws IOException underlying exception from SecurityUtil.login() call
177 public static void loginClient(Configuration conf
, String keytabFileKey
,
178 String userNameKey
, String hostname
) throws IOException
{
179 login(conf
, keytabFileKey
, userNameKey
, hostname
,
180 ZooKeeperSaslClient
.LOGIN_CONTEXT_NAME_KEY
,
181 JaasConfiguration
.CLIENT_KEYTAB_KERBEROS_CONFIG_NAME
);
185 * Log in the current process using the given configuration keys for the
186 * credential file and login principal.
188 * <p><strong>This is only applicable when running on secure hbase</strong>
189 * On regular HBase (without security features), this will safely be ignored.
192 * @param conf The configuration data to use
193 * @param keytabFileKey Property key used to configure the path to the credential file
194 * @param userNameKey Property key used to configure the login principal
195 * @param hostname Current hostname to use in any credentials
196 * @param loginContextProperty property name to expose the entry name
197 * @param loginContextName jaas entry name
198 * @throws IOException underlying exception from SecurityUtil.login() call
200 private static void login(Configuration conf
, String keytabFileKey
,
201 String userNameKey
, String hostname
,
202 String loginContextProperty
, String loginContextName
)
204 if (!isSecureZooKeeper(conf
))
207 // User has specified a jaas.conf, keep this one as the good one.
208 // HBASE_OPTS="-Djava.security.auth.login.config=jaas.conf"
209 if (System
.getProperty("java.security.auth.login.config") != null)
212 // No keytab specified, no auth
213 String keytabFilename
= conf
.get(keytabFileKey
);
214 if (keytabFilename
== null) {
215 LOG
.warn("no keytab specified for: " + keytabFileKey
);
219 String principalConfig
= conf
.get(userNameKey
, System
.getProperty("user.name"));
220 String principalName
= SecurityUtil
.getServerPrincipal(principalConfig
, hostname
);
222 // Initialize the "jaas.conf" for keyTab/principal,
223 // If keyTab is not specified use the Ticket Cache.
224 // and set the zookeeper login context name.
225 JaasConfiguration jaasConf
= new JaasConfiguration(loginContextName
,
226 principalName
, keytabFilename
);
227 javax
.security
.auth
.login
.Configuration
.setConfiguration(jaasConf
);
228 System
.setProperty(loginContextProperty
, loginContextName
);
232 * A JAAS configuration that defines the login modules that we want to use for login.
234 private static class JaasConfiguration
extends javax
.security
.auth
.login
.Configuration
{
235 private static final String SERVER_KEYTAB_KERBEROS_CONFIG_NAME
=
236 "zookeeper-server-keytab-kerberos";
237 private static final String CLIENT_KEYTAB_KERBEROS_CONFIG_NAME
=
238 "zookeeper-client-keytab-kerberos";
240 private static final Map
<String
, String
> BASIC_JAAS_OPTIONS
= new HashMap
<>();
242 String jaasEnvVar
= System
.getenv("HBASE_JAAS_DEBUG");
243 if (jaasEnvVar
!= null && "true".equalsIgnoreCase(jaasEnvVar
)) {
244 BASIC_JAAS_OPTIONS
.put("debug", "true");
248 private static final Map
<String
,String
> KEYTAB_KERBEROS_OPTIONS
= new HashMap
<>();
250 KEYTAB_KERBEROS_OPTIONS
.put("doNotPrompt", "true");
251 KEYTAB_KERBEROS_OPTIONS
.put("storeKey", "true");
252 KEYTAB_KERBEROS_OPTIONS
.put("refreshKrb5Config", "true");
253 KEYTAB_KERBEROS_OPTIONS
.putAll(BASIC_JAAS_OPTIONS
);
256 private static final AppConfigurationEntry KEYTAB_KERBEROS_LOGIN
=
257 new AppConfigurationEntry(KerberosUtil
.getKrb5LoginModuleName(),
258 LoginModuleControlFlag
.REQUIRED
,
259 KEYTAB_KERBEROS_OPTIONS
);
261 private static final AppConfigurationEntry
[] KEYTAB_KERBEROS_CONF
=
262 new AppConfigurationEntry
[]{KEYTAB_KERBEROS_LOGIN
};
264 private javax
.security
.auth
.login
.Configuration baseConfig
;
265 private final String loginContextName
;
266 private final boolean useTicketCache
;
267 private final String keytabFile
;
268 private final String principal
;
270 public JaasConfiguration(String loginContextName
, String principal
, String keytabFile
) {
271 this(loginContextName
, principal
, keytabFile
, keytabFile
== null || keytabFile
.length() == 0);
274 private JaasConfiguration(String loginContextName
, String principal
,
275 String keytabFile
, boolean useTicketCache
) {
277 this.baseConfig
= javax
.security
.auth
.login
.Configuration
.getConfiguration();
278 } catch (SecurityException e
) {
279 this.baseConfig
= null;
281 this.loginContextName
= loginContextName
;
282 this.useTicketCache
= useTicketCache
;
283 this.keytabFile
= keytabFile
;
284 this.principal
= principal
;
285 LOG
.info("JaasConfiguration loginContextName=" + loginContextName
+
286 " principal=" + principal
+ " useTicketCache=" + useTicketCache
+
287 " keytabFile=" + keytabFile
);
291 public AppConfigurationEntry
[] getAppConfigurationEntry(String appName
) {
292 if (loginContextName
.equals(appName
)) {
293 if (!useTicketCache
) {
294 KEYTAB_KERBEROS_OPTIONS
.put("keyTab", keytabFile
);
295 KEYTAB_KERBEROS_OPTIONS
.put("useKeyTab", "true");
297 KEYTAB_KERBEROS_OPTIONS
.put("principal", principal
);
298 KEYTAB_KERBEROS_OPTIONS
.put("useTicketCache", useTicketCache ?
"true" : "false");
299 return KEYTAB_KERBEROS_CONF
;
301 if (baseConfig
!= null) return baseConfig
.getAppConfigurationEntry(appName
);
311 * Join the prefix znode name with the suffix znode name to generate a proper
314 * Assumes prefix does not end with slash and suffix does not begin with it.
316 * @param prefix beginning of znode name
317 * @param suffix ending of znode name
318 * @return result of properly joining prefix with suffix
320 public static String
joinZNode(String prefix
, String suffix
) {
321 return prefix
+ ZNODE_PATH_SEPARATOR
+ suffix
;
325 * Returns the full path of the immediate parent of the specified node.
326 * @param node path to get parent of
327 * @return parent of path, null if passed the root node or an invalid node
329 public static String
getParent(String node
) {
330 int idx
= node
.lastIndexOf(ZNODE_PATH_SEPARATOR
);
331 return idx
<= 0 ?
null : node
.substring(0, idx
);
335 * Get the name of the current node from the specified fully-qualified path.
336 * @param path fully-qualified path
337 * @return name of the current node
339 public static String
getNodeName(String path
) {
340 return path
.substring(path
.lastIndexOf("/")+1);
344 // Existence checks and watches
348 * Watch the specified znode for delete/create/change events. The watcher is
349 * set whether or not the node exists. If the node already exists, the method
350 * returns true. If the node does not exist, the method returns false.
352 * @param zkw zk reference
353 * @param znode path of node to watch
354 * @return true if znode exists, false if does not exist or error
355 * @throws KeeperException if unexpected zookeeper exception
357 public static boolean watchAndCheckExists(ZooKeeperWatcher zkw
, String znode
)
358 throws KeeperException
{
360 Stat s
= zkw
.getRecoverableZooKeeper().exists(znode
, zkw
);
361 boolean exists
= s
!= null ?
true : false;
363 LOG
.debug(zkw
.prefix("Set watcher on existing znode=" + znode
));
365 LOG
.debug(zkw
.prefix("Set watcher on znode that does not yet exist, " + znode
));
368 } catch (KeeperException e
) {
369 LOG
.warn(zkw
.prefix("Unable to set watcher on znode " + znode
), e
);
370 zkw
.keeperException(e
);
372 } catch (InterruptedException e
) {
373 LOG
.warn(zkw
.prefix("Unable to set watcher on znode " + znode
), e
);
374 zkw
.interruptedException(e
);
380 * Watch the specified znode, but only if exists. Useful when watching
381 * for deletions. Uses .getData() (and handles NoNodeException) instead
382 * of .exists() to accomplish this, as .getData() will only set a watch if
384 * @param zkw zk reference
385 * @param znode path of node to watch
386 * @return true if the watch is set, false if node does not exists
387 * @throws KeeperException if unexpected zookeeper exception
389 public static boolean setWatchIfNodeExists(ZooKeeperWatcher zkw
, String znode
)
390 throws KeeperException
{
392 zkw
.getRecoverableZooKeeper().getData(znode
, true, null);
394 } catch (NoNodeException e
) {
396 } catch (InterruptedException e
) {
397 LOG
.warn(zkw
.prefix("Unable to set watcher on znode " + znode
), e
);
398 zkw
.interruptedException(e
);
404 * Check if the specified node exists. Sets no watches.
406 * @param zkw zk reference
407 * @param znode path of node to watch
408 * @return version of the node if it exists, -1 if does not exist
409 * @throws KeeperException if unexpected zookeeper exception
411 public static int checkExists(ZooKeeperWatcher zkw
, String znode
)
412 throws KeeperException
{
414 Stat s
= zkw
.getRecoverableZooKeeper().exists(znode
, null);
415 return s
!= null ? s
.getVersion() : -1;
416 } catch (KeeperException e
) {
417 LOG
.warn(zkw
.prefix("Unable to set watcher on znode (" + znode
+ ")"), e
);
418 zkw
.keeperException(e
);
420 } catch (InterruptedException e
) {
421 LOG
.warn(zkw
.prefix("Unable to set watcher on znode (" + znode
+ ")"), e
);
422 zkw
.interruptedException(e
);
432 * Lists the children znodes of the specified znode. Also sets a watch on
433 * the specified znode which will capture a NodeDeleted event on the specified
434 * znode as well as NodeChildrenChanged if any children of the specified znode
435 * are created or deleted.
437 * Returns null if the specified node does not exist. Otherwise returns a
438 * list of children of the specified node. If the node exists but it has no
439 * children, an empty list will be returned.
441 * @param zkw zk reference
442 * @param znode path of node to list and watch children of
443 * @return list of children of the specified node, an empty list if the node
444 * exists but has no children, and null if the node does not exist
445 * @throws KeeperException if unexpected zookeeper exception
447 public static List
<String
> listChildrenAndWatchForNewChildren(
448 ZooKeeperWatcher zkw
, String znode
)
449 throws KeeperException
{
451 List
<String
> children
= zkw
.getRecoverableZooKeeper().getChildren(znode
, zkw
);
453 } catch(KeeperException
.NoNodeException ke
) {
454 LOG
.debug(zkw
.prefix("Unable to list children of znode " + znode
+ " " +
455 "because node does not exist (not an error)"));
457 } catch (KeeperException e
) {
458 LOG
.warn(zkw
.prefix("Unable to list children of znode " + znode
+ " "), e
);
459 zkw
.keeperException(e
);
461 } catch (InterruptedException e
) {
462 LOG
.warn(zkw
.prefix("Unable to list children of znode " + znode
+ " "), e
);
463 zkw
.interruptedException(e
);
469 * List all the children of the specified znode, setting a watch for children
470 * changes and also setting a watch on every individual child in order to get
471 * the NodeCreated and NodeDeleted events.
472 * @param zkw zookeeper reference
473 * @param znode node to get children of and watch
474 * @return list of znode names, null if the node doesn't exist
475 * @throws KeeperException
477 public static List
<String
> listChildrenAndWatchThem(ZooKeeperWatcher zkw
,
478 String znode
) throws KeeperException
{
479 List
<String
> children
= listChildrenAndWatchForNewChildren(zkw
, znode
);
480 if (children
== null) {
483 for (String child
: children
) {
484 watchAndCheckExists(zkw
, joinZNode(znode
, child
));
490 * Lists the children of the specified znode without setting any watches.
492 * Sets no watches at all, this method is best effort.
494 * Returns an empty list if the node has no children. Returns null if the
495 * parent node itself does not exist.
497 * @param zkw zookeeper reference
498 * @param znode node to get children
499 * @return list of data of children of specified znode, empty if no children,
500 * null if parent does not exist
501 * @throws KeeperException if unexpected zookeeper exception
503 public static List
<String
> listChildrenNoWatch(ZooKeeperWatcher zkw
, String znode
)
504 throws KeeperException
{
505 List
<String
> children
= null;
507 // List the children without watching
508 children
= zkw
.getRecoverableZooKeeper().getChildren(znode
, null);
509 } catch(KeeperException
.NoNodeException nne
) {
511 } catch(InterruptedException ie
) {
512 zkw
.interruptedException(ie
);
518 * Simple class to hold a node path and node data.
522 public static class NodeAndData
{
524 private byte [] data
;
525 public NodeAndData(String node
, byte [] data
) {
529 public String
getNode() {
532 public byte [] getData() {
536 public String
toString() {
539 public boolean isEmpty() {
540 return (data
== null || data
.length
== 0);
545 * Checks if the specified znode has any children. Sets no watches.
547 * Returns true if the node exists and has children. Returns false if the
548 * node does not exist or if the node does not have any children.
550 * Used during master initialization to determine if the master is a
551 * failed-over-to master or the first master during initial cluster startup.
552 * If the directory for regionserver ephemeral nodes is empty then this is
553 * a cluster startup, if not then it is not cluster startup.
555 * @param zkw zk reference
556 * @param znode path of node to check for children of
557 * @return true if node has children, false if not or node does not exist
558 * @throws KeeperException if unexpected zookeeper exception
560 public static boolean nodeHasChildren(ZooKeeperWatcher zkw
, String znode
)
561 throws KeeperException
{
563 return !zkw
.getRecoverableZooKeeper().getChildren(znode
, null).isEmpty();
564 } catch(KeeperException
.NoNodeException ke
) {
565 LOG
.debug(zkw
.prefix("Unable to list children of znode " + znode
+ " " +
566 "because node does not exist (not an error)"));
568 } catch (KeeperException e
) {
569 LOG
.warn(zkw
.prefix("Unable to list children of znode " + znode
), e
);
570 zkw
.keeperException(e
);
572 } catch (InterruptedException e
) {
573 LOG
.warn(zkw
.prefix("Unable to list children of znode " + znode
), e
);
574 zkw
.interruptedException(e
);
580 * Get the number of children of the specified node.
582 * If the node does not exist or has no children, returns 0.
584 * Sets no watches at all.
586 * @param zkw zk reference
587 * @param znode path of node to count children of
588 * @return number of children of specified node, 0 if none or parent does not
590 * @throws KeeperException if unexpected zookeeper exception
592 public static int getNumberOfChildren(ZooKeeperWatcher zkw
, String znode
)
593 throws KeeperException
{
595 Stat stat
= zkw
.getRecoverableZooKeeper().exists(znode
, null);
596 return stat
== null ?
0 : stat
.getNumChildren();
597 } catch(KeeperException e
) {
598 LOG
.warn(zkw
.prefix("Unable to get children of node " + znode
));
599 zkw
.keeperException(e
);
600 } catch(InterruptedException e
) {
601 zkw
.interruptedException(e
);
611 * Get znode data. Does not set a watcher.
612 * @return ZNode data, null if the node does not exist or if there is an
615 public static byte [] getData(ZooKeeperWatcher zkw
, String znode
)
616 throws KeeperException
, InterruptedException
{
618 byte [] data
= zkw
.getRecoverableZooKeeper().getData(znode
, null, null);
619 logRetrievedMsg(zkw
, znode
, data
, false);
621 } catch (KeeperException
.NoNodeException e
) {
622 LOG
.debug(zkw
.prefix("Unable to get data of znode " + znode
+ " " +
623 "because node does not exist (not an error)"));
625 } catch (KeeperException e
) {
626 LOG
.warn(zkw
.prefix("Unable to get data of znode " + znode
), e
);
627 zkw
.keeperException(e
);
633 * Get the data at the specified znode and set a watch.
635 * Returns the data and sets a watch if the node exists. Returns null and no
636 * watch is set if the node does not exist or there is an exception.
638 * @param zkw zk reference
639 * @param znode path of node
640 * @return data of the specified znode, or null
641 * @throws KeeperException if unexpected zookeeper exception
643 public static byte [] getDataAndWatch(ZooKeeperWatcher zkw
, String znode
)
644 throws KeeperException
{
645 return getDataInternal(zkw
, znode
, null, true);
649 * Get the data at the specified znode and set a watch.
651 * Returns the data and sets a watch if the node exists. Returns null and no
652 * watch is set if the node does not exist or there is an exception.
654 * @param zkw zk reference
655 * @param znode path of node
656 * @param stat object to populate the version of the znode
657 * @return data of the specified znode, or null
658 * @throws KeeperException if unexpected zookeeper exception
660 public static byte[] getDataAndWatch(ZooKeeperWatcher zkw
, String znode
,
661 Stat stat
) throws KeeperException
{
662 return getDataInternal(zkw
, znode
, stat
, true);
665 private static byte[] getDataInternal(ZooKeeperWatcher zkw
, String znode
, Stat stat
,
667 throws KeeperException
{
669 byte [] data
= zkw
.getRecoverableZooKeeper().getData(znode
, zkw
, stat
);
670 logRetrievedMsg(zkw
, znode
, data
, watcherSet
);
672 } catch (KeeperException
.NoNodeException e
) {
673 // This log can get pretty annoying when we cycle on 100ms waits.
674 // Enable trace if you really want to see it.
675 LOG
.trace(zkw
.prefix("Unable to get data of znode " + znode
+ " " +
676 "because node does not exist (not an error)"));
678 } catch (KeeperException e
) {
679 LOG
.warn(zkw
.prefix("Unable to get data of znode " + znode
), e
);
680 zkw
.keeperException(e
);
682 } catch (InterruptedException e
) {
683 LOG
.warn(zkw
.prefix("Unable to get data of znode " + znode
), e
);
684 zkw
.interruptedException(e
);
690 * Get the data at the specified znode without setting a watch.
692 * Returns the data if the node exists. Returns null if the node does not
695 * Sets the stats of the node in the passed Stat object. Pass a null stat if
698 * @param zkw zk reference
699 * @param znode path of node
700 * @param stat node status to get if node exists
701 * @return data of the specified znode, or null if node does not exist
702 * @throws KeeperException if unexpected zookeeper exception
704 public static byte [] getDataNoWatch(ZooKeeperWatcher zkw
, String znode
,
706 throws KeeperException
{
708 byte [] data
= zkw
.getRecoverableZooKeeper().getData(znode
, null, stat
);
709 logRetrievedMsg(zkw
, znode
, data
, false);
711 } catch (KeeperException
.NoNodeException e
) {
712 LOG
.debug(zkw
.prefix("Unable to get data of znode " + znode
+ " " +
713 "because node does not exist (not necessarily an error)"));
715 } catch (KeeperException e
) {
716 LOG
.warn(zkw
.prefix("Unable to get data of znode " + znode
), e
);
717 zkw
.keeperException(e
);
719 } catch (InterruptedException e
) {
720 LOG
.warn(zkw
.prefix("Unable to get data of znode " + znode
), e
);
721 zkw
.interruptedException(e
);
727 * Returns the date of child znodes of the specified znode. Also sets a watch on
728 * the specified znode which will capture a NodeDeleted event on the specified
729 * znode as well as NodeChildrenChanged if any children of the specified znode
730 * are created or deleted.
732 * Returns null if the specified node does not exist. Otherwise returns a
733 * list of children of the specified node. If the node exists but it has no
734 * children, an empty list will be returned.
736 * @param zkw zk reference
737 * @param baseNode path of node to list and watch children of
738 * @return list of data of children of the specified node, an empty list if the node
739 * exists but has no children, and null if the node does not exist
740 * @throws KeeperException if unexpected zookeeper exception
744 public static List
<NodeAndData
> getChildDataAndWatchForNewChildren(
745 ZooKeeperWatcher zkw
, String baseNode
) throws KeeperException
{
747 ZKUtil
.listChildrenAndWatchForNewChildren(zkw
, baseNode
);
749 List
<NodeAndData
> newNodes
= new ArrayList
<>();
750 for (String node
: nodes
) {
751 String nodePath
= ZKUtil
.joinZNode(baseNode
, node
);
752 byte[] data
= ZKUtil
.getDataAndWatch(zkw
, nodePath
);
753 newNodes
.add(new NodeAndData(nodePath
, data
));
761 * Update the data of an existing node with the expected version to have the
764 * Throws an exception if there is a version mismatch or some other problem.
766 * Sets no watches under any conditions.
768 * @param zkw zk reference
771 * @param expectedVersion
772 * @throws KeeperException if unexpected zookeeper exception
773 * @throws KeeperException.BadVersionException if version mismatch
777 public static void updateExistingNodeData(ZooKeeperWatcher zkw
, String znode
,
778 byte [] data
, int expectedVersion
)
779 throws KeeperException
{
781 zkw
.getRecoverableZooKeeper().setData(znode
, data
, expectedVersion
);
782 } catch(InterruptedException ie
) {
783 zkw
.interruptedException(ie
);
792 * Sets the data of the existing znode to be the specified data. Ensures that
793 * the current data has the specified expected version.
795 * <p>If the node does not exist, a {@link NoNodeException} will be thrown.
797 * <p>If their is a version mismatch, method returns null.
799 * <p>No watches are set but setting data will trigger other watchers of this
802 * <p>If there is another problem, a KeeperException will be thrown.
804 * @param zkw zk reference
805 * @param znode path of node
806 * @param data data to set for node
807 * @param expectedVersion version expected when setting data
808 * @return true if data set, false if version mismatch
809 * @throws KeeperException if unexpected zookeeper exception
811 public static boolean setData(ZooKeeperWatcher zkw
, String znode
,
812 byte [] data
, int expectedVersion
)
813 throws KeeperException
, KeeperException
.NoNodeException
{
815 return zkw
.getRecoverableZooKeeper().setData(znode
, data
, expectedVersion
) != null;
816 } catch (InterruptedException e
) {
817 zkw
.interruptedException(e
);
823 * Set data into node creating node if it doesn't yet exist.
824 * Does not set watch.
826 * @param zkw zk reference
827 * @param znode path of node
828 * @param data data to set for node
829 * @throws KeeperException
831 public static void createSetData(final ZooKeeperWatcher zkw
, final String znode
,
833 throws KeeperException
{
834 if (checkExists(zkw
, znode
) == -1) {
835 ZKUtil
.createWithParents(zkw
, znode
, data
);
837 ZKUtil
.setData(zkw
, znode
, data
);
842 * Sets the data of the existing znode to be the specified data. The node
843 * must exist but no checks are done on the existing data or version.
845 * <p>If the node does not exist, a {@link NoNodeException} will be thrown.
847 * <p>No watches are set but setting data will trigger other watchers of this
850 * <p>If there is another problem, a KeeperException will be thrown.
852 * @param zkw zk reference
853 * @param znode path of node
854 * @param data data to set for node
855 * @throws KeeperException if unexpected zookeeper exception
857 public static void setData(ZooKeeperWatcher zkw
, String znode
, byte [] data
)
858 throws KeeperException
, KeeperException
.NoNodeException
{
859 setData(zkw
, (SetData
)ZKUtilOp
.setData(znode
, data
));
862 private static void setData(ZooKeeperWatcher zkw
, SetData setData
)
863 throws KeeperException
, KeeperException
.NoNodeException
{
864 SetDataRequest sd
= (SetDataRequest
)toZooKeeperOp(zkw
, setData
).toRequestRecord();
865 setData(zkw
, sd
.getPath(), sd
.getData(), sd
.getVersion());
869 * Returns whether or not secure authentication is enabled
870 * (whether <code>hbase.security.authentication</code> is set to
871 * <code>kerberos</code>.
873 public static boolean isSecureZooKeeper(Configuration conf
) {
874 // Detection for embedded HBase client with jaas configuration
875 // defined for third party programs.
877 javax
.security
.auth
.login
.Configuration testConfig
=
878 javax
.security
.auth
.login
.Configuration
.getConfiguration();
879 if (testConfig
.getAppConfigurationEntry("Client") == null
880 && testConfig
.getAppConfigurationEntry(
881 JaasConfiguration
.CLIENT_KEYTAB_KERBEROS_CONFIG_NAME
) == null
882 && testConfig
.getAppConfigurationEntry(
883 JaasConfiguration
.SERVER_KEYTAB_KERBEROS_CONFIG_NAME
) == null
884 && conf
.get(HConstants
.ZK_CLIENT_KERBEROS_PRINCIPAL
) == null
885 && conf
.get(HConstants
.ZK_SERVER_KERBEROS_PRINCIPAL
) == null) {
889 } catch(Exception e
) {
890 // No Jaas configuration defined.
894 // Master & RSs uses hbase.zookeeper.client.*
895 return "kerberos".equalsIgnoreCase(conf
.get("hbase.security.authentication"));
898 private static ArrayList
<ACL
> createACL(ZooKeeperWatcher zkw
, String node
) {
899 return createACL(zkw
, node
, isSecureZooKeeper(zkw
.getConfiguration()));
902 public static ArrayList
<ACL
> createACL(ZooKeeperWatcher zkw
, String node
,
903 boolean isSecureZooKeeper
) {
904 if (!node
.startsWith(zkw
.znodePaths
.baseZNode
)) {
905 return Ids
.OPEN_ACL_UNSAFE
;
907 if (isSecureZooKeeper
) {
908 ArrayList
<ACL
> acls
= new ArrayList
<>();
909 // add permission to hbase supper user
910 String
[] superUsers
= zkw
.getConfiguration().getStrings(Superusers
.SUPERUSER_CONF_KEY
);
911 String hbaseUser
= null;
913 hbaseUser
= UserGroupInformation
.getCurrentUser().getShortUserName();
914 } catch (IOException e
) {
915 LOG
.debug("Could not acquire current User.", e
);
917 if (superUsers
!= null) {
918 List
<String
> groups
= new ArrayList
<>();
919 for (String user
: superUsers
) {
920 if (AuthUtil
.isGroupPrincipal(user
)) {
921 // TODO: Set node ACL for groups when ZK supports this feature
924 if(!user
.equals(hbaseUser
)) {
925 acls
.add(new ACL(Perms
.ALL
, new Id("sasl", user
)));
929 if (!groups
.isEmpty()) {
930 LOG
.warn("Znode ACL setting for group " + groups
931 + " is skipped, ZooKeeper doesn't support this feature presently.");
934 // Certain znodes are accessed directly by the client,
935 // so they must be readable by non-authenticated clients
936 if (zkw
.isClientReadable(node
)) {
937 acls
.addAll(Ids
.CREATOR_ALL_ACL
);
938 acls
.addAll(Ids
.READ_ACL_UNSAFE
);
940 acls
.addAll(Ids
.CREATOR_ALL_ACL
);
944 return Ids
.OPEN_ACL_UNSAFE
;
954 * Set the specified znode to be an ephemeral node carrying the specified
957 * If the node is created successfully, a watcher is also set on the node.
959 * If the node is not created successfully because it already exists, this
960 * method will also set a watcher on the node.
962 * If there is another problem, a KeeperException will be thrown.
964 * @param zkw zk reference
965 * @param znode path of node
966 * @param data data of node
967 * @return true if node created, false if not, watch set in both cases
968 * @throws KeeperException if unexpected zookeeper exception
970 public static boolean createEphemeralNodeAndWatch(ZooKeeperWatcher zkw
,
971 String znode
, byte [] data
)
972 throws KeeperException
{
975 zkw
.getRecoverableZooKeeper().create(znode
, data
, createACL(zkw
, znode
),
976 CreateMode
.EPHEMERAL
);
977 } catch (KeeperException
.NodeExistsException nee
) {
979 } catch (InterruptedException e
) {
980 LOG
.info("Interrupted", e
);
981 Thread
.currentThread().interrupt();
983 if(!watchAndCheckExists(zkw
, znode
)) {
984 // It did exist but now it doesn't, try again
985 return createEphemeralNodeAndWatch(zkw
, znode
, data
);
991 * Creates the specified znode to be a persistent node carrying the specified
994 * Returns true if the node was successfully created, false if the node
997 * If the node is created successfully, a watcher is also set on the node.
999 * If the node is not created successfully because it already exists, this
1000 * method will also set a watcher on the node but return false.
1002 * If there is another problem, a KeeperException will be thrown.
1004 * @param zkw zk reference
1005 * @param znode path of node
1006 * @param data data of node
1007 * @return true if node created, false if not, watch set in both cases
1008 * @throws KeeperException if unexpected zookeeper exception
1010 public static boolean createNodeIfNotExistsAndWatch(
1011 ZooKeeperWatcher zkw
, String znode
, byte [] data
)
1012 throws KeeperException
{
1015 zkw
.getRecoverableZooKeeper().create(znode
, data
, createACL(zkw
, znode
),
1016 CreateMode
.PERSISTENT
);
1017 } catch (KeeperException
.NodeExistsException nee
) {
1019 } catch (InterruptedException e
) {
1020 zkw
.interruptedException(e
);
1024 zkw
.getRecoverableZooKeeper().exists(znode
, zkw
);
1025 } catch (InterruptedException e
) {
1026 zkw
.interruptedException(e
);
1033 * Creates the specified znode with the specified data but does not watch it.
1035 * Returns the znode of the newly created node
1037 * If there is another problem, a KeeperException will be thrown.
1039 * @param zkw zk reference
1040 * @param znode path of node
1041 * @param data data of node
1042 * @param createMode specifying whether the node to be created is ephemeral and/or sequential
1043 * @return true name of the newly created znode or null
1044 * @throws KeeperException if unexpected zookeeper exception
1046 public static String
createNodeIfNotExistsNoWatch(ZooKeeperWatcher zkw
, String znode
,
1047 byte[] data
, CreateMode createMode
) throws KeeperException
{
1049 String createdZNode
= null;
1051 createdZNode
= zkw
.getRecoverableZooKeeper().create(znode
, data
,
1052 createACL(zkw
, znode
), createMode
);
1053 } catch (KeeperException
.NodeExistsException nee
) {
1055 } catch (InterruptedException e
) {
1056 zkw
.interruptedException(e
);
1059 return createdZNode
;
1063 * Creates the specified node with the specified data and watches it.
1065 * <p>Throws an exception if the node already exists.
1067 * <p>The node created is persistent and open access.
1069 * <p>Returns the version number of the created node if successful.
1071 * @param zkw zk reference
1072 * @param znode path of node to create
1073 * @param data data of node to create
1074 * @return version of node created
1075 * @throws KeeperException if unexpected zookeeper exception
1076 * @throws KeeperException.NodeExistsException if node already exists
1078 public static int createAndWatch(ZooKeeperWatcher zkw
,
1079 String znode
, byte [] data
)
1080 throws KeeperException
, KeeperException
.NodeExistsException
{
1082 zkw
.getRecoverableZooKeeper().create(znode
, data
, createACL(zkw
, znode
),
1083 CreateMode
.PERSISTENT
);
1084 Stat stat
= zkw
.getRecoverableZooKeeper().exists(znode
, zkw
);
1086 // Likely a race condition. Someone deleted the znode.
1087 throw KeeperException
.create(KeeperException
.Code
.SYSTEMERROR
,
1088 "ZK.exists returned null (i.e.: znode does not exist) for znode=" + znode
);
1090 return stat
.getVersion();
1091 } catch (InterruptedException e
) {
1092 zkw
.interruptedException(e
);
1098 * Async creates the specified node with the specified data.
1100 * <p>Throws an exception if the node already exists.
1102 * <p>The node created is persistent and open access.
1104 * @param zkw zk reference
1105 * @param znode path of node to create
1106 * @param data data of node to create
1110 public static void asyncCreate(ZooKeeperWatcher zkw
,
1111 String znode
, byte [] data
, final AsyncCallback
.StringCallback cb
,
1113 zkw
.getRecoverableZooKeeper().getZooKeeper().create(znode
, data
,
1114 createACL(zkw
, znode
), CreateMode
.PERSISTENT
, cb
, ctx
);
1118 * Creates the specified node, iff the node does not exist. Does not set a
1119 * watch and fails silently if the node already exists.
1121 * The node created is persistent and open access.
1123 * @param zkw zk reference
1124 * @param znode path of node
1125 * @throws KeeperException if unexpected zookeeper exception
1127 public static void createAndFailSilent(ZooKeeperWatcher zkw
,
1128 String znode
) throws KeeperException
{
1129 createAndFailSilent(zkw
, znode
, new byte[0]);
1133 * Creates the specified node containing specified data, iff the node does not exist. Does
1134 * not set a watch and fails silently if the node already exists.
1136 * The node created is persistent and open access.
1138 * @param zkw zk reference
1139 * @param znode path of node
1140 * @param data a byte array data to store in the znode
1141 * @throws KeeperException if unexpected zookeeper exception
1143 public static void createAndFailSilent(ZooKeeperWatcher zkw
,
1144 String znode
, byte[] data
)
1145 throws KeeperException
{
1146 createAndFailSilent(zkw
,
1147 (CreateAndFailSilent
)ZKUtilOp
.createAndFailSilent(znode
, data
));
1150 private static void createAndFailSilent(ZooKeeperWatcher zkw
, CreateAndFailSilent cafs
)
1151 throws KeeperException
{
1152 CreateRequest create
= (CreateRequest
)toZooKeeperOp(zkw
, cafs
).toRequestRecord();
1153 String znode
= create
.getPath();
1155 RecoverableZooKeeper zk
= zkw
.getRecoverableZooKeeper();
1156 if (zk
.exists(znode
, false) == null) {
1157 zk
.create(znode
, create
.getData(), create
.getAcl(), CreateMode
.fromFlag(create
.getFlags()));
1159 } catch(KeeperException
.NodeExistsException nee
) {
1160 } catch(KeeperException
.NoAuthException nee
){
1162 if (null == zkw
.getRecoverableZooKeeper().exists(znode
, false)) {
1163 // If we failed to create the file and it does not already exist.
1166 } catch (InterruptedException ie
) {
1167 zkw
.interruptedException(ie
);
1169 } catch(InterruptedException ie
) {
1170 zkw
.interruptedException(ie
);
1175 * Creates the specified node and all parent nodes required for it to exist.
1177 * No watches are set and no errors are thrown if the node already exists.
1179 * The nodes created are persistent and open access.
1181 * @param zkw zk reference
1182 * @param znode path of node
1183 * @throws KeeperException if unexpected zookeeper exception
1185 public static void createWithParents(ZooKeeperWatcher zkw
, String znode
)
1186 throws KeeperException
{
1187 createWithParents(zkw
, znode
, new byte[0]);
1191 * Creates the specified node and all parent nodes required for it to exist. The creation of
1192 * parent znodes is not atomic with the leafe znode creation but the data is written atomically
1193 * when the leaf node is created.
1195 * No watches are set and no errors are thrown if the node already exists.
1197 * The nodes created are persistent and open access.
1199 * @param zkw zk reference
1200 * @param znode path of node
1201 * @throws KeeperException if unexpected zookeeper exception
1203 public static void createWithParents(ZooKeeperWatcher zkw
, String znode
, byte[] data
)
1204 throws KeeperException
{
1209 zkw
.getRecoverableZooKeeper().create(znode
, data
, createACL(zkw
, znode
),
1210 CreateMode
.PERSISTENT
);
1211 } catch(KeeperException
.NodeExistsException nee
) {
1213 } catch(KeeperException
.NoNodeException nne
) {
1214 createWithParents(zkw
, getParent(znode
));
1215 createWithParents(zkw
, znode
, data
);
1216 } catch(InterruptedException ie
) {
1217 zkw
.interruptedException(ie
);
1226 * Delete the specified node. Sets no watches. Throws all exceptions.
1228 public static void deleteNode(ZooKeeperWatcher zkw
, String node
)
1229 throws KeeperException
{
1230 deleteNode(zkw
, node
, -1);
1234 * Delete the specified node with the specified version. Sets no watches.
1235 * Throws all exceptions.
1237 public static boolean deleteNode(ZooKeeperWatcher zkw
, String node
,
1239 throws KeeperException
{
1241 zkw
.getRecoverableZooKeeper().delete(node
, version
);
1243 } catch(KeeperException
.BadVersionException bve
) {
1245 } catch(InterruptedException ie
) {
1246 zkw
.interruptedException(ie
);
1252 * Deletes the specified node. Fails silent if the node does not exist.
1255 * @throws KeeperException
1257 public static void deleteNodeFailSilent(ZooKeeperWatcher zkw
, String node
)
1258 throws KeeperException
{
1259 deleteNodeFailSilent(zkw
,
1260 (DeleteNodeFailSilent
)ZKUtilOp
.deleteNodeFailSilent(node
));
1263 private static void deleteNodeFailSilent(ZooKeeperWatcher zkw
,
1264 DeleteNodeFailSilent dnfs
) throws KeeperException
{
1265 DeleteRequest delete
= (DeleteRequest
)toZooKeeperOp(zkw
, dnfs
).toRequestRecord();
1267 zkw
.getRecoverableZooKeeper().delete(delete
.getPath(), delete
.getVersion());
1268 } catch(KeeperException
.NoNodeException nne
) {
1269 } catch(InterruptedException ie
) {
1270 zkw
.interruptedException(ie
);
1276 * Delete the specified node and all of it's children.
1278 * If the node does not exist, just returns.
1280 * Sets no watches. Throws all exceptions besides dealing with deletion of
1283 public static void deleteNodeRecursively(ZooKeeperWatcher zkw
, String node
)
1284 throws KeeperException
{
1285 deleteNodeRecursivelyMultiOrSequential(zkw
, true, node
);
1289 * Delete all the children of the specified node but not the node itself.
1291 * Sets no watches. Throws all exceptions besides dealing with deletion of
1294 * @throws KeeperException
1296 public static void deleteChildrenRecursively(ZooKeeperWatcher zkw
, String node
)
1297 throws KeeperException
{
1298 deleteChildrenRecursivelyMultiOrSequential(zkw
, true, node
);
1302 * Delete all the children of the specified node but not the node itself. This
1303 * will first traverse the znode tree for listing the children and then delete
1304 * these znodes using multi-update api or sequential based on the specified
1307 * Sets no watches. Throws all exceptions besides dealing with deletion of
1310 * If the following is true:
1312 * <li>runSequentialOnMultiFailure is true
1314 * on calling multi, we get a ZooKeeper exception that can be handled by a
1315 * sequential call(*), we retry the operations one-by-one (sequentially).
1319 * @param runSequentialOnMultiFailure
1320 * - if true when we get a ZooKeeper exception that could retry the
1321 * operations one-by-one (sequentially)
1323 * - path of the parent node(s)
1324 * @throws KeeperException.NotEmptyException
1325 * if node has children while deleting
1326 * @throws KeeperException
1327 * if unexpected ZooKeeper exception
1328 * @throws IllegalArgumentException
1329 * if an invalid path is specified
1331 public static void deleteChildrenRecursivelyMultiOrSequential(
1332 ZooKeeperWatcher zkw
, boolean runSequentialOnMultiFailure
,
1333 String
... pathRoots
) throws KeeperException
{
1334 if (pathRoots
== null || pathRoots
.length
<= 0) {
1335 LOG
.warn("Given path is not valid!");
1338 List
<ZKUtilOp
> ops
= new ArrayList
<>();
1339 for (String eachRoot
: pathRoots
) {
1340 List
<String
> children
= listChildrenBFSNoWatch(zkw
, eachRoot
);
1341 // Delete the leaves first and eventually get rid of the root
1342 for (int i
= children
.size() - 1; i
>= 0; --i
) {
1343 ops
.add(ZKUtilOp
.deleteNodeFailSilent(children
.get(i
)));
1346 // atleast one element should exist
1347 if (ops
.size() > 0) {
1348 multiOrSequential(zkw
, ops
, runSequentialOnMultiFailure
);
1353 * Delete the specified node and its children. This traverse the
1354 * znode tree for listing the children and then delete
1355 * these znodes including the parent using multi-update api or
1356 * sequential based on the specified configurations.
1358 * Sets no watches. Throws all exceptions besides dealing with deletion of
1361 * If the following is true:
1363 * <li>runSequentialOnMultiFailure is true
1365 * on calling multi, we get a ZooKeeper exception that can be handled by a
1366 * sequential call(*), we retry the operations one-by-one (sequentially).
1370 * @param runSequentialOnMultiFailure
1371 * - if true when we get a ZooKeeper exception that could retry the
1372 * operations one-by-one (sequentially)
1374 * - path of the parent node(s)
1375 * @throws KeeperException.NotEmptyException
1376 * if node has children while deleting
1377 * @throws KeeperException
1378 * if unexpected ZooKeeper exception
1379 * @throws IllegalArgumentException
1380 * if an invalid path is specified
1382 public static void deleteNodeRecursivelyMultiOrSequential(ZooKeeperWatcher zkw
,
1383 boolean runSequentialOnMultiFailure
, String
... pathRoots
) throws KeeperException
{
1384 if (pathRoots
== null || pathRoots
.length
<= 0) {
1385 LOG
.warn("Given path is not valid!");
1388 List
<ZKUtilOp
> ops
= new ArrayList
<>();
1389 for (String eachRoot
: pathRoots
) {
1390 // ZooKeeper Watches are one time triggers; When children of parent nodes are deleted
1391 // recursively, must set another watch, get notified of delete node
1392 List
<String
> children
= listChildrenBFSAndWatchThem(zkw
, eachRoot
);
1393 // Delete the leaves first and eventually get rid of the root
1394 for (int i
= children
.size() - 1; i
>= 0; --i
) {
1395 ops
.add(ZKUtilOp
.deleteNodeFailSilent(children
.get(i
)));
1398 if (zkw
.getRecoverableZooKeeper().exists(eachRoot
, zkw
) != null) {
1399 ops
.add(ZKUtilOp
.deleteNodeFailSilent(eachRoot
));
1401 } catch (InterruptedException e
) {
1402 zkw
.interruptedException(e
);
1405 // atleast one element should exist
1406 if (ops
.size() > 0) {
1407 multiOrSequential(zkw
, ops
, runSequentialOnMultiFailure
);
1412 * BFS Traversal of all the children under path, with the entries in the list,
1413 * in the same order as that of the traversal. Lists all the children without
1414 * setting any watches.
1420 * @return list of children znodes under the path
1421 * @throws KeeperException
1422 * if unexpected ZooKeeper exception
1424 private static List
<String
> listChildrenBFSNoWatch(ZooKeeperWatcher zkw
,
1425 final String znode
) throws KeeperException
{
1426 Deque
<String
> queue
= new LinkedList
<>();
1427 List
<String
> tree
= new ArrayList
<>();
1430 String node
= queue
.pollFirst();
1434 List
<String
> children
= listChildrenNoWatch(zkw
, node
);
1435 if (children
== null) {
1438 for (final String child
: children
) {
1439 final String childPath
= node
+ "/" + child
;
1440 queue
.add(childPath
);
1441 tree
.add(childPath
);
1448 * BFS Traversal of all the children under path, with the entries in the list,
1449 * in the same order as that of the traversal.
1450 * Lists all the children and set watches on to them.
1456 * @return list of children znodes under the path
1457 * @throws KeeperException
1458 * if unexpected ZooKeeper exception
1460 private static List
<String
> listChildrenBFSAndWatchThem(ZooKeeperWatcher zkw
, final String znode
)
1461 throws KeeperException
{
1462 Deque
<String
> queue
= new LinkedList
<>();
1463 List
<String
> tree
= new ArrayList
<>();
1466 String node
= queue
.pollFirst();
1470 List
<String
> children
= listChildrenAndWatchThem(zkw
, node
);
1471 if (children
== null) {
1474 for (final String child
: children
) {
1475 final String childPath
= node
+ "/" + child
;
1476 queue
.add(childPath
);
1477 tree
.add(childPath
);
1484 * Represents an action taken by ZKUtil, e.g. createAndFailSilent.
1485 * These actions are higher-level than ZKOp actions, which represent
1486 * individual actions in the ZooKeeper API, like create.
1488 public abstract static class ZKUtilOp
{
1489 private String path
;
1491 private ZKUtilOp(String path
) {
1496 * @return a createAndFailSilent ZKUtilOp
1498 public static ZKUtilOp
createAndFailSilent(String path
, byte[] data
) {
1499 return new CreateAndFailSilent(path
, data
);
1503 * @return a deleteNodeFailSilent ZKUtilOP
1505 public static ZKUtilOp
deleteNodeFailSilent(String path
) {
1506 return new DeleteNodeFailSilent(path
);
1510 * @return a setData ZKUtilOp
1512 public static ZKUtilOp
setData(String path
, byte [] data
) {
1513 return new SetData(path
, data
);
1517 * @return path to znode where the ZKOp will occur
1519 public String
getPath() {
1524 * ZKUtilOp representing createAndFailSilent in ZooKeeper
1525 * (attempt to create node, ignore error if already exists)
1527 public static class CreateAndFailSilent
extends ZKUtilOp
{
1528 private byte [] data
;
1530 private CreateAndFailSilent(String path
, byte [] data
) {
1535 public byte[] getData() {
1540 public boolean equals(Object o
) {
1541 if (this == o
) return true;
1542 if (!(o
instanceof CreateAndFailSilent
)) return false;
1544 CreateAndFailSilent op
= (CreateAndFailSilent
) o
;
1545 return getPath().equals(op
.getPath()) && Arrays
.equals(data
, op
.data
);
1549 public int hashCode() {
1550 int ret
= 17 + getPath().hashCode() * 31;
1551 return ret
* 31 + Bytes
.hashCode(data
);
1556 * ZKUtilOp representing deleteNodeFailSilent in ZooKeeper
1557 * (attempt to delete node, ignore error if node doesn't exist)
1559 public static class DeleteNodeFailSilent
extends ZKUtilOp
{
1560 private DeleteNodeFailSilent(String path
) {
1565 public boolean equals(Object o
) {
1566 if (this == o
) return true;
1567 if (!(o
instanceof DeleteNodeFailSilent
)) return false;
1569 return super.equals(o
);
1573 public int hashCode() {
1574 return getPath().hashCode();
1579 * ZKUtilOp representing setData in ZooKeeper
1581 public static class SetData
extends ZKUtilOp
{
1582 private byte [] data
;
1584 private SetData(String path
, byte [] data
) {
1589 public byte[] getData() {
1594 public boolean equals(Object o
) {
1595 if (this == o
) return true;
1596 if (!(o
instanceof SetData
)) return false;
1598 SetData op
= (SetData
) o
;
1599 return getPath().equals(op
.getPath()) && Arrays
.equals(data
, op
.data
);
1603 public int hashCode() {
1604 int ret
= getPath().hashCode();
1605 return ret
* 31 + Bytes
.hashCode(data
);
1611 * Convert from ZKUtilOp to ZKOp
1613 private static Op
toZooKeeperOp(ZooKeeperWatcher zkw
, ZKUtilOp op
)
1614 throws UnsupportedOperationException
{
1615 if(op
== null) return null;
1617 if (op
instanceof CreateAndFailSilent
) {
1618 CreateAndFailSilent cafs
= (CreateAndFailSilent
)op
;
1619 return Op
.create(cafs
.getPath(), cafs
.getData(), createACL(zkw
, cafs
.getPath()),
1620 CreateMode
.PERSISTENT
);
1621 } else if (op
instanceof DeleteNodeFailSilent
) {
1622 DeleteNodeFailSilent dnfs
= (DeleteNodeFailSilent
)op
;
1623 return Op
.delete(dnfs
.getPath(), -1);
1624 } else if (op
instanceof SetData
) {
1625 SetData sd
= (SetData
)op
;
1626 return Op
.setData(sd
.getPath(), sd
.getData(), -1);
1628 throw new UnsupportedOperationException("Unexpected ZKUtilOp type: "
1629 + op
.getClass().getName());
1634 * Use ZooKeeper's multi-update functionality.
1636 * If all of the following are true:
1637 * - runSequentialOnMultiFailure is true
1638 * - on calling multi, we get a ZooKeeper exception that can be handled by a sequential call(*)
1640 * - we retry the operations one-by-one (sequentially)
1642 * Note *: an example is receiving a NodeExistsException from a "create" call. Without multi,
1643 * a user could call "createAndFailSilent" to ensure that a node exists if they don't care who
1644 * actually created the node (i.e. the NodeExistsException from ZooKeeper is caught).
1645 * This will cause all operations in the multi to fail, however, because
1646 * the NodeExistsException that zk.create throws will fail the multi transaction.
1647 * In this case, if the previous conditions hold, the commands are run sequentially, which should
1648 * result in the correct final state, but means that the operations will not run atomically.
1650 * @throws KeeperException
1652 public static void multiOrSequential(ZooKeeperWatcher zkw
, List
<ZKUtilOp
> ops
,
1653 boolean runSequentialOnMultiFailure
) throws KeeperException
{
1654 if (zkw
.getConfiguration().get("hbase.zookeeper.useMulti") != null) {
1655 LOG
.warn("hbase.zookeeper.useMulti is deprecated. Default to true always.");
1657 if (ops
== null) return;
1659 List
<Op
> zkOps
= new LinkedList
<>();
1660 for (ZKUtilOp op
: ops
) {
1661 zkOps
.add(toZooKeeperOp(zkw
, op
));
1664 zkw
.getRecoverableZooKeeper().multi(zkOps
);
1665 } catch (KeeperException ke
) {
1666 switch (ke
.code()) {
1671 // if we get an exception that could be solved by running sequentially
1672 // (and the client asked us to), then break out and run sequentially
1673 if (runSequentialOnMultiFailure
) {
1674 LOG
.info("On call to ZK.multi, received exception: " + ke
.toString() + "."
1675 + " Attempting to run operations sequentially because"
1676 + " runSequentialOnMultiFailure is: " + runSequentialOnMultiFailure
+ ".");
1677 processSequentially(zkw
, ops
);
1683 } catch (InterruptedException ie
) {
1684 zkw
.interruptedException(ie
);
1688 private static void processSequentially(ZooKeeperWatcher zkw
, List
<ZKUtilOp
> ops
)
1689 throws KeeperException
, NoNodeException
{
1690 for (ZKUtilOp op
: ops
) {
1691 if (op
instanceof CreateAndFailSilent
) {
1692 createAndFailSilent(zkw
, (CreateAndFailSilent
) op
);
1693 } else if (op
instanceof DeleteNodeFailSilent
) {
1694 deleteNodeFailSilent(zkw
, (DeleteNodeFailSilent
) op
);
1695 } else if (op
instanceof SetData
) {
1696 setData(zkw
, (SetData
) op
);
1698 throw new UnsupportedOperationException("Unexpected ZKUtilOp type: "
1699 + op
.getClass().getName());
1705 // ZooKeeper cluster information
1708 /** @return String dump of everything in ZooKeeper. */
1709 public static String
dump(ZooKeeperWatcher zkw
) {
1710 StringBuilder sb
= new StringBuilder();
1712 sb
.append("HBase is rooted at ").append(zkw
.znodePaths
.baseZNode
);
1713 sb
.append("\nActive master address: ");
1715 sb
.append(MasterAddressTracker
.getMasterAddress(zkw
));
1716 } catch (IOException e
) {
1717 sb
.append("<<FAILED LOOKUP: " + e
.getMessage() + ">>");
1719 sb
.append("\nBackup master addresses:");
1720 for (String child
: listChildrenNoWatch(zkw
, zkw
.znodePaths
.backupMasterAddressesZNode
)) {
1721 sb
.append("\n ").append(child
);
1723 sb
.append("\nRegion server holding hbase:meta: "
1724 + new MetaTableLocator().getMetaRegionLocation(zkw
));
1725 Configuration conf
= HBaseConfiguration
.create();
1726 int numMetaReplicas
= conf
.getInt(HConstants
.META_REPLICAS_NUM
,
1727 HConstants
.DEFAULT_META_REPLICA_NUM
);
1728 for (int i
= 1; i
< numMetaReplicas
; i
++) {
1729 sb
.append("\nRegion server holding hbase:meta, replicaId " + i
+ " "
1730 + new MetaTableLocator().getMetaRegionLocation(zkw
, i
));
1732 sb
.append("\nRegion servers:");
1733 for (String child
: listChildrenNoWatch(zkw
, zkw
.znodePaths
.rsZNode
)) {
1734 sb
.append("\n ").append(child
);
1737 getReplicationZnodesDump(zkw
, sb
);
1738 } catch (KeeperException ke
) {
1739 LOG
.warn("Couldn't get the replication znode dump", ke
);
1741 sb
.append("\nQuorum Server Statistics:");
1742 String
[] servers
= zkw
.getQuorum().split(",");
1743 for (String server
: servers
) {
1744 sb
.append("\n ").append(server
);
1746 String
[] stat
= getServerStats(server
, ZKUtil
.zkDumpConnectionTimeOut
);
1749 sb
.append("[Error] invalid quorum server: " + server
);
1753 for (String s
: stat
) {
1754 sb
.append("\n ").append(s
);
1756 } catch (Exception e
) {
1757 sb
.append("\n ERROR: ").append(e
.getMessage());
1760 } catch (KeeperException ke
) {
1761 sb
.append("\nFATAL ZooKeeper Exception!\n");
1762 sb
.append("\n" + ke
.getMessage());
1764 return sb
.toString();
1768 * Appends replication znodes to the passed StringBuilder.
1771 * @throws KeeperException
1773 private static void getReplicationZnodesDump(ZooKeeperWatcher zkw
, StringBuilder sb
)
1774 throws KeeperException
{
1775 String replicationZNodeName
= zkw
.getConfiguration().get("zookeeper.znode.replication",
1777 String replicationZnode
= joinZNode(zkw
.znodePaths
.baseZNode
, replicationZNodeName
);
1778 if (ZKUtil
.checkExists(zkw
, replicationZnode
) == -1) return;
1779 // do a ls -r on this znode
1780 sb
.append("\n").append(replicationZnode
).append(": ");
1781 List
<String
> children
= ZKUtil
.listChildrenNoWatch(zkw
, replicationZnode
);
1782 for (String child
: children
) {
1783 String znode
= joinZNode(replicationZnode
, child
);
1784 if (child
.equals(zkw
.getConfiguration().get("zookeeper.znode.replication.peers", "peers"))) {
1785 appendPeersZnodes(zkw
, znode
, sb
);
1786 } else if (child
.equals(zkw
.getConfiguration().
1787 get("zookeeper.znode.replication.rs", "rs"))) {
1788 appendRSZnodes(zkw
, znode
, sb
);
1789 } else if (child
.equals(zkw
.getConfiguration().get(
1790 ReplicationStateZKBase
.ZOOKEEPER_ZNODE_REPLICATION_HFILE_REFS_KEY
,
1791 ReplicationStateZKBase
.ZOOKEEPER_ZNODE_REPLICATION_HFILE_REFS_DEFAULT
))) {
1792 appendHFileRefsZnodes(zkw
, znode
, sb
);
1797 private static void appendHFileRefsZnodes(ZooKeeperWatcher zkw
, String hfileRefsZnode
,
1798 StringBuilder sb
) throws KeeperException
{
1799 sb
.append("\n").append(hfileRefsZnode
).append(": ");
1800 for (String peerIdZnode
: ZKUtil
.listChildrenNoWatch(zkw
, hfileRefsZnode
)) {
1801 String znodeToProcess
= ZKUtil
.joinZNode(hfileRefsZnode
, peerIdZnode
);
1802 sb
.append("\n").append(znodeToProcess
).append(": ");
1803 List
<String
> peerHFileRefsZnodes
= ZKUtil
.listChildrenNoWatch(zkw
, znodeToProcess
);
1804 int size
= peerHFileRefsZnodes
.size();
1805 for (int i
= 0; i
< size
; i
++) {
1806 sb
.append(peerHFileRefsZnodes
.get(i
));
1807 if (i
!= size
- 1) {
1815 * Returns a string with replication znodes and position of the replication log
1817 * @return aq string of replication znodes and log positions
1819 public static String
getReplicationZnodesDump(ZooKeeperWatcher zkw
) throws KeeperException
{
1820 StringBuilder sb
= new StringBuilder();
1821 getReplicationZnodesDump(zkw
, sb
);
1822 return sb
.toString();
1825 private static void appendRSZnodes(ZooKeeperWatcher zkw
, String znode
, StringBuilder sb
)
1826 throws KeeperException
{
1827 List
<String
> stack
= new LinkedList
<>();
1830 String znodeToProcess
= stack
.remove(stack
.size() - 1);
1831 sb
.append("\n").append(znodeToProcess
).append(": ");
1834 data
= ZKUtil
.getData(zkw
, znodeToProcess
);
1835 } catch (InterruptedException e
) {
1836 zkw
.interruptedException(e
);
1839 if (data
!= null && data
.length
> 0) { // log position
1842 position
= ZKUtil
.parseWALPositionFrom(ZKUtil
.getData(zkw
, znodeToProcess
));
1843 sb
.append(position
);
1844 } catch (DeserializationException ignored
) {
1845 } catch (InterruptedException e
) {
1846 zkw
.interruptedException(e
);
1850 for (String zNodeChild
: ZKUtil
.listChildrenNoWatch(zkw
, znodeToProcess
)) {
1851 stack
.add(ZKUtil
.joinZNode(znodeToProcess
, zNodeChild
));
1853 } while (stack
.size() > 0);
1856 private static void appendPeersZnodes(ZooKeeperWatcher zkw
, String peersZnode
,
1857 StringBuilder sb
) throws KeeperException
{
1858 int pblen
= ProtobufUtil
.lengthOfPBMagic();
1859 sb
.append("\n").append(peersZnode
).append(": ");
1860 for (String peerIdZnode
: ZKUtil
.listChildrenNoWatch(zkw
, peersZnode
)) {
1861 String znodeToProcess
= ZKUtil
.joinZNode(peersZnode
, peerIdZnode
);
1864 data
= ZKUtil
.getData(zkw
, znodeToProcess
);
1865 } catch (InterruptedException e
) {
1866 zkw
.interruptedException(e
);
1869 // parse the data of the above peer znode.
1871 ReplicationProtos
.ReplicationPeer
.Builder builder
=
1872 ReplicationProtos
.ReplicationPeer
.newBuilder();
1873 ProtobufUtil
.mergeFrom(builder
, data
, pblen
, data
.length
- pblen
);
1874 String clusterKey
= builder
.getClusterkey();
1875 sb
.append("\n").append(znodeToProcess
).append(": ").append(clusterKey
);
1876 // add the peer-state.
1877 appendPeerState(zkw
, znodeToProcess
, sb
);
1878 } catch (IOException ipbe
) {
1879 LOG
.warn("Got Exception while parsing peer: " + znodeToProcess
, ipbe
);
1884 private static void appendPeerState(ZooKeeperWatcher zkw
, String znodeToProcess
,
1885 StringBuilder sb
) throws KeeperException
, InvalidProtocolBufferException
{
1886 String peerState
= zkw
.getConfiguration().get("zookeeper.znode.replication.peers.state",
1888 int pblen
= ProtobufUtil
.lengthOfPBMagic();
1889 for (String child
: ZKUtil
.listChildrenNoWatch(zkw
, znodeToProcess
)) {
1890 if (!child
.equals(peerState
)) continue;
1891 String peerStateZnode
= ZKUtil
.joinZNode(znodeToProcess
, child
);
1892 sb
.append("\n").append(peerStateZnode
).append(": ");
1893 byte[] peerStateData
;
1895 peerStateData
= ZKUtil
.getData(zkw
, peerStateZnode
);
1896 ReplicationProtos
.ReplicationState
.Builder builder
=
1897 ReplicationProtos
.ReplicationState
.newBuilder();
1898 ProtobufUtil
.mergeFrom(builder
, peerStateData
, pblen
, peerStateData
.length
- pblen
);
1899 sb
.append(builder
.getState().name());
1900 } catch (IOException ipbe
) {
1901 LOG
.warn("Got Exception while parsing peer: " + znodeToProcess
, ipbe
);
1902 } catch (InterruptedException e
) {
1903 zkw
.interruptedException(e
);
1910 * Gets the statistics from the given server.
1912 * @param server The server to get the statistics from.
1913 * @param timeout The socket timeout to use.
1914 * @return The array of response strings.
1915 * @throws IOException When the socket communication fails.
1917 public static String
[] getServerStats(String server
, int timeout
)
1918 throws IOException
{
1919 String
[] sp
= server
.split(":");
1920 if (sp
== null || sp
.length
== 0) {
1924 String host
= sp
[0];
1925 int port
= sp
.length
> 1 ? Integer
.parseInt(sp
[1])
1926 : HConstants
.DEFAULT_ZOOKEPER_CLIENT_PORT
;
1928 InetSocketAddress sockAddr
= new InetSocketAddress(host
, port
);
1929 try (Socket socket
= new Socket()) {
1930 socket
.connect(sockAddr
, timeout
);
1932 socket
.setSoTimeout(timeout
);
1933 try (PrintWriter out
= new PrintWriter(socket
.getOutputStream(), true);
1934 BufferedReader in
= new BufferedReader(new InputStreamReader(socket
.getInputStream()))) {
1935 out
.println("stat");
1937 ArrayList
<String
> res
= new ArrayList
<>();
1939 String line
= in
.readLine();
1946 return res
.toArray(new String
[res
.size()]);
1951 private static void logRetrievedMsg(final ZooKeeperWatcher zkw
,
1952 final String znode
, final byte [] data
, final boolean watcherSet
) {
1953 if (!LOG
.isTraceEnabled()) return;
1954 LOG
.trace(zkw
.prefix("Retrieved " + ((data
== null)?
0: data
.length
) +
1955 " byte(s) of data from znode " + znode
+
1956 (watcherSet?
" and set watcher; ": "; data=") +
1957 (data
== null?
"null": data
.length
== 0?
"empty": (
1958 znode
.startsWith(zkw
.znodePaths
.metaZNodePrefix
)?
1959 getServerNameOrEmptyString(data
):
1960 znode
.startsWith(zkw
.znodePaths
.backupMasterAddressesZNode
)?
1961 getServerNameOrEmptyString(data
):
1962 StringUtils
.abbreviate(Bytes
.toStringBinary(data
), 32)))));
1965 private static String
getServerNameOrEmptyString(final byte [] data
) {
1967 return ProtobufUtil
.parseServerNameFrom(data
).toString();
1968 } catch (DeserializationException e
) {
1974 * Waits for HBase installation's base (parent) znode to become available.
1975 * @throws IOException on ZK errors
1977 public static void waitForBaseZNode(Configuration conf
) throws IOException
{
1978 LOG
.info("Waiting until the base znode is available");
1979 String parentZNode
= conf
.get(HConstants
.ZOOKEEPER_ZNODE_PARENT
,
1980 HConstants
.DEFAULT_ZOOKEEPER_ZNODE_PARENT
);
1981 ZooKeeper zk
= new ZooKeeper(ZKConfig
.getZKQuorumServersString(conf
),
1982 conf
.getInt(HConstants
.ZK_SESSION_TIMEOUT
,
1983 HConstants
.DEFAULT_ZK_SESSION_TIMEOUT
), EmptyWatcher
.instance
);
1985 final int maxTimeMs
= 10000;
1986 final int maxNumAttempts
= maxTimeMs
/ HConstants
.SOCKET_RETRY_WAIT_MS
;
1988 KeeperException keeperEx
= null;
1991 for (int attempt
= 0; attempt
< maxNumAttempts
; ++attempt
) {
1993 if (zk
.exists(parentZNode
, false) != null) {
1994 LOG
.info("Parent znode exists: " + parentZNode
);
1998 } catch (KeeperException e
) {
2001 Threads
.sleepWithoutInterrupt(HConstants
.SOCKET_RETRY_WAIT_MS
);
2006 } catch (InterruptedException ex
) {
2007 Thread
.currentThread().interrupt();
2010 if (keeperEx
!= null) {
2011 throw new IOException(keeperEx
);
2016 * Convert a {@link DeserializationException} to a more palatable {@link KeeperException}.
2017 * Used when can't let a {@link DeserializationException} out w/o changing public API.
2018 * @param e Exception to convert
2019 * @return Converted exception
2021 public static KeeperException
convert(final DeserializationException e
) {
2022 KeeperException ke
= new KeeperException
.DataInconsistencyException();
2028 * Recursively print the current state of ZK (non-transactional)
2029 * @param root name of the root directory in zk to print
2031 public static void logZKTree(ZooKeeperWatcher zkw
, String root
) {
2032 if (!LOG
.isDebugEnabled()) return;
2033 LOG
.debug("Current zk system:");
2034 String prefix
= "|-";
2035 LOG
.debug(prefix
+ root
);
2037 logZKTree(zkw
, root
, prefix
);
2038 } catch (KeeperException e
) {
2039 throw new RuntimeException(e
);
2044 * Helper method to print the current state of the ZK tree.
2045 * @see #logZKTree(ZooKeeperWatcher, String)
2046 * @throws KeeperException if an unexpected exception occurs
2048 protected static void logZKTree(ZooKeeperWatcher zkw
, String root
, String prefix
)
2049 throws KeeperException
{
2050 List
<String
> children
= ZKUtil
.listChildrenNoWatch(zkw
, root
);
2051 if (children
== null) return;
2052 for (String child
: children
) {
2053 LOG
.debug(prefix
+ child
);
2054 String node
= ZKUtil
.joinZNode(root
.equals("/") ?
"" : root
, child
);
2055 logZKTree(zkw
, node
, prefix
+ "---");
2061 * @return Serialized protobuf of <code>position</code> with pb magic prefix prepended suitable
2062 * for use as content of an wal position in a replication queue.
2064 public static byte[] positionToByteArray(final long position
) {
2065 byte[] bytes
= ReplicationProtos
.ReplicationHLogPosition
.newBuilder().setPosition(position
)
2066 .build().toByteArray();
2067 return ProtobufUtil
.prependPBMagic(bytes
);
2071 * @param bytes - Content of a WAL position znode.
2072 * @return long - The current WAL position.
2073 * @throws DeserializationException
2075 public static long parseWALPositionFrom(final byte[] bytes
) throws DeserializationException
{
2076 if (bytes
== null) {
2077 throw new DeserializationException("Unable to parse null WAL position.");
2079 if (ProtobufUtil
.isPBMagicPrefix(bytes
)) {
2080 int pblen
= ProtobufUtil
.lengthOfPBMagic();
2081 ReplicationProtos
.ReplicationHLogPosition
.Builder builder
=
2082 ReplicationProtos
.ReplicationHLogPosition
.newBuilder();
2083 ReplicationProtos
.ReplicationHLogPosition position
;
2085 ProtobufUtil
.mergeFrom(builder
, bytes
, pblen
, bytes
.length
- pblen
);
2086 position
= builder
.build();
2087 } catch (IOException e
) {
2088 throw new DeserializationException(e
);
2090 return position
.getPosition();
2092 if (bytes
.length
> 0) {
2093 return Bytes
.toLong(bytes
);
2100 * @param regionLastFlushedSequenceId the flushed sequence id of a region which is the min of its
2102 * @param storeSequenceIds column family to sequence Id map
2103 * @return Serialized protobuf of <code>RegionSequenceIds</code> with pb magic prefix prepended
2104 * suitable for use to filter wal edits in distributedLogReplay mode
2106 public static byte[] regionSequenceIdsToByteArray(final Long regionLastFlushedSequenceId
,
2107 final Map
<byte[], Long
> storeSequenceIds
) {
2108 ClusterStatusProtos
.RegionStoreSequenceIds
.Builder regionSequenceIdsBuilder
=
2109 ClusterStatusProtos
.RegionStoreSequenceIds
.newBuilder();
2110 ClusterStatusProtos
.StoreSequenceId
.Builder storeSequenceIdBuilder
=
2111 ClusterStatusProtos
.StoreSequenceId
.newBuilder();
2112 if (storeSequenceIds
!= null) {
2113 for (Map
.Entry
<byte[], Long
> e
: storeSequenceIds
.entrySet()){
2114 byte[] columnFamilyName
= e
.getKey();
2115 Long curSeqId
= e
.getValue();
2116 storeSequenceIdBuilder
.setFamilyName(UnsafeByteOperations
.unsafeWrap(columnFamilyName
));
2117 storeSequenceIdBuilder
.setSequenceId(curSeqId
);
2118 regionSequenceIdsBuilder
.addStoreSequenceId(storeSequenceIdBuilder
.build());
2119 storeSequenceIdBuilder
.clear();
2122 regionSequenceIdsBuilder
.setLastFlushedSequenceId(regionLastFlushedSequenceId
);
2123 byte[] result
= regionSequenceIdsBuilder
.build().toByteArray();
2124 return ProtobufUtil
.prependPBMagic(result
);
2128 * @param bytes Content of serialized data of RegionStoreSequenceIds
2129 * @return a RegionStoreSequenceIds object
2130 * @throws DeserializationException
2132 public static RegionStoreSequenceIds
parseRegionStoreSequenceIds(final byte[] bytes
)
2133 throws DeserializationException
{
2134 if (bytes
== null || !ProtobufUtil
.isPBMagicPrefix(bytes
)) {
2135 throw new DeserializationException("Unable to parse RegionStoreSequenceIds.");
2137 RegionStoreSequenceIds
.Builder regionSequenceIdsBuilder
=
2138 ClusterStatusProtos
.RegionStoreSequenceIds
.newBuilder();
2139 int pblen
= ProtobufUtil
.lengthOfPBMagic();
2140 RegionStoreSequenceIds storeIds
= null;
2142 ProtobufUtil
.mergeFrom(regionSequenceIdsBuilder
, bytes
, pblen
, bytes
.length
- pblen
);
2143 storeIds
= regionSequenceIdsBuilder
.build();
2144 } catch (IOException e
) {
2145 throw new DeserializationException(e
);