You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@zookeeper.apache.org by GitBox <gi...@apache.org> on 2021/05/11 15:22:02 UTC
[GitHub] [zookeeper] muse-dev[bot] commented on a change in pull request #1695: Comment
muse-dev[bot] commented on a change in pull request #1695:
URL: https://github.com/apache/zookeeper/pull/1695#discussion_r630277728
##########
File path: zookeeper-server/src/main/java/org/apache/zookeeper/server/LogFormatter.java
##########
@@ -0,0 +1,106 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.server;
+
+import java.io.EOFException;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.text.DateFormat;
+import java.util.Date;
+import java.util.zip.Adler32;
+import java.util.zip.Checksum;
+
+import org.apache.jute.BinaryInputArchive;
+import org.apache.jute.Record;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.zookeeper.server.persistence.FileHeader;
+import org.apache.zookeeper.server.persistence.FileTxnLog;
+import org.apache.zookeeper.server.util.SerializeUtils;
+import org.apache.zookeeper.txn.TxnHeader;
+
+@InterfaceAudience.Public
+public class LogFormatter {
+ private static final Logger LOG = LoggerFactory.getLogger(LogFormatter.class);
+
+ /**
+ * @param args
+ */
+ public static void main(String[] args) throws Exception {
+ if (args.length != 1) {
+ System.err.println("USAGE: LogFormatter log_file");
+ System.exit(2);
+ }
+ FileInputStream fis = new FileInputStream(args[0]);
Review comment:
*PATH_TRAVERSAL_IN:* This API (java/io/FileInputStream.<init>(Ljava/lang/String;)V) reads a file whose location might be specified by user input [(details)](https://find-sec-bugs.github.io/bugs.htm#PATH_TRAVERSAL_IN)
(at-me [in a reply](https://docs.muse.dev/docs/talk-to-muse/) with `help` or `ignore`)
##########
File path: zookeeper-server/src/main/java/org/apache/zookeeper/server/SnapshotFormatter.java
##########
@@ -0,0 +1,128 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.server;
+
+import java.io.BufferedInputStream;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.zip.Adler32;
+import java.util.zip.CheckedInputStream;
+
+import org.apache.jute.BinaryInputArchive;
+import org.apache.jute.InputArchive;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.apache.zookeeper.data.StatPersisted;
+import org.apache.zookeeper.server.persistence.FileSnap;
+
+/**
+ * Dump a snapshot file to stdout.
+ */
+@InterfaceAudience.Public
+public class SnapshotFormatter {
+
+ /**
+ * USAGE: SnapshotFormatter snapshot_file
+ */
+ public static void main(String[] args) throws Exception {
+ if (args.length != 1) {
+ System.err.println("USAGE: SnapshotFormatter snapshot_file");
+ System.exit(2);
+ }
+
+ new SnapshotFormatter().run(args[0]);
+ }
+
+ public void run(String snapshotFileName) throws IOException {
+ InputStream is = new CheckedInputStream(
Review comment:
*PATH_TRAVERSAL_IN:* This API (java/io/FileInputStream.<init>(Ljava/lang/String;)V) reads a file whose location might be specified by user input [(details)](https://find-sec-bugs.github.io/bugs.htm#PATH_TRAVERSAL_IN)
(at-me [in a reply](https://docs.muse.dev/docs/talk-to-muse/) with `help` or `ignore`)
##########
File path: zookeeper-server/src/main/java/org/apache/zookeeper/ClientCnxn.java
##########
@@ -0,0 +1,1665 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper;
+
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.io.BufferedReader;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.net.ConnectException;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.net.SocketAddress;
+import java.net.SocketException;
+import java.nio.ByteBuffer;
+import java.util.HashSet;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.Map.Entry;
+import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import javax.security.auth.login.LoginException;
+import javax.security.sasl.SaslException;
+
+import org.apache.jute.BinaryInputArchive;
+import org.apache.jute.BinaryOutputArchive;
+import org.apache.jute.Record;
+import org.apache.zookeeper.AsyncCallback.ACLCallback;
+import org.apache.zookeeper.AsyncCallback.Children2Callback;
+import org.apache.zookeeper.AsyncCallback.ChildrenCallback;
+import org.apache.zookeeper.AsyncCallback.Create2Callback;
+import org.apache.zookeeper.AsyncCallback.DataCallback;
+import org.apache.zookeeper.AsyncCallback.MultiCallback;
+import org.apache.zookeeper.AsyncCallback.StatCallback;
+import org.apache.zookeeper.AsyncCallback.StringCallback;
+import org.apache.zookeeper.AsyncCallback.VoidCallback;
+import org.apache.zookeeper.KeeperException.Code;
+import org.apache.zookeeper.OpResult.ErrorResult;
+import org.apache.zookeeper.Watcher.Event;
+import org.apache.zookeeper.Watcher.Event.EventType;
+import org.apache.zookeeper.Watcher.Event.KeeperState;
+import org.apache.zookeeper.ZooDefs.OpCode;
+import org.apache.zookeeper.ZooKeeper.States;
+import org.apache.zookeeper.ZooKeeper.WatchRegistration;
+import org.apache.zookeeper.client.ZKClientConfig;
+import org.apache.zookeeper.client.HostProvider;
+import org.apache.zookeeper.client.ZooKeeperSaslClient;
+import org.apache.zookeeper.common.Time;
+import org.apache.zookeeper.proto.AuthPacket;
+import org.apache.zookeeper.proto.ConnectRequest;
+import org.apache.zookeeper.proto.Create2Response;
+import org.apache.zookeeper.proto.CreateResponse;
+import org.apache.zookeeper.proto.ExistsResponse;
+import org.apache.zookeeper.proto.GetACLResponse;
+import org.apache.zookeeper.proto.GetChildren2Response;
+import org.apache.zookeeper.proto.GetChildrenResponse;
+import org.apache.zookeeper.proto.GetDataResponse;
+import org.apache.zookeeper.proto.GetSASLRequest;
+import org.apache.zookeeper.proto.ReplyHeader;
+import org.apache.zookeeper.proto.RequestHeader;
+import org.apache.zookeeper.proto.SetACLResponse;
+import org.apache.zookeeper.proto.SetDataResponse;
+import org.apache.zookeeper.proto.SetWatches;
+import org.apache.zookeeper.proto.WatcherEvent;
+import org.apache.zookeeper.server.ByteBufferInputStream;
+import org.apache.zookeeper.server.ZooKeeperThread;
+import org.apache.zookeeper.server.ZooTrace;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.slf4j.MDC;
+
+/**
+ * This class manages the socket i/o for the client. ClientCnxn maintains a list
+ * of available servers to connect to and "transparently" switches servers it is
+ * connected to as needed.
+ *
+ */
+@SuppressFBWarnings({"EI_EXPOSE_REP", "EI_EXPOSE_REP2"})
+public class ClientCnxn {
+ private static final Logger LOG = LoggerFactory.getLogger(ClientCnxn.class);
+
+ /* ZOOKEEPER-706: If a session has a large number of watches set then
+ * attempting to re-establish those watches after a connection loss may
+ * fail due to the SetWatches request exceeding the server's configured
+ * jute.maxBuffer value. To avoid this we instead split the watch
+ * re-establishement across multiple SetWatches calls. This constant
+ * controls the size of each call. It is set to 128kB to be conservative
+ * with respect to the server's 1MB default for jute.maxBuffer.
+ */
+ private static final int SET_WATCHES_MAX_LENGTH = 128 * 1024;
+
+ static class AuthData {
+ AuthData(String scheme, byte data[]) {
+ this.scheme = scheme;
+ this.data = data;
+ }
+
+ String scheme;
+
+ byte data[];
+ }
+
+ private final CopyOnWriteArraySet<AuthData> authInfo = new CopyOnWriteArraySet<AuthData>();
+
+ /**
+ * These are the packets that have been sent and are waiting for a response.
+ */
+ private final LinkedList<Packet> pendingQueue = new LinkedList<Packet>();
+
+ /**
+ * These are the packets that need to be sent.
+ */
+ private final LinkedBlockingDeque<Packet> outgoingQueue = new LinkedBlockingDeque<Packet>();
+
+ private int connectTimeout;
+
+ /**
+ * The timeout in ms the client negotiated with the server. This is the
+ * "real" timeout, not the timeout request by the client (which may have
+ * been increased/decreased by the server which applies bounds to this
+ * value.
+ */
+ private volatile int negotiatedSessionTimeout;
+
+ private int readTimeout;
+
+ private final int sessionTimeout;
+
+ private final ZooKeeper zooKeeper;
+
+ private final ClientWatchManager watcher;
+
+ private long sessionId;
+
+ private byte sessionPasswd[] = new byte[16];
+
+ /**
+ * If true, the connection is allowed to go to r-o mode. This field's value
+ * is sent, besides other data, during session creation handshake. If the
+ * server on the other side of the wire is partitioned it'll accept
+ * read-only clients only.
+ */
+ private boolean readOnly;
+
+ final String chrootPath;
+
+ final SendThread sendThread;
+
+ final EventThread eventThread;
+
+ /**
+ * Set to true when close is called. Latches the connection such that we
+ * don't attempt to re-connect to the server if in the middle of closing the
+ * connection (client sends session disconnect to server as part of close
+ * operation)
+ */
+ private volatile boolean closing = false;
+
+ /**
+ * A set of ZooKeeper hosts this client could connect to.
+ */
+ private final HostProvider hostProvider;
+
+ /**
+ * Is set to true when a connection to a r/w server is established for the
+ * first time; never changed afterwards.
+ * <p>
+ * Is used to handle situations when client without sessionId connects to a
+ * read-only server. Such client receives "fake" sessionId from read-only
+ * server, but this sessionId is invalid for other servers. So when such
+ * client finds a r/w server, it sends 0 instead of fake sessionId during
+ * connection handshake and establishes new, valid session.
+ * <p>
+ * If this field is false (which implies we haven't seen r/w server before)
+ * then non-zero sessionId is fake, otherwise it is valid.
+ */
+ volatile boolean seenRwServerBefore = false;
+
+
+ public ZooKeeperSaslClient zooKeeperSaslClient;
+
+ private final ZKClientConfig clientConfig;
+ /**
+ * If any request's response in not received in configured requestTimeout
+ * then it is assumed that the response packet is lost.
+ */
+ private long requestTimeout;
+
+ public long getSessionId() {
+ return sessionId;
+ }
+
+ public byte[] getSessionPasswd() {
+ return sessionPasswd;
+ }
+
+ public int getSessionTimeout() {
+ return negotiatedSessionTimeout;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder();
+
+ SocketAddress local = sendThread.getClientCnxnSocket().getLocalSocketAddress();
+ SocketAddress remote = sendThread.getClientCnxnSocket().getRemoteSocketAddress();
+ sb
+ .append("sessionid:0x").append(Long.toHexString(getSessionId()))
+ .append(" local:").append(local)
+ .append(" remoteserver:").append(remote)
+ .append(" lastZxid:").append(lastZxid)
+ .append(" xid:").append(xid)
+ .append(" sent:").append(sendThread.getClientCnxnSocket().getSentCount())
+ .append(" recv:").append(sendThread.getClientCnxnSocket().getRecvCount())
+ .append(" queuedpkts:").append(outgoingQueue.size())
+ .append(" pendingresp:").append(pendingQueue.size())
+ .append(" queuedevents:").append(eventThread.waitingEvents.size());
+
+ return sb.toString();
+ }
+
+ /**
+ * This class allows us to pass the headers and the relevant records around.
+ */
+ static class Packet {
+ RequestHeader requestHeader;
+
+ ReplyHeader replyHeader;
+
+ Record request;
+
+ Record response;
+
+ ByteBuffer bb;
+
+ /** Client's view of the path (may differ due to chroot) **/
+ String clientPath;
+ /** Servers's view of the path (may differ due to chroot) **/
+ String serverPath;
+
+ boolean finished;
+
+ AsyncCallback cb;
+
+ Object ctx;
+
+ WatchRegistration watchRegistration;
+
+ public boolean readOnly;
+
+ WatchDeregistration watchDeregistration;
+
+ /** Convenience ctor */
+ Packet(RequestHeader requestHeader, ReplyHeader replyHeader,
+ Record request, Record response,
+ WatchRegistration watchRegistration) {
+ this(requestHeader, replyHeader, request, response,
+ watchRegistration, false);
+ }
+
+ Packet(RequestHeader requestHeader, ReplyHeader replyHeader,
+ Record request, Record response,
+ WatchRegistration watchRegistration, boolean readOnly) {
+
+ this.requestHeader = requestHeader;
+ this.replyHeader = replyHeader;
+ this.request = request;
+ this.response = response;
+ this.readOnly = readOnly;
+ this.watchRegistration = watchRegistration;
+ }
+
+ public void createBB() {
+ try {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
+ boa.writeInt(-1, "len"); // We'll fill this in later
+ if (requestHeader != null) {
+ requestHeader.serialize(boa, "header");
+ }
+ if (request instanceof ConnectRequest) {
+ request.serialize(boa, "connect");
+ // append "am-I-allowed-to-be-readonly" flag
+ boa.writeBool(readOnly, "readOnly");
+ } else if (request != null) {
+ request.serialize(boa, "request");
+ }
+ baos.close();
+ this.bb = ByteBuffer.wrap(baos.toByteArray());
+ this.bb.putInt(this.bb.capacity() - 4);
+ this.bb.rewind();
+ } catch (IOException e) {
+ LOG.warn("Ignoring unexpected exception", e);
+ }
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder();
+
+ sb.append("clientPath:" + clientPath);
+ sb.append(" serverPath:" + serverPath);
+ sb.append(" finished:" + finished);
+
+ sb.append(" header:: " + requestHeader);
+ sb.append(" replyHeader:: " + replyHeader);
+ sb.append(" request:: " + request);
+ sb.append(" response:: " + response);
+
+ // jute toString is horrible, remove unnecessary newlines
+ return sb.toString().replaceAll("\r*\n+", " ");
+ }
+ }
+
+ /**
+ * Creates a connection object. The actual network connect doesn't get
+ * established until needed. The start() instance method must be called
+ * subsequent to construction.
+ *
+ * @param chrootPath - the chroot of this client. Should be removed from this Class in ZOOKEEPER-838
+ * @param hostProvider
+ * the list of ZooKeeper servers to connect to
+ * @param sessionTimeout
+ * the timeout for connections.
+ * @param zooKeeper
+ * the zookeeper object that this connection is related to.
+ * @param watcher watcher for this connection
+ * @param clientCnxnSocket
+ * the socket implementation used (e.g. NIO/Netty)
+ * @param canBeReadOnly
+ * whether the connection is allowed to go to read-only
+ * mode in case of partitioning
+ * @throws IOException
+ */
+ public ClientCnxn(String chrootPath, HostProvider hostProvider, int sessionTimeout, ZooKeeper zooKeeper,
+ ClientWatchManager watcher, ClientCnxnSocket clientCnxnSocket, boolean canBeReadOnly)
+ throws IOException {
+ this(chrootPath, hostProvider, sessionTimeout, zooKeeper, watcher,
+ clientCnxnSocket, 0, new byte[16], canBeReadOnly);
+ }
+
+ /**
+ * Creates a connection object. The actual network connect doesn't get
+ * established until needed. The start() instance method must be called
+ * subsequent to construction.
+ *
+ * @param chrootPath - the chroot of this client. Should be removed from this Class in ZOOKEEPER-838
+ * @param hostProvider
+ * the list of ZooKeeper servers to connect to
+ * @param sessionTimeout
+ * the timeout for connections.
+ * @param zooKeeper
+ * the zookeeper object that this connection is related to.
+ * @param watcher watcher for this connection
+ * @param clientCnxnSocket
+ * the socket implementation used (e.g. NIO/Netty)
+ * @param sessionId session id if re-establishing session
+ * @param sessionPasswd session passwd if re-establishing session
+ * @param canBeReadOnly
+ * whether the connection is allowed to go to read-only
+ * mode in case of partitioning
+ * @throws IOException
+ */
+ public ClientCnxn(String chrootPath, HostProvider hostProvider, int sessionTimeout, ZooKeeper zooKeeper,
+ ClientWatchManager watcher, ClientCnxnSocket clientCnxnSocket,
+ long sessionId, byte[] sessionPasswd, boolean canBeReadOnly) {
+ this.zooKeeper = zooKeeper;
+ this.watcher = watcher;
+ this.sessionId = sessionId;
+ this.sessionPasswd = sessionPasswd;
+ this.sessionTimeout = sessionTimeout;
+ this.hostProvider = hostProvider;
+ this.chrootPath = chrootPath;
+
+ connectTimeout = sessionTimeout / hostProvider.size();
+ readTimeout = sessionTimeout * 2 / 3;
+ readOnly = canBeReadOnly;
+
+ sendThread = new SendThread(clientCnxnSocket);
+ eventThread = new EventThread();
+ this.clientConfig=zooKeeper.getClientConfig();
+ initRequestTimeout();
+ }
+
+ public void start() {
+ sendThread.start();
+ eventThread.start();
+ }
+
+ private Object eventOfDeath = new Object();
+
+ private static class WatcherSetEventPair {
+ private final Set<Watcher> watchers;
+ private final WatchedEvent event;
+
+ public WatcherSetEventPair(Set<Watcher> watchers, WatchedEvent event) {
+ this.watchers = watchers;
+ this.event = event;
+ }
+ }
+
+ /**
+ * Guard against creating "-EventThread-EventThread-EventThread-..." thread
+ * names when ZooKeeper object is being created from within a watcher.
+ * See ZOOKEEPER-795 for details.
+ */
+ private static String makeThreadName(String suffix) {
+ String name = Thread.currentThread().getName().
+ replaceAll("-EventThread", "");
+ return name + suffix;
+ }
+
+ class EventThread extends ZooKeeperThread {
+ private final LinkedBlockingQueue<Object> waitingEvents =
+ new LinkedBlockingQueue<Object>();
+
+ /** This is really the queued session state until the event
+ * thread actually processes the event and hands it to the watcher.
+ * But for all intents and purposes this is the state.
+ */
+ private volatile KeeperState sessionState = KeeperState.Disconnected;
+
+ private volatile boolean wasKilled = false;
+ private volatile boolean isRunning = false;
+
+ EventThread() {
+ super(makeThreadName("-EventThread"));
+ setDaemon(true);
+ }
+
+ public void queueEvent(WatchedEvent event) {
+ queueEvent(event, null);
+ }
+
+ private void queueEvent(WatchedEvent event,
+ Set<Watcher> materializedWatchers) {
+ if (event.getType() == EventType.None
+ && sessionState == event.getState()) {
+ return;
+ }
+ sessionState = event.getState();
+ final Set<Watcher> watchers;
+ if (materializedWatchers == null) {
+ // materialize the watchers based on the event
+ watchers = watcher.materialize(event.getState(),
+ event.getType(), event.getPath());
+ } else {
+ watchers = new HashSet<Watcher>();
+ watchers.addAll(materializedWatchers);
+ }
+ WatcherSetEventPair pair = new WatcherSetEventPair(watchers, event);
+ // queue the pair (watch set & event) for later processing
+ waitingEvents.add(pair);
+ }
+
+ public void queueCallback(AsyncCallback cb, int rc, String path,
+ Object ctx) {
+ waitingEvents.add(new LocalCallback(cb, rc, path, ctx));
+ }
+
+ @SuppressFBWarnings("JLM_JSR166_UTILCONCURRENT_MONITORENTER")
+ public void queuePacket(Packet packet) {
+ if (wasKilled) {
+ synchronized (waitingEvents) {
+ if (isRunning) waitingEvents.add(packet);
+ else processEvent(packet);
+ }
+ } else {
+ waitingEvents.add(packet);
+ }
+ }
+
+ public void queueEventOfDeath() {
+ waitingEvents.add(eventOfDeath);
+ }
+
+ @Override
+ @SuppressFBWarnings("JLM_JSR166_UTILCONCURRENT_MONITORENTER")
+ public void run() {
+ try {
+ isRunning = true;
+ while (true) {
+ Object event = waitingEvents.take();
+ if (event == eventOfDeath) {
+ wasKilled = true;
+ } else {
+ processEvent(event);
+ }
+ if (wasKilled)
+ synchronized (waitingEvents) {
+ if (waitingEvents.isEmpty()) {
+ isRunning = false;
+ break;
+ }
+ }
+ }
+ } catch (InterruptedException e) {
+ LOG.error("Event thread exiting due to interruption", e);
+ }
+
+ LOG.info("EventThread shut down for session: 0x{}",
+ Long.toHexString(getSessionId()));
+ }
+
+ private void processEvent(Object event) {
+ try {
+ if (event instanceof WatcherSetEventPair) {
+ // each watcher will process the event
+ WatcherSetEventPair pair = (WatcherSetEventPair) event;
+ for (Watcher watcher : pair.watchers) {
+ try {
+ watcher.process(pair.event);
+ } catch (Throwable t) {
+ LOG.error("Error while calling watcher ", t);
+ }
+ }
+ } else if (event instanceof LocalCallback) {
+ LocalCallback lcb = (LocalCallback) event;
+ if (lcb.cb instanceof StatCallback) {
+ ((StatCallback) lcb.cb).processResult(lcb.rc, lcb.path,
+ lcb.ctx, null);
+ } else if (lcb.cb instanceof DataCallback) {
+ ((DataCallback) lcb.cb).processResult(lcb.rc, lcb.path,
+ lcb.ctx, null, null);
+ } else if (lcb.cb instanceof ACLCallback) {
+ ((ACLCallback) lcb.cb).processResult(lcb.rc, lcb.path,
+ lcb.ctx, null, null);
+ } else if (lcb.cb instanceof ChildrenCallback) {
+ ((ChildrenCallback) lcb.cb).processResult(lcb.rc,
+ lcb.path, lcb.ctx, null);
+ } else if (lcb.cb instanceof Children2Callback) {
+ ((Children2Callback) lcb.cb).processResult(lcb.rc,
+ lcb.path, lcb.ctx, null, null);
+ } else if (lcb.cb instanceof StringCallback) {
+ ((StringCallback) lcb.cb).processResult(lcb.rc,
+ lcb.path, lcb.ctx, null);
+ } else {
+ ((VoidCallback) lcb.cb).processResult(lcb.rc, lcb.path,
+ lcb.ctx);
+ }
+ } else {
+ Packet p = (Packet) event;
+ int rc = 0;
+ String clientPath = p.clientPath;
+ if (p.replyHeader.getErr() != 0) {
+ rc = p.replyHeader.getErr();
+ }
+ if (p.cb == null) {
+ LOG.warn("Somehow a null cb got to EventThread!");
+ } else if (p.response instanceof ExistsResponse
+ || p.response instanceof SetDataResponse
+ || p.response instanceof SetACLResponse) {
+ StatCallback cb = (StatCallback) p.cb;
+ if (rc == 0) {
+ if (p.response instanceof ExistsResponse) {
+ cb.processResult(rc, clientPath, p.ctx,
+ ((ExistsResponse) p.response)
+ .getStat());
+ } else if (p.response instanceof SetDataResponse) {
+ cb.processResult(rc, clientPath, p.ctx,
+ ((SetDataResponse) p.response)
+ .getStat());
+ } else if (p.response instanceof SetACLResponse) {
+ cb.processResult(rc, clientPath, p.ctx,
+ ((SetACLResponse) p.response)
+ .getStat());
+ }
+ } else {
+ cb.processResult(rc, clientPath, p.ctx, null);
+ }
+ } else if (p.response instanceof GetDataResponse) {
+ DataCallback cb = (DataCallback) p.cb;
+ GetDataResponse rsp = (GetDataResponse) p.response;
+ if (rc == 0) {
+ cb.processResult(rc, clientPath, p.ctx, rsp
+ .getData(), rsp.getStat());
+ } else {
+ cb.processResult(rc, clientPath, p.ctx, null,
+ null);
+ }
+ } else if (p.response instanceof GetACLResponse) {
+ ACLCallback cb = (ACLCallback) p.cb;
+ GetACLResponse rsp = (GetACLResponse) p.response;
+ if (rc == 0) {
+ cb.processResult(rc, clientPath, p.ctx, rsp
+ .getAcl(), rsp.getStat());
+ } else {
+ cb.processResult(rc, clientPath, p.ctx, null,
+ null);
+ }
+ } else if (p.response instanceof GetChildrenResponse) {
+ ChildrenCallback cb = (ChildrenCallback) p.cb;
+ GetChildrenResponse rsp = (GetChildrenResponse) p.response;
+ if (rc == 0) {
+ cb.processResult(rc, clientPath, p.ctx, rsp
+ .getChildren());
+ } else {
+ cb.processResult(rc, clientPath, p.ctx, null);
+ }
+ } else if (p.response instanceof GetChildren2Response) {
+ Children2Callback cb = (Children2Callback) p.cb;
+ GetChildren2Response rsp = (GetChildren2Response) p.response;
+ if (rc == 0) {
+ cb.processResult(rc, clientPath, p.ctx, rsp
+ .getChildren(), rsp.getStat());
+ } else {
+ cb.processResult(rc, clientPath, p.ctx, null, null);
+ }
+ } else if (p.response instanceof CreateResponse) {
+ StringCallback cb = (StringCallback) p.cb;
+ CreateResponse rsp = (CreateResponse) p.response;
+ if (rc == 0) {
+ cb.processResult(rc, clientPath, p.ctx,
+ (chrootPath == null
+ ? rsp.getPath()
+ : rsp.getPath()
+ .substring(chrootPath.length())));
+ } else {
+ cb.processResult(rc, clientPath, p.ctx, null);
+ }
+ } else if (p.response instanceof Create2Response) {
+ Create2Callback cb = (Create2Callback) p.cb;
+ Create2Response rsp = (Create2Response) p.response;
+ if (rc == 0) {
+ cb.processResult(rc, clientPath, p.ctx,
+ (chrootPath == null
+ ? rsp.getPath()
+ : rsp.getPath()
+ .substring(chrootPath.length())), rsp.getStat());
+ } else {
+ cb.processResult(rc, clientPath, p.ctx, null, null);
+ }
+ } else if (p.response instanceof MultiResponse) {
+ MultiCallback cb = (MultiCallback) p.cb;
+ MultiResponse rsp = (MultiResponse) p.response;
+ if (rc == 0) {
+ List<OpResult> results = rsp.getResultList();
+ int newRc = rc;
+ for (OpResult result : results) {
+ if (result instanceof ErrorResult
+ && KeeperException.Code.OK.intValue() != (newRc = ((ErrorResult) result)
+ .getErr())) {
+ break;
+ }
+ }
+ cb.processResult(newRc, clientPath, p.ctx, results);
+ } else {
+ cb.processResult(rc, clientPath, p.ctx, null);
+ }
+ } else if (p.cb instanceof VoidCallback) {
+ VoidCallback cb = (VoidCallback) p.cb;
+ cb.processResult(rc, clientPath, p.ctx);
+ }
+ }
+ } catch (Throwable t) {
+ LOG.error("Caught unexpected throwable", t);
+ }
+ }
+ }
+
+ // @VisibleForTesting
+ protected void finishPacket(Packet p) {
+ int err = p.replyHeader.getErr();
+ if (p.watchRegistration != null) {
+ p.watchRegistration.register(err);
+ }
+ // Add all the removed watch events to the event queue, so that the
+ // clients will be notified with 'Data/Child WatchRemoved' event type.
+ if (p.watchDeregistration != null) {
+ Map<EventType, Set<Watcher>> materializedWatchers = null;
+ try {
+ materializedWatchers = p.watchDeregistration.unregister(err);
+ for (Entry<EventType, Set<Watcher>> entry : materializedWatchers
+ .entrySet()) {
+ Set<Watcher> watchers = entry.getValue();
+ if (watchers.size() > 0) {
+ queueEvent(p.watchDeregistration.getClientPath(), err,
+ watchers, entry.getKey());
+ // ignore connectionloss when removing from local
+ // session
+ p.replyHeader.setErr(Code.OK.intValue());
+ }
+ }
+ } catch (KeeperException.NoWatcherException nwe) {
+ p.replyHeader.setErr(nwe.code().intValue());
+ } catch (KeeperException ke) {
+ p.replyHeader.setErr(ke.code().intValue());
+ }
+ }
+
+ if (p.cb == null) {
+ synchronized (p) {
+ p.finished = true;
+ p.notifyAll();
+ }
+ } else {
+ p.finished = true;
+ eventThread.queuePacket(p);
+ }
+ }
+
+ void queueEvent(String clientPath, int err,
+ Set<Watcher> materializedWatchers, EventType eventType) {
+ KeeperState sessionState = KeeperState.SyncConnected;
+ if (KeeperException.Code.SESSIONEXPIRED.intValue() == err
+ || KeeperException.Code.CONNECTIONLOSS.intValue() == err) {
+ sessionState = Event.KeeperState.Disconnected;
+ }
+ WatchedEvent event = new WatchedEvent(eventType, sessionState,
+ clientPath);
+ eventThread.queueEvent(event, materializedWatchers);
+ }
+
+ void queueCallback(AsyncCallback cb, int rc, String path, Object ctx) {
+ eventThread.queueCallback(cb, rc, path, ctx);
+ }
+
+ private void conLossPacket(Packet p) {
+ if (p.replyHeader == null) {
+ return;
+ }
+ switch (state) {
+ case AUTH_FAILED:
+ p.replyHeader.setErr(KeeperException.Code.AUTHFAILED.intValue());
+ break;
+ case CLOSED:
+ p.replyHeader.setErr(KeeperException.Code.SESSIONEXPIRED.intValue());
+ break;
+ default:
+ p.replyHeader.setErr(KeeperException.Code.CONNECTIONLOSS.intValue());
+ }
+ finishPacket(p);
+ }
+
+ private volatile long lastZxid;
+
+ public long getLastZxid() {
+ return lastZxid;
+ }
+
+ static class EndOfStreamException extends IOException {
+ private static final long serialVersionUID = -5438877188796231422L;
+
+ public EndOfStreamException(String msg) {
+ super(msg);
+ }
+
+ @Override
+ public String toString() {
+ return "EndOfStreamException: " + getMessage();
+ }
+ }
+
+ private static class SessionTimeoutException extends IOException {
+ private static final long serialVersionUID = 824482094072071178L;
+
+ public SessionTimeoutException(String msg) {
+ super(msg);
+ }
+ }
+
+ private static class SessionExpiredException extends IOException {
+ private static final long serialVersionUID = -1388816932076193249L;
+
+ public SessionExpiredException(String msg) {
+ super(msg);
+ }
+ }
+
+ private static class RWServerFoundException extends IOException {
+ private static final long serialVersionUID = 90431199887158758L;
+
+ public RWServerFoundException(String msg) {
+ super(msg);
+ }
+ }
+
+ /**
+ * This class services the outgoing request queue and generates the heart
+ * beats. It also spawns the ReadThread.
+ */
+ class SendThread extends ZooKeeperThread {
+ private long lastPingSentNs;
+ private final ClientCnxnSocket clientCnxnSocket;
+ private Random r = new Random();
Review comment:
*PREDICTABLE_RANDOM:* This random generator (java.util.Random) is predictable [(details)](https://find-sec-bugs.github.io/bugs.htm#PREDICTABLE_RANDOM)
(at-me [in a reply](https://docs.muse.dev/docs/talk-to-muse/) with `help` or `ignore`)
##########
File path: zookeeper-server/src/main/java/org/apache/zookeeper/Login.java
##########
@@ -0,0 +1,422 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper;
+
+/**
+ * This class is responsible for refreshing Kerberos credentials for
+ * logins for both Zookeeper client and server.
+ * See ZooKeeperSaslServer for server-side usage.
+ * See ZooKeeperSaslClient for client-side usage.
+ */
+
+import javax.security.auth.kerberos.KerberosPrincipal;
+import javax.security.auth.login.AppConfigurationEntry;
+import javax.security.auth.login.Configuration;
+import javax.security.auth.login.LoginContext;
+import javax.security.auth.login.LoginException;
+import javax.security.auth.callback.CallbackHandler;
+
+import org.apache.zookeeper.client.ZKClientConfig;
+import org.apache.zookeeper.common.ZKConfig;
+import org.apache.zookeeper.server.ZooKeeperSaslServer;
+import org.apache.zookeeper.common.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.security.auth.kerberos.KerberosTicket;
+import javax.security.auth.Subject;
+
+import java.util.Date;
+import java.util.Random;
+import java.util.Set;
+
+public class Login {
+ private static final String KINIT_COMMAND_DEFAULT = "/usr/bin/kinit";
+ private static final Logger LOG = LoggerFactory.getLogger(Login.class);
+ public CallbackHandler callbackHandler;
+
+ // LoginThread will sleep until 80% of time from last refresh to
+ // ticket's expiry has been reached, at which time it will wake
+ // and try to renew the ticket.
+ private static final float TICKET_RENEW_WINDOW = 0.80f;
+
+ /**
+ * Percentage of random jitter added to the renewal time
+ */
+ private static final float TICKET_RENEW_JITTER = 0.05f;
+
+ // Regardless of TICKET_RENEW_WINDOW setting above and the ticket expiry time,
+ // thread will not sleep between refresh attempts any less than 1 minute (60*1000 milliseconds = 1 minute).
+ // Change the '1' to e.g. 5, to change this to 5 minutes.
+ private static final long MIN_TIME_BEFORE_RELOGIN = 1 * 60 * 1000L;
+
+ private Subject subject = null;
+ private Thread t = null;
+ private boolean isKrbTicket = false;
+ private boolean isUsingTicketCache = false;
+
+ /** Random number generator */
+ private static Random rng = new Random();
Review comment:
*PREDICTABLE_RANDOM:* This random generator (java.util.Random) is predictable [(details)](https://find-sec-bugs.github.io/bugs.htm#PREDICTABLE_RANDOM)
(at-me [in a reply](https://docs.muse.dev/docs/talk-to-muse/) with `help` or `ignore`)
##########
File path: zookeeper-server/src/main/java/org/apache/zookeeper/server/SyncRequestProcessor.java
##########
@@ -0,0 +1,217 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.server;
+
+import java.io.Flushable;
+import java.io.IOException;
+import java.util.LinkedList;
+import java.util.Random;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This RequestProcessor logs requests to disk. It batches the requests to do
+ * the io efficiently. The request is not passed to the next RequestProcessor
+ * until its log has been synced to disk.
+ *
+ * SyncRequestProcessor is used in 3 different cases
+ * 1. Leader - Sync request to disk and forward it to AckRequestProcessor which
+ * send ack back to itself.
+ * 2. Follower - Sync request to disk and forward request to
+ * SendAckRequestProcessor which send the packets to leader.
+ * SendAckRequestProcessor is flushable which allow us to force
+ * push packets to leader.
+ * 3. Observer - Sync committed request to disk (received as INFORM packet).
+ * It never send ack back to the leader, so the nextProcessor will
+ * be null. This change the semantic of txnlog on the observer
+ * since it only contains committed txns.
+ */
+public class SyncRequestProcessor extends ZooKeeperCriticalThread implements
+ RequestProcessor {
+ private static final Logger LOG = LoggerFactory.getLogger(SyncRequestProcessor.class);
+ private final ZooKeeperServer zks;
+ private final LinkedBlockingQueue<Request> queuedRequests =
+ new LinkedBlockingQueue<Request>();
+ private final RequestProcessor nextProcessor;
+
+ private Thread snapInProcess = null;
+ volatile private boolean running;
+
+ /**
+ * Transactions that have been written and are waiting to be flushed to
+ * disk. Basically this is the list of SyncItems whose callbacks will be
+ * invoked after flush returns successfully.
+ */
+ private final LinkedList<Request> toFlush = new LinkedList<Request>();
+ private final Random r = new Random();
Review comment:
*PREDICTABLE_RANDOM:* This random generator (java.util.Random) is predictable [(details)](https://find-sec-bugs.github.io/bugs.htm#PREDICTABLE_RANDOM)
(at-me [in a reply](https://docs.muse.dev/docs/talk-to-muse/) with `help` or `ignore`)
##########
File path: zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/AuthFastLeaderElection.java
##########
@@ -0,0 +1,986 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.server.quorum;
+
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.io.IOException;
+import java.net.DatagramPacket;
+import java.net.DatagramSocket;
+import java.net.InetSocketAddress;
+import java.net.SocketException;
+import java.nio.ByteBuffer;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.Semaphore;
+
+import java.util.concurrent.TimeUnit;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.zookeeper.common.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.zookeeper.jmx.MBeanRegistry;
+import org.apache.zookeeper.server.ZooKeeperThread;
+import org.apache.zookeeper.server.quorum.Election;
+import org.apache.zookeeper.server.quorum.Vote;
+import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
+import org.apache.zookeeper.server.quorum.QuorumPeer.ServerState;
+
+/**
+ * @deprecated This class has been deprecated as of release 3.4.0.
+ */
+@Deprecated
+public class AuthFastLeaderElection implements Election {
+ private static final Logger LOG = LoggerFactory.getLogger(AuthFastLeaderElection.class);
+
+ /* Sequence numbers for messages */
+ static int sequencer = 0;
+ static int maxTag = 0;
+
+ /*
+ * Determine how much time a process has to wait once it believes that it
+ * has reached the end of leader election.
+ */
+ static int finalizeWait = 100;
+
+ /*
+ * Challenge counter to avoid replay attacks
+ */
+
+ static int challengeCounter = 0;
+
+ /*
+ * Flag to determine whether to authenticate or not
+ */
+
+ private boolean authEnabled = false;
+
+ static public class Notification {
+ /*
+ * Proposed leader
+ */
+ long leader;
+
+ /*
+ * zxid of the proposed leader
+ */
+ long zxid;
+
+ /*
+ * Epoch
+ */
+ long epoch;
+
+ /*
+ * current state of sender
+ */
+ QuorumPeer.ServerState state;
+
+ /*
+ * Address of the sender
+ */
+ InetSocketAddress addr;
+ }
+
+ /*
+ * Messages to send, both Notifications and Acks
+ */
+ static public class ToSend {
+ static enum mType {
+ crequest, challenge, notification, ack
+ }
+
+ ToSend(mType type, long tag, long leader, long zxid, long epoch,
+ ServerState state, InetSocketAddress addr) {
+
+ switch (type) {
+ case crequest:
+ this.type = 0;
+ this.tag = tag;
+ this.leader = leader;
+ this.zxid = zxid;
+ this.epoch = epoch;
+ this.state = state;
+ this.addr = addr;
+
+ break;
+ case challenge:
+ this.type = 1;
+ this.tag = tag;
+ this.leader = leader;
+ this.zxid = zxid;
+ this.epoch = epoch;
+ this.state = state;
+ this.addr = addr;
+
+ break;
+ case notification:
+ this.type = 2;
+ this.leader = leader;
+ this.zxid = zxid;
+ this.epoch = epoch;
+ this.state = QuorumPeer.ServerState.LOOKING;
+ this.tag = tag;
+ this.addr = addr;
+
+ break;
+ case ack:
+ this.type = 3;
+ this.tag = tag;
+ this.leader = leader;
+ this.zxid = zxid;
+ this.epoch = epoch;
+ this.state = state;
+ this.addr = addr;
+
+ break;
+ default:
+ break;
+ }
+ }
+
+ /*
+ * Message type: 0 notification, 1 acknowledgement
+ */
+ int type;
+
+ /*
+ * Proposed leader in the case of notification
+ */
+ long leader;
+
+ /*
+ * id contains the tag for acks, and zxid for notifications
+ */
+ long zxid;
+
+ /*
+ * Epoch
+ */
+ long epoch;
+
+ /*
+ * Current state;
+ */
+ QuorumPeer.ServerState state;
+
+ /*
+ * Message tag
+ */
+ long tag;
+
+ InetSocketAddress addr;
+ }
+
+ LinkedBlockingQueue<ToSend> sendqueue;
+
+ LinkedBlockingQueue<Notification> recvqueue;
+
+ private class Messenger {
+
+ final DatagramSocket mySocket;
+ long lastProposedLeader;
+ long lastProposedZxid;
+ long lastEpoch;
+ final Set<Long> ackset;
+ final ConcurrentHashMap<Long, Long> challengeMap;
+ final ConcurrentHashMap<Long, Semaphore> challengeMutex;
+ final ConcurrentHashMap<Long, Semaphore> ackMutex;
+ final ConcurrentHashMap<InetSocketAddress, ConcurrentHashMap<Long, Long>> addrChallengeMap;
+
+ class WorkerReceiver extends ZooKeeperThread {
+
+ DatagramSocket mySocket;
+ Messenger myMsg;
+
+ WorkerReceiver(DatagramSocket s, Messenger msg) {
+ super("WorkerReceiver-" + s.getRemoteSocketAddress());
+ mySocket = s;
+ myMsg = msg;
+ }
+
+ boolean saveChallenge(long tag, long challenge) {
+ Semaphore s = challengeMutex.get(tag);
+ if (s != null) {
+ synchronized (Messenger.this) {
+ challengeMap.put(tag, challenge);
+ challengeMutex.remove(tag);
+ }
+
+
+ s.release();
+ } else {
+ LOG.error("No challenge mutex object");
+ }
+
+
+ return true;
+ }
+
+ public void run() {
+ byte responseBytes[] = new byte[48];
+ ByteBuffer responseBuffer = ByteBuffer.wrap(responseBytes);
+ DatagramPacket responsePacket = new DatagramPacket(
+ responseBytes, responseBytes.length);
+ while (true) {
+ // Sleeps on receive
+ try {
+ responseBuffer.clear();
+ mySocket.receive(responsePacket);
+ } catch (IOException e) {
+ LOG.warn("Ignoring exception receiving", e);
+ }
+ // Receive new message
+ if (responsePacket.getLength() != responseBytes.length) {
+ LOG.warn("Got a short response: "
+ + responsePacket.getLength() + " "
+ + responsePacket.toString());
+ continue;
+ }
+ responseBuffer.clear();
+ int type = responseBuffer.getInt();
+ if ((type > 3) || (type < 0)) {
+ LOG.warn("Got bad Msg type: " + type);
+ continue;
+ }
+ long tag = responseBuffer.getLong();
+
+ QuorumPeer.ServerState ackstate = QuorumPeer.ServerState.LOOKING;
+ switch (responseBuffer.getInt()) {
+ case 0:
+ ackstate = QuorumPeer.ServerState.LOOKING;
+ break;
+ case 1:
+ ackstate = QuorumPeer.ServerState.LEADING;
+ break;
+ case 2:
+ ackstate = QuorumPeer.ServerState.FOLLOWING;
+ break;
+ default:
+ LOG.warn("unknown type " + responseBuffer.getInt());
+ break;
+ }
+
+ Vote current = self.getCurrentVote();
+
+ switch (type) {
+ case 0:
+ // Receive challenge request
+ ToSend c = new ToSend(ToSend.mType.challenge, tag,
+ current.getId(), current.getZxid(),
+ logicalclock.get(), self.getPeerState(),
+ (InetSocketAddress) responsePacket
+ .getSocketAddress());
+ sendqueue.offer(c);
+ break;
+ case 1:
+ // Receive challenge and store somewhere else
+ long challenge = responseBuffer.getLong();
+ saveChallenge(tag, challenge);
+
+ break;
+ case 2:
+ Notification n = new Notification();
+ n.leader = responseBuffer.getLong();
+ n.zxid = responseBuffer.getLong();
+ n.epoch = responseBuffer.getLong();
+ n.state = ackstate;
+ n.addr = (InetSocketAddress) responsePacket
+ .getSocketAddress();
+
+ if ((myMsg.lastEpoch <= n.epoch)
+ && ((n.zxid > myMsg.lastProposedZxid)
+ || ((n.zxid == myMsg.lastProposedZxid)
+ && (n.leader > myMsg.lastProposedLeader)))) {
+ myMsg.lastProposedZxid = n.zxid;
+ myMsg.lastProposedLeader = n.leader;
+ myMsg.lastEpoch = n.epoch;
+ }
+
+ long recChallenge;
+ InetSocketAddress addr = (InetSocketAddress) responsePacket
+ .getSocketAddress();
+ if (authEnabled) {
+ ConcurrentHashMap<Long, Long> tmpMap = addrChallengeMap.get(addr);
+ if(tmpMap != null){
+ if (tmpMap.get(tag) != null) {
+ recChallenge = responseBuffer.getLong();
+
+ if (tmpMap.get(tag) == recChallenge) {
+ recvqueue.offer(n);
+
+ ToSend a = new ToSend(ToSend.mType.ack,
+ tag, current.getId(),
+ current.getZxid(),
+ logicalclock.get(), self.getPeerState(),
+ addr);
+
+ sendqueue.offer(a);
+ } else {
+ LOG.warn("Incorrect challenge: "
+ + recChallenge + ", "
+ + addrChallengeMap.toString());
+ }
+ } else {
+ LOG.warn("No challenge for host: " + addr
+ + " " + tag);
+ }
+ }
+ } else {
+ recvqueue.offer(n);
+
+ ToSend a = new ToSend(ToSend.mType.ack, tag,
+ current.getId(), current.getZxid(),
+ logicalclock.get(), self.getPeerState(),
+ (InetSocketAddress) responsePacket
+ .getSocketAddress());
+
+ sendqueue.offer(a);
+ }
+ break;
+
+ // Upon reception of an ack message, remove it from the
+ // queue
+ case 3:
+ Semaphore s = ackMutex.get(tag);
+
+ if(s != null)
+ s.release();
+ else LOG.error("Empty ack semaphore");
+
+ ackset.add(tag);
+
+ if (authEnabled) {
+ ConcurrentHashMap<Long, Long> tmpMap = addrChallengeMap.get(responsePacket
+ .getSocketAddress());
+ if(tmpMap != null) {
+ tmpMap.remove(tag);
+ } else {
+ LOG.warn("No such address in the ensemble configuration " + responsePacket
+ .getSocketAddress());
+ }
+ }
+
+ if (ackstate != QuorumPeer.ServerState.LOOKING) {
+ Notification outofsync = new Notification();
+ outofsync.leader = responseBuffer.getLong();
+ outofsync.zxid = responseBuffer.getLong();
+ outofsync.epoch = responseBuffer.getLong();
+ outofsync.state = ackstate;
+ outofsync.addr = (InetSocketAddress) responsePacket
+ .getSocketAddress();
+
+ recvqueue.offer(outofsync);
+ }
+
+ break;
+ // Default case
+ default:
+ LOG.warn("Received message of incorrect type " + type);
+ break;
+ }
+ }
+ }
+ }
+
+ class WorkerSender extends ZooKeeperThread {
+
+ Random rand;
+ int maxAttempts;
+ int ackWait = finalizeWait;
+
+ /*
+ * Receives a socket and max number of attempts as input
+ */
+
+ WorkerSender(int attempts) {
+ super("WorkerSender");
+ maxAttempts = attempts;
+ rand = new Random(java.lang.Thread.currentThread().getId()
+ + Time.currentElapsedTime());
Review comment:
*PREDICTABLE_RANDOM:* This random generator (java.util.Random) is predictable [(details)](https://find-sec-bugs.github.io/bugs.htm#PREDICTABLE_RANDOM)
(at-me [in a reply](https://docs.muse.dev/docs/talk-to-muse/) with `help` or `ignore`)
##########
File path: zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeper.java
##########
@@ -0,0 +1,3114 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper;
+
+import org.apache.jute.Record;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.apache.zookeeper.AsyncCallback.ACLCallback;
+import org.apache.zookeeper.AsyncCallback.Children2Callback;
+import org.apache.zookeeper.AsyncCallback.ChildrenCallback;
+import org.apache.zookeeper.AsyncCallback.Create2Callback;
+import org.apache.zookeeper.AsyncCallback.DataCallback;
+import org.apache.zookeeper.AsyncCallback.MultiCallback;
+import org.apache.zookeeper.AsyncCallback.StatCallback;
+import org.apache.zookeeper.AsyncCallback.StringCallback;
+import org.apache.zookeeper.AsyncCallback.VoidCallback;
+import org.apache.zookeeper.KeeperException.Code;
+import org.apache.zookeeper.KeeperException.NoWatcherException;
+import org.apache.zookeeper.OpResult.ErrorResult;
+import org.apache.zookeeper.Watcher.Event.EventType;
+import org.apache.zookeeper.Watcher.WatcherType;
+import org.apache.zookeeper.client.ConnectStringParser;
+import org.apache.zookeeper.client.HostProvider;
+import org.apache.zookeeper.client.StaticHostProvider;
+import org.apache.zookeeper.client.ZKClientConfig;
+import org.apache.zookeeper.client.ZooKeeperSaslClient;
+import org.apache.zookeeper.common.PathUtils;
+import org.apache.zookeeper.common.StringUtils;
+import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.data.Stat;
+import org.apache.zookeeper.proto.CheckWatchesRequest;
+import org.apache.zookeeper.proto.Create2Response;
+import org.apache.zookeeper.proto.CreateRequest;
+import org.apache.zookeeper.proto.CreateResponse;
+import org.apache.zookeeper.proto.CreateTTLRequest;
+import org.apache.zookeeper.proto.DeleteRequest;
+import org.apache.zookeeper.proto.ExistsRequest;
+import org.apache.zookeeper.proto.GetACLRequest;
+import org.apache.zookeeper.proto.GetACLResponse;
+import org.apache.zookeeper.proto.GetChildren2Request;
+import org.apache.zookeeper.proto.GetChildren2Response;
+import org.apache.zookeeper.proto.GetChildrenRequest;
+import org.apache.zookeeper.proto.GetChildrenResponse;
+import org.apache.zookeeper.proto.GetDataRequest;
+import org.apache.zookeeper.proto.GetDataResponse;
+import org.apache.zookeeper.proto.ReconfigRequest;
+import org.apache.zookeeper.proto.RemoveWatchesRequest;
+import org.apache.zookeeper.proto.ReplyHeader;
+import org.apache.zookeeper.proto.RequestHeader;
+import org.apache.zookeeper.proto.SetACLRequest;
+import org.apache.zookeeper.proto.SetACLResponse;
+import org.apache.zookeeper.proto.SetDataRequest;
+import org.apache.zookeeper.proto.SetDataResponse;
+import org.apache.zookeeper.proto.SyncRequest;
+import org.apache.zookeeper.proto.SyncResponse;
+import org.apache.zookeeper.server.DataTree;
+import org.apache.zookeeper.server.EphemeralType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.lang.reflect.Constructor;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * This is the main class of ZooKeeper client library. To use a ZooKeeper
+ * service, an application must first instantiate an object of ZooKeeper class.
+ * All the iterations will be done by calling the methods of ZooKeeper class.
+ * The methods of this class are thread-safe unless otherwise noted.
+ * <p>
+ * Once a connection to a server is established, a session ID is assigned to the
+ * client. The client will send heart beats to the server periodically to keep
+ * the session valid.
+ * <p>
+ * The application can call ZooKeeper APIs through a client as long as the
+ * session ID of the client remains valid.
+ * <p>
+ * If for some reason, the client fails to send heart beats to the server for a
+ * prolonged period of time (exceeding the sessionTimeout value, for instance),
+ * the server will expire the session, and the session ID will become invalid.
+ * The client object will no longer be usable. To make ZooKeeper API calls, the
+ * application must create a new client object.
+ * <p>
+ * If the ZooKeeper server the client currently connects to fails or otherwise
+ * does not respond, the client will automatically try to connect to another
+ * server before its session ID expires. If successful, the application can
+ * continue to use the client.
+ * <p>
+ * The ZooKeeper API methods are either synchronous or asynchronous. Synchronous
+ * methods blocks until the server has responded. Asynchronous methods just queue
+ * the request for sending and return immediately. They take a callback object that
+ * will be executed either on successful execution of the request or on error with
+ * an appropriate return code (rc) indicating the error.
+ * <p>
+ * Some successful ZooKeeper API calls can leave watches on the "data nodes" in
+ * the ZooKeeper server. Other successful ZooKeeper API calls can trigger those
+ * watches. Once a watch is triggered, an event will be delivered to the client
+ * which left the watch at the first place. Each watch can be triggered only
+ * once. Thus, up to one event will be delivered to a client for every watch it
+ * leaves.
+ * <p>
+ * A client needs an object of a class implementing Watcher interface for
+ * processing the events delivered to the client.
+ *
+ * When a client drops the current connection and re-connects to a server, all the
+ * existing watches are considered as being triggered but the undelivered events
+ * are lost. To emulate this, the client will generate a special event to tell
+ * the event handler a connection has been dropped. This special event has
+ * EventType None and KeeperState Disconnected.
+ *
+ */
+/*
+ * We suppress the "try" warning here because the close() method's signature
+ * allows it to throw InterruptedException which is strongly advised against
+ * by AutoCloseable (see: http://docs.oracle.com/javase/7/docs/api/java/lang/AutoCloseable.html#close()).
+ * close() will never throw an InterruptedException but the exception remains in the
+ * signature for backwards compatibility purposes.
+*/
+@SuppressWarnings("try")
+@InterfaceAudience.Public
+public class ZooKeeper implements AutoCloseable {
+
+ /**
+ * @deprecated Use {@link ZKClientConfig#ZOOKEEPER_CLIENT_CNXN_SOCKET}
+ * instead.
+ */
+ @Deprecated
+ public static final String ZOOKEEPER_CLIENT_CNXN_SOCKET = "zookeeper.clientCnxnSocket";
+ // Setting this to "true" will enable encrypted client-server communication.
+
+ /**
+ * @deprecated Use {@link ZKClientConfig#SECURE_CLIENT}
+ * instead.
+ */
+ @Deprecated
+ public static final String SECURE_CLIENT = "zookeeper.client.secure";
+
+ protected final ClientCnxn cnxn;
+ private static final Logger LOG;
+ static {
+ //Keep these two lines together to keep the initialization order explicit
+ LOG = LoggerFactory.getLogger(ZooKeeper.class);
+ Environment.logEnv("Client environment:", LOG);
+ }
+
+ protected final HostProvider hostProvider;
+
+ /**
+ * This function allows a client to update the connection string by providing
+ * a new comma separated list of host:port pairs, each corresponding to a
+ * ZooKeeper server.
+ * <p>
+ * The function invokes a <a href="https://issues.apache.org/jira/browse/ZOOKEEPER-1355">
+ * probabilistic load-balancing algorithm</a> which may cause the client to disconnect from
+ * its current host with the goal to achieve expected uniform number of connections per server
+ * in the new list. In case the current host to which the client is connected is not in the new
+ * list this call will always cause the connection to be dropped. Otherwise, the decision
+ * is based on whether the number of servers has increased or decreased and by how much.
+ * For example, if the previous connection string contained 3 hosts and now the list contains
+ * these 3 hosts and 2 more hosts, 40% of clients connected to each of the 3 hosts will
+ * move to one of the new hosts in order to balance the load. The algorithm will disconnect
+ * from the current host with probability 0.4 and in this case cause the client to connect
+ * to one of the 2 new hosts, chosen at random.
+ * <p>
+ * If the connection is dropped, the client moves to a special mode "reconfigMode" where he chooses
+ * a new server to connect to using the probabilistic algorithm. After finding a server,
+ * or exhausting all servers in the new list after trying all of them and failing to connect,
+ * the client moves back to the normal mode of operation where it will pick an arbitrary server
+ * from the connectString and attempt to connect to it. If establishment of
+ * the connection fails, another server in the connect string will be tried
+ * (the order is non-deterministic, as we random shuffle the list), until a
+ * connection is established. The client will continue attempts until the
+ * session is explicitly closed (or the session is expired by the server).
+
+ * @param connectString
+ * comma separated host:port pairs, each corresponding to a zk
+ * server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002"
+ * If the optional chroot suffix is used the example would look
+ * like: "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002/app/a"
+ * where the client would be rooted at "/app/a" and all paths
+ * would be relative to this root - ie getting/setting/etc...
+ * "/foo/bar" would result in operations being run on
+ * "/app/a/foo/bar" (from the server perspective).
+ *
+ * @throws IOException in cases of network failure
+ */
+ public void updateServerList(String connectString) throws IOException {
+ ConnectStringParser connectStringParser = new ConnectStringParser(connectString);
+ Collection<InetSocketAddress> serverAddresses = connectStringParser.getServerAddresses();
+
+ ClientCnxnSocket clientCnxnSocket = cnxn.sendThread.getClientCnxnSocket();
+ InetSocketAddress currentHost = (InetSocketAddress) clientCnxnSocket.getRemoteSocketAddress();
+
+ boolean reconfigMode = hostProvider.updateServerList(serverAddresses, currentHost);
+
+ // cause disconnection - this will cause next to be called
+ // which will in turn call nextReconfigMode
+ if (reconfigMode) clientCnxnSocket.testableCloseSocket();
+ }
+
+ public ZooKeeperSaslClient getSaslClient() {
+ return cnxn.zooKeeperSaslClient;
+ }
+
+ protected final ZKWatchManager watchManager;
+
+ private final ZKClientConfig clientConfig;
+
+ public ZKClientConfig getClientConfig() {
+ return clientConfig;
+ }
+
+ protected List<String> getDataWatches() {
+ synchronized(watchManager.dataWatches) {
+ List<String> rc = new ArrayList<String>(watchManager.dataWatches.keySet());
+ return rc;
+ }
+ }
+ protected List<String> getExistWatches() {
+ synchronized(watchManager.existWatches) {
+ List<String> rc = new ArrayList<String>(watchManager.existWatches.keySet());
+ return rc;
+ }
+ }
+ protected List<String> getChildWatches() {
+ synchronized(watchManager.childWatches) {
+ List<String> rc = new ArrayList<String>(watchManager.childWatches.keySet());
+ return rc;
+ }
+ }
+
+ /**
+ * Manage watchers & handle events generated by the ClientCnxn object.
+ *
+ * We are implementing this as a nested class of ZooKeeper so that
+ * the public methods will not be exposed as part of the ZooKeeper client
+ * API.
+ */
+ static class ZKWatchManager implements ClientWatchManager {
+ private final Map<String, Set<Watcher>> dataWatches =
+ new HashMap<String, Set<Watcher>>();
+ private final Map<String, Set<Watcher>> existWatches =
+ new HashMap<String, Set<Watcher>>();
+ private final Map<String, Set<Watcher>> childWatches =
+ new HashMap<String, Set<Watcher>>();
+ private boolean disableAutoWatchReset;
+
+ ZKWatchManager(boolean disableAutoWatchReset) {
+ this.disableAutoWatchReset = disableAutoWatchReset;
+ }
+
+ protected volatile Watcher defaultWatcher;
+
+ final private void addTo(Set<Watcher> from, Set<Watcher> to) {
+ if (from != null) {
+ to.addAll(from);
+ }
+ }
+
+ public Map<EventType, Set<Watcher>> removeWatcher(String clientPath,
+ Watcher watcher, WatcherType watcherType, boolean local, int rc)
+ throws KeeperException {
+ // Validate the provided znode path contains the given watcher of
+ // watcherType
+ containsWatcher(clientPath, watcher, watcherType);
+
+ Map<EventType, Set<Watcher>> removedWatchers = new HashMap<EventType, Set<Watcher>>();
+ HashSet<Watcher> childWatchersToRem = new HashSet<Watcher>();
+ removedWatchers
+ .put(EventType.ChildWatchRemoved, childWatchersToRem);
+ HashSet<Watcher> dataWatchersToRem = new HashSet<Watcher>();
+ removedWatchers.put(EventType.DataWatchRemoved, dataWatchersToRem);
+ boolean removedWatcher = false;
+ switch (watcherType) {
+ case Children: {
+ synchronized (childWatches) {
+ removedWatcher = removeWatches(childWatches, watcher,
+ clientPath, local, rc, childWatchersToRem);
+ }
+ break;
+ }
+ case Data: {
+ synchronized (dataWatches) {
+ removedWatcher = removeWatches(dataWatches, watcher,
+ clientPath, local, rc, dataWatchersToRem);
+ }
+
+ synchronized (existWatches) {
+ boolean removedDataWatcher = removeWatches(existWatches,
+ watcher, clientPath, local, rc, dataWatchersToRem);
+ removedWatcher |= removedDataWatcher;
+ }
+ break;
+ }
+ case Any: {
+ synchronized (childWatches) {
+ removedWatcher = removeWatches(childWatches, watcher,
+ clientPath, local, rc, childWatchersToRem);
+ }
+
+ synchronized (dataWatches) {
+ boolean removedDataWatcher = removeWatches(dataWatches,
+ watcher, clientPath, local, rc, dataWatchersToRem);
+ removedWatcher |= removedDataWatcher;
+ }
+ synchronized (existWatches) {
+ boolean removedDataWatcher = removeWatches(existWatches,
+ watcher, clientPath, local, rc, dataWatchersToRem);
+ removedWatcher |= removedDataWatcher;
+ }
+ }
+ }
+ // Watcher function doesn't exists for the specified params
+ if (!removedWatcher) {
+ throw new KeeperException.NoWatcherException(clientPath);
+ }
+ return removedWatchers;
+ }
+
+ private boolean contains(String path, Watcher watcherObj,
+ Map<String, Set<Watcher>> pathVsWatchers) {
+ boolean watcherExists = true;
+ if (pathVsWatchers == null || pathVsWatchers.size() == 0) {
+ watcherExists = false;
+ } else {
+ Set<Watcher> watchers = pathVsWatchers.get(path);
+ if (watchers == null) {
+ watcherExists = false;
+ } else if (watcherObj == null) {
+ watcherExists = watchers.size() > 0;
+ } else {
+ watcherExists = watchers.contains(watcherObj);
+ }
+ }
+ return watcherExists;
+ }
+
+ /**
+ * Validate the provided znode path contains the given watcher and
+ * watcherType
+ *
+ * @param path
+ * - client path
+ * @param watcher
+ * - watcher object reference
+ * @param watcherType
+ * - type of the watcher
+ * @throws NoWatcherException
+ */
+ void containsWatcher(String path, Watcher watcher,
+ WatcherType watcherType) throws NoWatcherException{
+ boolean containsWatcher = false;
+ switch (watcherType) {
+ case Children: {
+ synchronized (childWatches) {
+ containsWatcher = contains(path, watcher, childWatches);
+ }
+ break;
+ }
+ case Data: {
+ synchronized (dataWatches) {
+ containsWatcher = contains(path, watcher, dataWatches);
+ }
+
+ synchronized (existWatches) {
+ boolean contains_temp = contains(path, watcher,
+ existWatches);
+ containsWatcher |= contains_temp;
+ }
+ break;
+ }
+ case Any: {
+ synchronized (childWatches) {
+ containsWatcher = contains(path, watcher, childWatches);
+ }
+
+ synchronized (dataWatches) {
+ boolean contains_temp = contains(path, watcher, dataWatches);
+ containsWatcher |= contains_temp;
+ }
+ synchronized (existWatches) {
+ boolean contains_temp = contains(path, watcher,
+ existWatches);
+ containsWatcher |= contains_temp;
+ }
+ }
+ }
+ // Watcher function doesn't exists for the specified params
+ if (!containsWatcher) {
+ throw new KeeperException.NoWatcherException(path);
+ }
+ }
+
+ protected boolean removeWatches(Map<String, Set<Watcher>> pathVsWatcher,
+ Watcher watcher, String path, boolean local, int rc,
+ Set<Watcher> removedWatchers) throws KeeperException {
+ if (!local && rc != Code.OK.intValue()) {
+ throw KeeperException
+ .create(KeeperException.Code.get(rc), path);
+ }
+ boolean success = false;
+ // When local flag is true, remove watchers for the given path
+ // irrespective of rc. Otherwise shouldn't remove watchers locally
+ // when sees failure from server.
+ if (rc == Code.OK.intValue() || (local && rc != Code.OK.intValue())) {
+ // Remove all the watchers for the given path
+ if (watcher == null) {
+ Set<Watcher> pathWatchers = pathVsWatcher.remove(path);
+ if (pathWatchers != null) {
+ // found path watchers
+ removedWatchers.addAll(pathWatchers);
+ success = true;
+ }
+ } else {
+ Set<Watcher> watchers = pathVsWatcher.get(path);
+ if (watchers != null) {
+ if (watchers.remove(watcher)) {
+ // found path watcher
+ removedWatchers.add(watcher);
+ // cleanup <path vs watchlist>
+ if (watchers.size() <= 0) {
+ pathVsWatcher.remove(path);
+ }
+ success = true;
+ }
+ }
+ }
+ }
+ return success;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.zookeeper.ClientWatchManager#materialize(Event.KeeperState,
+ * Event.EventType, java.lang.String)
+ */
+ @Override
+ public Set<Watcher> materialize(Watcher.Event.KeeperState state,
+ Watcher.Event.EventType type,
+ String clientPath)
+ {
+ Set<Watcher> result = new HashSet<Watcher>();
+
+ switch (type) {
+ case None:
+ result.add(defaultWatcher);
+ boolean clear = disableAutoWatchReset && state != Watcher.Event.KeeperState.SyncConnected;
+ synchronized(dataWatches) {
+ for(Set<Watcher> ws: dataWatches.values()) {
+ result.addAll(ws);
+ }
+ if (clear) {
+ dataWatches.clear();
+ }
+ }
+
+ synchronized(existWatches) {
+ for(Set<Watcher> ws: existWatches.values()) {
+ result.addAll(ws);
+ }
+ if (clear) {
+ existWatches.clear();
+ }
+ }
+
+ synchronized(childWatches) {
+ for(Set<Watcher> ws: childWatches.values()) {
+ result.addAll(ws);
+ }
+ if (clear) {
+ childWatches.clear();
+ }
+ }
+
+ return result;
+ case NodeDataChanged:
+ case NodeCreated:
+ synchronized (dataWatches) {
+ addTo(dataWatches.remove(clientPath), result);
+ }
+ synchronized (existWatches) {
+ addTo(existWatches.remove(clientPath), result);
+ }
+ break;
+ case NodeChildrenChanged:
+ synchronized (childWatches) {
+ addTo(childWatches.remove(clientPath), result);
+ }
+ break;
+ case NodeDeleted:
+ synchronized (dataWatches) {
+ addTo(dataWatches.remove(clientPath), result);
+ }
+ // XXX This shouldn't be needed, but just in case
+ synchronized (existWatches) {
+ Set<Watcher> list = existWatches.remove(clientPath);
+ if (list != null) {
+ addTo(list, result);
+ LOG.warn("We are triggering an exists watch for delete! Shouldn't happen!");
+ }
+ }
+ synchronized (childWatches) {
+ addTo(childWatches.remove(clientPath), result);
+ }
+ break;
+ default:
+ String msg = "Unhandled watch event type " + type
+ + " with state " + state + " on path " + clientPath;
+ LOG.error(msg);
+ throw new RuntimeException(msg);
+ }
+
+ return result;
+ }
+ }
+
+ /**
+ * Register a watcher for a particular path.
+ */
+ public abstract class WatchRegistration {
+ private Watcher watcher;
+ private String clientPath;
+ public WatchRegistration(Watcher watcher, String clientPath)
+ {
+ this.watcher = watcher;
+ this.clientPath = clientPath;
+ }
+
+ abstract protected Map<String, Set<Watcher>> getWatches(int rc);
+
+ /**
+ * Register the watcher with the set of watches on path.
+ * @param rc the result code of the operation that attempted to
+ * add the watch on the path.
+ */
+ public void register(int rc) {
+ if (shouldAddWatch(rc)) {
+ Map<String, Set<Watcher>> watches = getWatches(rc);
+ synchronized(watches) {
+ Set<Watcher> watchers = watches.get(clientPath);
+ if (watchers == null) {
+ watchers = new HashSet<Watcher>();
+ watches.put(clientPath, watchers);
+ }
+ watchers.add(watcher);
+ }
+ }
+ }
+ /**
+ * Determine whether the watch should be added based on return code.
+ * @param rc the result code of the operation that attempted to add the
+ * watch on the node
+ * @return true if the watch should be added, otw false
+ */
+ protected boolean shouldAddWatch(int rc) {
+ return rc == 0;
+ }
+ }
+
+ /** Handle the special case of exists watches - they add a watcher
+ * even in the case where NONODE result code is returned.
+ */
+ class ExistsWatchRegistration extends WatchRegistration {
+ public ExistsWatchRegistration(Watcher watcher, String clientPath) {
+ super(watcher, clientPath);
+ }
+
+ @Override
+ protected Map<String, Set<Watcher>> getWatches(int rc) {
+ return rc == 0 ? watchManager.dataWatches : watchManager.existWatches;
+ }
+
+ @Override
+ protected boolean shouldAddWatch(int rc) {
+ return rc == 0 || rc == KeeperException.Code.NONODE.intValue();
+ }
+ }
+
+ class DataWatchRegistration extends WatchRegistration {
+ public DataWatchRegistration(Watcher watcher, String clientPath) {
+ super(watcher, clientPath);
+ }
+
+ @Override
+ protected Map<String, Set<Watcher>> getWatches(int rc) {
+ return watchManager.dataWatches;
+ }
+ }
+
+ class ChildWatchRegistration extends WatchRegistration {
+ public ChildWatchRegistration(Watcher watcher, String clientPath) {
+ super(watcher, clientPath);
+ }
+
+ @Override
+ protected Map<String, Set<Watcher>> getWatches(int rc) {
+ return watchManager.childWatches;
+ }
+ }
+
+ @InterfaceAudience.Public
+ public enum States {
+ CONNECTING, ASSOCIATING, CONNECTED, CONNECTEDREADONLY,
+ CLOSED, AUTH_FAILED, NOT_CONNECTED;
+
+ public boolean isAlive() {
+ return this != CLOSED && this != AUTH_FAILED;
+ }
+
+ /**
+ * Returns whether we are connected to a server (which
+ * could possibly be read-only, if this client is allowed
+ * to go to read-only mode)
+ * */
+ public boolean isConnected() {
+ return this == CONNECTED || this == CONNECTEDREADONLY;
+ }
+ }
+
+ /**
+ * To create a ZooKeeper client object, the application needs to pass a
+ * connection string containing a comma separated list of host:port pairs,
+ * each corresponding to a ZooKeeper server.
+ * <p>
+ * Session establishment is asynchronous. This constructor will initiate
+ * connection to the server and return immediately - potentially (usually)
+ * before the session is fully established. The watcher argument specifies
+ * the watcher that will be notified of any changes in state. This
+ * notification can come at any point before or after the constructor call
+ * has returned.
+ * <p>
+ * The instantiated ZooKeeper client object will pick an arbitrary server
+ * from the connectString and attempt to connect to it. If establishment of
+ * the connection fails, another server in the connect string will be tried
+ * (the order is non-deterministic, as we random shuffle the list), until a
+ * connection is established. The client will continue attempts until the
+ * session is explicitly closed.
+ * <p>
+ * Added in 3.2.0: An optional "chroot" suffix may also be appended to the
+ * connection string. This will run the client commands while interpreting
+ * all paths relative to this root (similar to the unix chroot command).
+ *
+ * @param connectString
+ * comma separated host:port pairs, each corresponding to a zk
+ * server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002" If
+ * the optional chroot suffix is used the example would look
+ * like: "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002/app/a"
+ * where the client would be rooted at "/app/a" and all paths
+ * would be relative to this root - ie getting/setting/etc...
+ * "/foo/bar" would result in operations being run on
+ * "/app/a/foo/bar" (from the server perspective).
+ * @param sessionTimeout
+ * session timeout in milliseconds
+ * @param watcher
+ * a watcher object which will be notified of state changes, may
+ * also be notified for node events
+ *
+ * @throws IOException
+ * in cases of network failure
+ * @throws IllegalArgumentException
+ * if an invalid chroot path is specified
+ */
+ public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher)
+ throws IOException
+ {
+ this(connectString, sessionTimeout, watcher, false);
+ }
+
+ /**
+ * To create a ZooKeeper client object, the application needs to pass a
+ * connection string containing a comma separated list of host:port pairs,
+ * each corresponding to a ZooKeeper server.
+ * <p>
+ * Session establishment is asynchronous. This constructor will initiate
+ * connection to the server and return immediately - potentially (usually)
+ * before the session is fully established. The watcher argument specifies
+ * the watcher that will be notified of any changes in state. This
+ * notification can come at any point before or after the constructor call
+ * has returned.
+ * <p>
+ * The instantiated ZooKeeper client object will pick an arbitrary server
+ * from the connectString and attempt to connect to it. If establishment of
+ * the connection fails, another server in the connect string will be tried
+ * (the order is non-deterministic, as we random shuffle the list), until a
+ * connection is established. The client will continue attempts until the
+ * session is explicitly closed.
+ * <p>
+ * Added in 3.2.0: An optional "chroot" suffix may also be appended to the
+ * connection string. This will run the client commands while interpreting
+ * all paths relative to this root (similar to the unix chroot command).
+ *
+ * @param connectString
+ * comma separated host:port pairs, each corresponding to a zk
+ * server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002" If
+ * the optional chroot suffix is used the example would look
+ * like: "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002/app/a"
+ * where the client would be rooted at "/app/a" and all paths
+ * would be relative to this root - ie getting/setting/etc...
+ * "/foo/bar" would result in operations being run on
+ * "/app/a/foo/bar" (from the server perspective).
+ * @param sessionTimeout
+ * session timeout in milliseconds
+ * @param watcher
+ * a watcher object which will be notified of state changes, may
+ * also be notified for node events
+ * @param conf
+ * (added in 3.5.2) passing this conf object gives each client the flexibility of
+ * configuring properties differently compared to other instances
+ * @throws IOException
+ * in cases of network failure
+ * @throws IllegalArgumentException
+ * if an invalid chroot path is specified
+ */
+ public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher,
+ ZKClientConfig conf) throws IOException {
+ this(connectString, sessionTimeout, watcher, false, conf);
+ }
+
+ /**
+ * To create a ZooKeeper client object, the application needs to pass a
+ * connection string containing a comma separated list of host:port pairs,
+ * each corresponding to a ZooKeeper server.
+ * <p>
+ * Session establishment is asynchronous. This constructor will initiate
+ * connection to the server and return immediately - potentially (usually)
+ * before the session is fully established. The watcher argument specifies
+ * the watcher that will be notified of any changes in state. This
+ * notification can come at any point before or after the constructor call
+ * has returned.
+ * <p>
+ * The instantiated ZooKeeper client object will pick an arbitrary server
+ * from the connectString and attempt to connect to it. If establishment of
+ * the connection fails, another server in the connect string will be tried
+ * (the order is non-deterministic, as we random shuffle the list), until a
+ * connection is established. The client will continue attempts until the
+ * session is explicitly closed.
+ * <p>
+ * Added in 3.2.0: An optional "chroot" suffix may also be appended to the
+ * connection string. This will run the client commands while interpreting
+ * all paths relative to this root (similar to the unix chroot command).
+ * <p>
+ * For backward compatibility, there is another version
+ * {@link #ZooKeeper(String, int, Watcher, boolean)} which uses
+ * default {@link StaticHostProvider}
+ *
+ * @param connectString
+ * comma separated host:port pairs, each corresponding to a zk
+ * server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002" If
+ * the optional chroot suffix is used the example would look
+ * like: "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002/app/a"
+ * where the client would be rooted at "/app/a" and all paths
+ * would be relative to this root - ie getting/setting/etc...
+ * "/foo/bar" would result in operations being run on
+ * "/app/a/foo/bar" (from the server perspective).
+ * @param sessionTimeout
+ * session timeout in milliseconds
+ * @param watcher
+ * a watcher object which will be notified of state changes, may
+ * also be notified for node events
+ * @param canBeReadOnly
+ * (added in 3.4) whether the created client is allowed to go to
+ * read-only mode in case of partitioning. Read-only mode
+ * basically means that if the client can't find any majority
+ * servers but there's partitioned server it could reach, it
+ * connects to one in read-only mode, i.e. read requests are
+ * allowed while write requests are not. It continues seeking for
+ * majority in the background.
+ * @param aHostProvider
+ * use this as HostProvider to enable custom behaviour.
+ *
+ * @throws IOException
+ * in cases of network failure
+ * @throws IllegalArgumentException
+ * if an invalid chroot path is specified
+ */
+ public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher,
+ boolean canBeReadOnly, HostProvider aHostProvider)
+ throws IOException {
+ this(connectString, sessionTimeout, watcher, canBeReadOnly,
+ aHostProvider, null);
+ }
+
+
+ /**
+ * To create a ZooKeeper client object, the application needs to pass a
+ * connection string containing a comma separated list of host:port pairs,
+ * each corresponding to a ZooKeeper server.
+ * <p>
+ * Session establishment is asynchronous. This constructor will initiate
+ * connection to the server and return immediately - potentially (usually)
+ * before the session is fully established. The watcher argument specifies
+ * the watcher that will be notified of any changes in state. This
+ * notification can come at any point before or after the constructor call
+ * has returned.
+ * <p>
+ * The instantiated ZooKeeper client object will pick an arbitrary server
+ * from the connectString and attempt to connect to it. If establishment of
+ * the connection fails, another server in the connect string will be tried
+ * (the order is non-deterministic, as we random shuffle the list), until a
+ * connection is established. The client will continue attempts until the
+ * session is explicitly closed.
+ * <p>
+ * Added in 3.2.0: An optional "chroot" suffix may also be appended to the
+ * connection string. This will run the client commands while interpreting
+ * all paths relative to this root (similar to the unix chroot command).
+ * <p>
+ * For backward compatibility, there is another version
+ * {@link #ZooKeeper(String, int, Watcher, boolean)} which uses default
+ * {@link StaticHostProvider}
+ *
+ * @param connectString
+ * comma separated host:port pairs, each corresponding to a zk
+ * server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002" If
+ * the optional chroot suffix is used the example would look
+ * like: "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002/app/a"
+ * where the client would be rooted at "/app/a" and all paths
+ * would be relative to this root - ie getting/setting/etc...
+ * "/foo/bar" would result in operations being run on
+ * "/app/a/foo/bar" (from the server perspective).
+ * @param sessionTimeout
+ * session timeout in milliseconds
+ * @param watcher
+ * a watcher object which will be notified of state changes, may
+ * also be notified for node events
+ * @param canBeReadOnly
+ * (added in 3.4) whether the created client is allowed to go to
+ * read-only mode in case of partitioning. Read-only mode
+ * basically means that if the client can't find any majority
+ * servers but there's partitioned server it could reach, it
+ * connects to one in read-only mode, i.e. read requests are
+ * allowed while write requests are not. It continues seeking for
+ * majority in the background.
+ * @param aHostProvider
+ * use this as HostProvider to enable custom behaviour.
+ * @param clientConfig
+ * (added in 3.5.2) passing this conf object gives each client the flexibility of
+ * configuring properties differently compared to other instances
+ * @throws IOException
+ * in cases of network failure
+ * @throws IllegalArgumentException
+ * if an invalid chroot path is specified
+ */
+ public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher,
+ boolean canBeReadOnly, HostProvider aHostProvider,
+ ZKClientConfig clientConfig) throws IOException {
+ LOG.info("Initiating client connection, connectString=" + connectString
+ + " sessionTimeout=" + sessionTimeout + " watcher=" + watcher);
+
+ if (clientConfig == null) {
+ clientConfig = new ZKClientConfig();
+ }
+ this.clientConfig = clientConfig;
+ watchManager = defaultWatchManager();
+ watchManager.defaultWatcher = watcher;
+ ConnectStringParser connectStringParser = new ConnectStringParser(
+ connectString);
+ hostProvider = aHostProvider;
+
+ cnxn = createConnection(connectStringParser.getChrootPath(),
+ hostProvider, sessionTimeout, this, watchManager,
+ getClientCnxnSocket(), canBeReadOnly);
+ cnxn.start();
+ }
+
+ // @VisibleForTesting
+ protected ClientCnxn createConnection(String chrootPath,
+ HostProvider hostProvider, int sessionTimeout, ZooKeeper zooKeeper,
+ ClientWatchManager watcher, ClientCnxnSocket clientCnxnSocket,
+ boolean canBeReadOnly) throws IOException {
+ return new ClientCnxn(chrootPath, hostProvider, sessionTimeout, this,
+ watchManager, clientCnxnSocket, canBeReadOnly);
+ }
+
+ /**
+ * To create a ZooKeeper client object, the application needs to pass a
+ * connection string containing a comma separated list of host:port pairs,
+ * each corresponding to a ZooKeeper server.
+ * <p>
+ * Session establishment is asynchronous. This constructor will initiate
+ * connection to the server and return immediately - potentially (usually)
+ * before the session is fully established. The watcher argument specifies
+ * the watcher that will be notified of any changes in state. This
+ * notification can come at any point before or after the constructor call
+ * has returned.
+ * <p>
+ * The instantiated ZooKeeper client object will pick an arbitrary server
+ * from the connectString and attempt to connect to it. If establishment of
+ * the connection fails, another server in the connect string will be tried
+ * (the order is non-deterministic, as we random shuffle the list), until a
+ * connection is established. The client will continue attempts until the
+ * session is explicitly closed.
+ * <p>
+ * Added in 3.2.0: An optional "chroot" suffix may also be appended to the
+ * connection string. This will run the client commands while interpreting
+ * all paths relative to this root (similar to the unix chroot command).
+ * <p>
+ *
+ * @param connectString
+ * comma separated host:port pairs, each corresponding to a zk
+ * server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002" If
+ * the optional chroot suffix is used the example would look
+ * like: "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002/app/a"
+ * where the client would be rooted at "/app/a" and all paths
+ * would be relative to this root - ie getting/setting/etc...
+ * "/foo/bar" would result in operations being run on
+ * "/app/a/foo/bar" (from the server perspective).
+ * @param sessionTimeout
+ * session timeout in milliseconds
+ * @param watcher
+ * a watcher object which will be notified of state changes, may
+ * also be notified for node events
+ * @param canBeReadOnly
+ * (added in 3.4) whether the created client is allowed to go to
+ * read-only mode in case of partitioning. Read-only mode
+ * basically means that if the client can't find any majority
+ * servers but there's partitioned server it could reach, it
+ * connects to one in read-only mode, i.e. read requests are
+ * allowed while write requests are not. It continues seeking for
+ * majority in the background.
+ *
+ * @throws IOException
+ * in cases of network failure
+ * @throws IllegalArgumentException
+ * if an invalid chroot path is specified
+ */
+ public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher,
+ boolean canBeReadOnly) throws IOException {
+ this(connectString, sessionTimeout, watcher, canBeReadOnly,
+ createDefaultHostProvider(connectString));
+ }
+
+ /**
+ * To create a ZooKeeper client object, the application needs to pass a
+ * connection string containing a comma separated list of host:port pairs,
+ * each corresponding to a ZooKeeper server.
+ * <p>
+ * Session establishment is asynchronous. This constructor will initiate
+ * connection to the server and return immediately - potentially (usually)
+ * before the session is fully established. The watcher argument specifies
+ * the watcher that will be notified of any changes in state. This
+ * notification can come at any point before or after the constructor call
+ * has returned.
+ * <p>
+ * The instantiated ZooKeeper client object will pick an arbitrary server
+ * from the connectString and attempt to connect to it. If establishment of
+ * the connection fails, another server in the connect string will be tried
+ * (the order is non-deterministic, as we random shuffle the list), until a
+ * connection is established. The client will continue attempts until the
+ * session is explicitly closed.
+ * <p>
+ * Added in 3.2.0: An optional "chroot" suffix may also be appended to the
+ * connection string. This will run the client commands while interpreting
+ * all paths relative to this root (similar to the unix chroot command).
+ * <p>
+ *
+ * @param connectString
+ * comma separated host:port pairs, each corresponding to a zk
+ * server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002" If
+ * the optional chroot suffix is used the example would look
+ * like: "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002/app/a"
+ * where the client would be rooted at "/app/a" and all paths
+ * would be relative to this root - ie getting/setting/etc...
+ * "/foo/bar" would result in operations being run on
+ * "/app/a/foo/bar" (from the server perspective).
+ * @param sessionTimeout
+ * session timeout in milliseconds
+ * @param watcher
+ * a watcher object which will be notified of state changes, may
+ * also be notified for node events
+ * @param canBeReadOnly
+ * (added in 3.4) whether the created client is allowed to go to
+ * read-only mode in case of partitioning. Read-only mode
+ * basically means that if the client can't find any majority
+ * servers but there's partitioned server it could reach, it
+ * connects to one in read-only mode, i.e. read requests are
+ * allowed while write requests are not. It continues seeking for
+ * majority in the background.
+ * @param conf
+ * (added in 3.5.2) passing this conf object gives each client the flexibility of
+ * configuring properties differently compared to other instances
+ * @throws IOException
+ * in cases of network failure
+ * @throws IllegalArgumentException
+ * if an invalid chroot path is specified
+ */
+ public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher,
+ boolean canBeReadOnly, ZKClientConfig conf) throws IOException {
+ this(connectString, sessionTimeout, watcher, canBeReadOnly,
+ createDefaultHostProvider(connectString), conf);
+ }
+
+ /**
+ * To create a ZooKeeper client object, the application needs to pass a
+ * connection string containing a comma separated list of host:port pairs,
+ * each corresponding to a ZooKeeper server.
+ * <p>
+ * Session establishment is asynchronous. This constructor will initiate
+ * connection to the server and return immediately - potentially (usually)
+ * before the session is fully established. The watcher argument specifies
+ * the watcher that will be notified of any changes in state. This
+ * notification can come at any point before or after the constructor call
+ * has returned.
+ * <p>
+ * The instantiated ZooKeeper client object will pick an arbitrary server
+ * from the connectString and attempt to connect to it. If establishment of
+ * the connection fails, another server in the connect string will be tried
+ * (the order is non-deterministic, as we random shuffle the list), until a
+ * connection is established. The client will continue attempts until the
+ * session is explicitly closed (or the session is expired by the server).
+ * <p>
+ * Added in 3.2.0: An optional "chroot" suffix may also be appended to the
+ * connection string. This will run the client commands while interpreting
+ * all paths relative to this root (similar to the unix chroot command).
+ * <p>
+ * Use {@link #getSessionId} and {@link #getSessionPasswd} on an established
+ * client connection, these values must be passed as sessionId and
+ * sessionPasswd respectively if reconnecting. Otherwise, if not
+ * reconnecting, use the other constructor which does not require these
+ * parameters.
+ *
+ * @param connectString
+ * comma separated host:port pairs, each corresponding to a zk
+ * server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002"
+ * If the optional chroot suffix is used the example would look
+ * like: "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002/app/a"
+ * where the client would be rooted at "/app/a" and all paths
+ * would be relative to this root - ie getting/setting/etc...
+ * "/foo/bar" would result in operations being run on
+ * "/app/a/foo/bar" (from the server perspective).
+ * @param sessionTimeout
+ * session timeout in milliseconds
+ * @param watcher
+ * a watcher object which will be notified of state changes, may
+ * also be notified for node events
+ * @param sessionId
+ * specific session id to use if reconnecting
+ * @param sessionPasswd
+ * password for this session
+ *
+ * @throws IOException in cases of network failure
+ * @throws IllegalArgumentException if an invalid chroot path is specified
+ * @throws IllegalArgumentException for an invalid list of ZooKeeper hosts
+ */
+ public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher,
+ long sessionId, byte[] sessionPasswd)
+ throws IOException
+ {
+ this(connectString, sessionTimeout, watcher, sessionId, sessionPasswd, false);
+ }
+
+ /**
+ * To create a ZooKeeper client object, the application needs to pass a
+ * connection string containing a comma separated list of host:port pairs,
+ * each corresponding to a ZooKeeper server.
+ * <p>
+ * Session establishment is asynchronous. This constructor will initiate
+ * connection to the server and return immediately - potentially (usually)
+ * before the session is fully established. The watcher argument specifies
+ * the watcher that will be notified of any changes in state. This
+ * notification can come at any point before or after the constructor call
+ * has returned.
+ * <p>
+ * The instantiated ZooKeeper client object will pick an arbitrary server
+ * from the connectString and attempt to connect to it. If establishment of
+ * the connection fails, another server in the connect string will be tried
+ * (the order is non-deterministic, as we random shuffle the list), until a
+ * connection is established. The client will continue attempts until the
+ * session is explicitly closed (or the session is expired by the server).
+ * <p>
+ * Added in 3.2.0: An optional "chroot" suffix may also be appended to the
+ * connection string. This will run the client commands while interpreting
+ * all paths relative to this root (similar to the unix chroot command).
+ * <p>
+ * Use {@link #getSessionId} and {@link #getSessionPasswd} on an established
+ * client connection, these values must be passed as sessionId and
+ * sessionPasswd respectively if reconnecting. Otherwise, if not
+ * reconnecting, use the other constructor which does not require these
+ * parameters.
+ * <p>
+ * For backward compatibility, there is another version
+ * {@link #ZooKeeper(String, int, Watcher, long, byte[], boolean)} which uses
+ * default {@link StaticHostProvider}
+ *
+ * @param connectString
+ * comma separated host:port pairs, each corresponding to a zk
+ * server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002"
+ * If the optional chroot suffix is used the example would look
+ * like: "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002/app/a"
+ * where the client would be rooted at "/app/a" and all paths
+ * would be relative to this root - ie getting/setting/etc...
+ * "/foo/bar" would result in operations being run on
+ * "/app/a/foo/bar" (from the server perspective).
+ * @param sessionTimeout
+ * session timeout in milliseconds
+ * @param watcher
+ * a watcher object which will be notified of state changes, may
+ * also be notified for node events
+ * @param sessionId
+ * specific session id to use if reconnecting
+ * @param sessionPasswd
+ * password for this session
+ * @param canBeReadOnly
+ * (added in 3.4) whether the created client is allowed to go to
+ * read-only mode in case of partitioning. Read-only mode
+ * basically means that if the client can't find any majority
+ * servers but there's partitioned server it could reach, it
+ * connects to one in read-only mode, i.e. read requests are
+ * allowed while write requests are not. It continues seeking for
+ * majority in the background.
+ * @param aHostProvider
+ * use this as HostProvider to enable custom behaviour.
+ * @throws IOException in cases of network failure
+ * @throws IllegalArgumentException if an invalid chroot path is specified
+ */
+ public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher,
+ long sessionId, byte[] sessionPasswd, boolean canBeReadOnly,
+ HostProvider aHostProvider) throws IOException {
+ this(connectString, sessionTimeout, watcher, sessionId, sessionPasswd,
+ canBeReadOnly, aHostProvider, null);
+ }
+
+ /**
+ * To create a ZooKeeper client object, the application needs to pass a
+ * connection string containing a comma separated list of host:port pairs,
+ * each corresponding to a ZooKeeper server.
+ * <p>
+ * Session establishment is asynchronous. This constructor will initiate
+ * connection to the server and return immediately - potentially (usually)
+ * before the session is fully established. The watcher argument specifies
+ * the watcher that will be notified of any changes in state. This
+ * notification can come at any point before or after the constructor call
+ * has returned.
+ * <p>
+ * The instantiated ZooKeeper client object will pick an arbitrary server
+ * from the connectString and attempt to connect to it. If establishment of
+ * the connection fails, another server in the connect string will be tried
+ * (the order is non-deterministic, as we random shuffle the list), until a
+ * connection is established. The client will continue attempts until the
+ * session is explicitly closed (or the session is expired by the server).
+ * <p>
+ * Added in 3.2.0: An optional "chroot" suffix may also be appended to the
+ * connection string. This will run the client commands while interpreting
+ * all paths relative to this root (similar to the unix chroot command).
+ * <p>
+ * Use {@link #getSessionId} and {@link #getSessionPasswd} on an established
+ * client connection, these values must be passed as sessionId and
+ * sessionPasswd respectively if reconnecting. Otherwise, if not
+ * reconnecting, use the other constructor which does not require these
+ * parameters.
+ * <p>
+ * For backward compatibility, there is another version
+ * {@link #ZooKeeper(String, int, Watcher, long, byte[], boolean)} which uses
+ * default {@link StaticHostProvider}
+ *
+ * @param connectString
+ * comma separated host:port pairs, each corresponding to a zk
+ * server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002"
+ * If the optional chroot suffix is used the example would look
+ * like: "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002/app/a"
+ * where the client would be rooted at "/app/a" and all paths
+ * would be relative to this root - ie getting/setting/etc...
+ * "/foo/bar" would result in operations being run on
+ * "/app/a/foo/bar" (from the server perspective).
+ * @param sessionTimeout
+ * session timeout in milliseconds
+ * @param watcher
+ * a watcher object which will be notified of state changes, may
+ * also be notified for node events
+ * @param sessionId
+ * specific session id to use if reconnecting
+ * @param sessionPasswd
+ * password for this session
+ * @param canBeReadOnly
+ * (added in 3.4) whether the created client is allowed to go to
+ * read-only mode in case of partitioning. Read-only mode
+ * basically means that if the client can't find any majority
+ * servers but there's partitioned server it could reach, it
+ * connects to one in read-only mode, i.e. read requests are
+ * allowed while write requests are not. It continues seeking for
+ * majority in the background.
+ * @param aHostProvider
+ * use this as HostProvider to enable custom behaviour.
+ * @param clientConfig
+ * (added in 3.5.2) passing this conf object gives each client the flexibility of
+ * configuring properties differently compared to other instances
+ * @throws IOException in cases of network failure
+ * @throws IllegalArgumentException if an invalid chroot path is specified
+ *
+ * @since 3.5.5
+ */
+ public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher,
+ long sessionId, byte[] sessionPasswd, boolean canBeReadOnly,
+ HostProvider aHostProvider, ZKClientConfig clientConfig) throws IOException {
+ LOG.info("Initiating client connection, connectString=" + connectString
+ + " sessionTimeout=" + sessionTimeout
+ + " watcher=" + watcher
+ + " sessionId=" + Long.toHexString(sessionId)
+ + " sessionPasswd="
+ + (sessionPasswd == null ? "<null>" : "<hidden>"));
+
+ if (clientConfig == null) {
+ clientConfig = new ZKClientConfig();
+ }
+ this.clientConfig = clientConfig;
+ watchManager = defaultWatchManager();
+ watchManager.defaultWatcher = watcher;
+
+ ConnectStringParser connectStringParser = new ConnectStringParser(
+ connectString);
+ hostProvider = aHostProvider;
+
+ cnxn = new ClientCnxn(connectStringParser.getChrootPath(),
+ hostProvider, sessionTimeout, this, watchManager,
+ getClientCnxnSocket(), sessionId, sessionPasswd, canBeReadOnly);
+ cnxn.seenRwServerBefore = true; // since user has provided sessionId
+ cnxn.start();
+ }
+
+ /**
+ * To create a ZooKeeper client object, the application needs to pass a
+ * connection string containing a comma separated list of host:port pairs,
+ * each corresponding to a ZooKeeper server.
+ * <p>
+ * Session establishment is asynchronous. This constructor will initiate
+ * connection to the server and return immediately - potentially (usually)
+ * before the session is fully established. The watcher argument specifies
+ * the watcher that will be notified of any changes in state. This
+ * notification can come at any point before or after the constructor call
+ * has returned.
+ * <p>
+ * The instantiated ZooKeeper client object will pick an arbitrary server
+ * from the connectString and attempt to connect to it. If establishment of
+ * the connection fails, another server in the connect string will be tried
+ * (the order is non-deterministic, as we random shuffle the list), until a
+ * connection is established. The client will continue attempts until the
+ * session is explicitly closed (or the session is expired by the server).
+ * <p>
+ * Added in 3.2.0: An optional "chroot" suffix may also be appended to the
+ * connection string. This will run the client commands while interpreting
+ * all paths relative to this root (similar to the unix chroot command).
+ * <p>
+ * Use {@link #getSessionId} and {@link #getSessionPasswd} on an established
+ * client connection, these values must be passed as sessionId and
+ * sessionPasswd respectively if reconnecting. Otherwise, if not
+ * reconnecting, use the other constructor which does not require these
+ * parameters.
+ * <p>
+ * This constructor uses a StaticHostProvider; there is another one
+ * to enable custom behaviour.
+ *
+ * @param connectString
+ * comma separated host:port pairs, each corresponding to a zk
+ * server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002"
+ * If the optional chroot suffix is used the example would look
+ * like: "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002/app/a"
+ * where the client would be rooted at "/app/a" and all paths
+ * would be relative to this root - ie getting/setting/etc...
+ * "/foo/bar" would result in operations being run on
+ * "/app/a/foo/bar" (from the server perspective).
+ * @param sessionTimeout
+ * session timeout in milliseconds
+ * @param watcher
+ * a watcher object which will be notified of state changes, may
+ * also be notified for node events
+ * @param sessionId
+ * specific session id to use if reconnecting
+ * @param sessionPasswd
+ * password for this session
+ * @param canBeReadOnly
+ * (added in 3.4) whether the created client is allowed to go to
+ * read-only mode in case of partitioning. Read-only mode
+ * basically means that if the client can't find any majority
+ * servers but there's partitioned server it could reach, it
+ * connects to one in read-only mode, i.e. read requests are
+ * allowed while write requests are not. It continues seeking for
+ * majority in the background.
+ * @throws IOException in cases of network failure
+ * @throws IllegalArgumentException if an invalid chroot path is specified
+ */
+ public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher,
+ long sessionId, byte[] sessionPasswd, boolean canBeReadOnly)
+ throws IOException {
+ this(connectString, sessionTimeout, watcher, sessionId, sessionPasswd,
+ canBeReadOnly, createDefaultHostProvider(connectString));
+ }
+
+ // default hostprovider
+ private static HostProvider createDefaultHostProvider(String connectString) {
+ return new StaticHostProvider(
+ new ConnectStringParser(connectString).getServerAddresses());
+ }
+
+ // VisibleForTesting
+ public Testable getTestable() {
+ return new ZooKeeperTestable(this, cnxn);
+ }
+
+ /* Useful for testing watch handling behavior */
+ protected ZKWatchManager defaultWatchManager() {
+ return new ZKWatchManager(getClientConfig().getBoolean(ZKClientConfig.DISABLE_AUTO_WATCH_RESET));
+ }
+
+ /**
+ * The session id for this ZooKeeper client instance. The value returned is
+ * not valid until the client connects to a server and may change after a
+ * re-connect.
+ *
+ * This method is NOT thread safe
+ *
+ * @return current session id
+ */
+ public long getSessionId() {
+ return cnxn.getSessionId();
+ }
+
+ /**
+ * The session password for this ZooKeeper client instance. The value
+ * returned is not valid until the client connects to a server and may
+ * change after a re-connect.
+ *
+ * This method is NOT thread safe
+ *
+ * @return current session password
+ */
+ public byte[] getSessionPasswd() {
+ return cnxn.getSessionPasswd();
+ }
+
+ /**
+ * The negotiated session timeout for this ZooKeeper client instance. The
+ * value returned is not valid until the client connects to a server and
+ * may change after a re-connect.
+ *
+ * This method is NOT thread safe
+ *
+ * @return current session timeout
+ */
+ public int getSessionTimeout() {
+ return cnxn.getSessionTimeout();
+ }
+
+ /**
+ * Add the specified scheme:auth information to this connection.
+ *
+ * This method is NOT thread safe
+ *
+ * @param scheme
+ * @param auth
+ */
+ public void addAuthInfo(String scheme, byte auth[]) {
+ cnxn.addAuthInfo(scheme, auth);
+ }
+
+ /**
+ * Specify the default watcher for the connection (overrides the one
+ * specified during construction).
+ *
+ * @param watcher
+ */
+ public synchronized void register(Watcher watcher) {
+ watchManager.defaultWatcher = watcher;
+ }
+
+ /**
+ * Close this client object. Once the client is closed, its session becomes
+ * invalid. All the ephemeral nodes in the ZooKeeper server associated with
+ * the session will be removed. The watches left on those nodes (and on
+ * their parents) will be triggered.
+ * <p>
+ * Added in 3.5.3: <a href="https://docs.oracle.com/javase/tutorial/essential/exceptions/tryResourceClose.html">try-with-resources</a>
+ * may be used instead of calling close directly.
+ * </p>
+ * <p>
+ * This method does not wait for all internal threads to exit.
+ * Use the {@link #close(int) } method to wait for all resources to be released
+ * </p>
+ *
+ * @throws InterruptedException
+ */
+ public synchronized void close() throws InterruptedException {
+ if (!cnxn.getState().isAlive()) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Close called on already closed client");
+ }
+ return;
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Closing session: 0x" + Long.toHexString(getSessionId()));
+ }
+
+ try {
+ cnxn.close();
+ } catch (IOException e) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Ignoring unexpected exception during close", e);
+ }
+ }
+
+ LOG.info("Session: 0x" + Long.toHexString(getSessionId()) + " closed");
+ }
+
+ /**
+ * Close this client object as the {@link #close() } method.
+ * This method will wait for internal resources to be released.
+ *
+ * @param waitForShutdownTimeoutMs timeout (in milliseconds) to wait for resources to be released.
+ * Use zero or a negative value to skip the wait
+ * @throws InterruptedException
+ * @return true if waitForShutdownTimeout is greater than zero and all of the resources have been released
+ *
+ * @since 3.5.4
+ */
+ public boolean close(int waitForShutdownTimeoutMs) throws InterruptedException {
+ close();
+ return testableWaitForShutdown(waitForShutdownTimeoutMs);
+ }
+
+ /**
+ * Prepend the chroot to the client path (if present). The expectation of
+ * this function is that the client path has been validated before this
+ * function is called
+ * @param clientPath path to the node
+ * @return server view of the path (chroot prepended to client path)
+ */
+ private String prependChroot(String clientPath) {
+ if (cnxn.chrootPath != null) {
+ // handle clientPath = "/"
+ if (clientPath.length() == 1) {
+ return cnxn.chrootPath;
+ }
+ return cnxn.chrootPath + clientPath;
+ } else {
+ return clientPath;
+ }
+ }
+
+ /**
+ * Create a node with the given path. The node data will be the given data,
+ * and node acl will be the given acl.
+ * <p>
+ * The flags argument specifies whether the created node will be ephemeral
+ * or not.
+ * <p>
+ * An ephemeral node will be removed by the ZooKeeper automatically when the
+ * session associated with the creation of the node expires.
+ * <p>
+ * The flags argument can also specify to create a sequential node. The
+ * actual path name of a sequential node will be the given path plus a
+ * suffix "i" where i is the current sequential number of the node. The sequence
+ * number is always fixed length of 10 digits, 0 padded. Once
+ * such a node is created, the sequential number will be incremented by one.
+ * <p>
+ * If a node with the same actual path already exists in the ZooKeeper, a
+ * KeeperException with error code KeeperException.NodeExists will be
+ * thrown. Note that since a different actual path is used for each
+ * invocation of creating sequential node with the same path argument, the
+ * call will never throw "file exists" KeeperException.
+ * <p>
+ * If the parent node does not exist in the ZooKeeper, a KeeperException
+ * with error code KeeperException.NoNode will be thrown.
+ * <p>
+ * An ephemeral node cannot have children. If the parent node of the given
+ * path is ephemeral, a KeeperException with error code
+ * KeeperException.NoChildrenForEphemerals will be thrown.
+ * <p>
+ * This operation, if successful, will trigger all the watches left on the
+ * node of the given path by exists and getData API calls, and the watches
+ * left on the parent node by getChildren API calls.
+ * <p>
+ * If a node is created successfully, the ZooKeeper server will trigger the
+ * watches on the path left by exists calls, and the watches on the parent
+ * of the node by getChildren calls.
+ * <p>
+ * The maximum allowable size of the data array is 1 MB (1,048,576 bytes).
+ * Arrays larger than this will cause a KeeperExecption to be thrown.
+ *
+ * @param path
+ * the path for the node
+ * @param data
+ * the initial data for the node
+ * @param acl
+ * the acl for the node
+ * @param createMode
+ * specifying whether the node to be created is ephemeral
+ * and/or sequential
+ * @return the actual path of the created node
+ * @throws KeeperException if the server returns a non-zero error code
+ * @throws KeeperException.InvalidACLException if the ACL is invalid, null, or empty
+ * @throws InterruptedException if the transaction is interrupted
+ * @throws IllegalArgumentException if an invalid path is specified
+ */
+ public String create(final String path, byte data[], List<ACL> acl,
+ CreateMode createMode)
+ throws KeeperException, InterruptedException
+ {
+ final String clientPath = path;
+ PathUtils.validatePath(clientPath, createMode.isSequential());
+ EphemeralType.validateTTL(createMode, -1);
+
+ final String serverPath = prependChroot(clientPath);
+
+ RequestHeader h = new RequestHeader();
+ h.setType(createMode.isContainer() ? ZooDefs.OpCode.createContainer : ZooDefs.OpCode.create);
+ CreateRequest request = new CreateRequest();
+ CreateResponse response = new CreateResponse();
+ request.setData(data);
+ request.setFlags(createMode.toFlag());
+ request.setPath(serverPath);
+ if (acl != null && acl.size() == 0) {
+ throw new KeeperException.InvalidACLException();
+ }
+ request.setAcl(acl);
+ ReplyHeader r = cnxn.submitRequest(h, request, response, null);
+ if (r.getErr() != 0) {
+ throw KeeperException.create(KeeperException.Code.get(r.getErr()),
+ clientPath);
+ }
+ if (cnxn.chrootPath == null) {
+ return response.getPath();
+ } else {
+ return response.getPath().substring(cnxn.chrootPath.length());
+ }
+ }
+
+ /**
+ * Create a node with the given path and returns the Stat of that node. The
+ * node data will be the given data and node acl will be the given acl.
+ * <p>
+ * The flags argument specifies whether the created node will be ephemeral
+ * or not.
+ * <p>
+ * An ephemeral node will be removed by the ZooKeeper automatically when the
+ * session associated with the creation of the node expires.
+ * <p>
+ * The flags argument can also specify to create a sequential node. The
+ * actual path name of a sequential node will be the given path plus a
+ * suffix "i" where i is the current sequential number of the node. The sequence
+ * number is always fixed length of 10 digits, 0 padded. Once
+ * such a node is created, the sequential number will be incremented by one.
+ * <p>
+ * If a node with the same actual path already exists in the ZooKeeper, a
+ * KeeperException with error code KeeperException.NodeExists will be
+ * thrown. Note that since a different actual path is used for each
+ * invocation of creating sequential node with the same path argument, the
+ * call will never throw "file exists" KeeperException.
+ * <p>
+ * If the parent node does not exist in the ZooKeeper, a KeeperException
+ * with error code KeeperException.NoNode will be thrown.
+ * <p>
+ * An ephemeral node cannot have children. If the parent node of the given
+ * path is ephemeral, a KeeperException with error code
+ * KeeperException.NoChildrenForEphemerals will be thrown.
+ * <p>
+ * This operation, if successful, will trigger all the watches left on the
+ * node of the given path by exists and getData API calls, and the watches
+ * left on the parent node by getChildren API calls.
+ * <p>
+ * If a node is created successfully, the ZooKeeper server will trigger the
+ * watches on the path left by exists calls, and the watches on the parent
+ * of the node by getChildren calls.
+ * <p>
+ * The maximum allowable size of the data array is 1 MB (1,048,576 bytes).
+ * Arrays larger than this will cause a KeeperExecption to be thrown.
+ *
+ * @param path
+ * the path for the node
+ * @param data
+ * the initial data for the node
+ * @param acl
+ * the acl for the node
+ * @param createMode
+ * specifying whether the node to be created is ephemeral
+ * and/or sequential
+ * @param stat
+ * The output Stat object.
+ * @return the actual path of the created node
+ * @throws KeeperException if the server returns a non-zero error code
+ * @throws KeeperException.InvalidACLException if the ACL is invalid, null, or empty
+ * @throws InterruptedException if the transaction is interrupted
+ * @throws IllegalArgumentException if an invalid path is specified
+ */
+ public String create(final String path, byte data[], List<ACL> acl,
+ CreateMode createMode, Stat stat)
+ throws KeeperException, InterruptedException {
+ return create(path, data, acl, createMode, stat, -1);
+ }
+
+ /**
+ * same as {@link #create(String, byte[], List, CreateMode, Stat)} but
+ * allows for specifying a TTL when mode is {@link CreateMode#PERSISTENT_WITH_TTL}
+ * or {@link CreateMode#PERSISTENT_SEQUENTIAL_WITH_TTL}. If the znode has not been modified
+ * within the given TTL, it will be deleted once it has no children. The TTL unit is
+ * milliseconds and must be greater than 0 and less than or equal to
+ * {@link EphemeralType#maxValue()} for {@link EphemeralType#TTL}.
+ */
+ public String create(final String path, byte data[], List<ACL> acl,
+ CreateMode createMode, Stat stat, long ttl)
+ throws KeeperException, InterruptedException {
+ final String clientPath = path;
+ PathUtils.validatePath(clientPath, createMode.isSequential());
+ EphemeralType.validateTTL(createMode, ttl);
+
+ final String serverPath = prependChroot(clientPath);
+
+ RequestHeader h = new RequestHeader();
+ setCreateHeader(createMode, h);
+ Create2Response response = new Create2Response();
+ if (acl != null && acl.size() == 0) {
+ throw new KeeperException.InvalidACLException();
+ }
+ Record record = makeCreateRecord(createMode, serverPath, data, acl, ttl);
+ ReplyHeader r = cnxn.submitRequest(h, record, response, null);
+ if (r.getErr() != 0) {
+ throw KeeperException.create(KeeperException.Code.get(r.getErr()),
+ clientPath);
+ }
+ if (stat != null) {
+ DataTree.copyStat(response.getStat(), stat);
+ }
+ if (cnxn.chrootPath == null) {
+ return response.getPath();
+ } else {
+ return response.getPath().substring(cnxn.chrootPath.length());
+ }
+ }
+
+ private void setCreateHeader(CreateMode createMode, RequestHeader h) {
+ if (createMode.isTTL()) {
+ h.setType(ZooDefs.OpCode.createTTL);
+ } else {
+ h.setType(createMode.isContainer() ? ZooDefs.OpCode.createContainer : ZooDefs.OpCode.create2);
+ }
+ }
+
+ private Record makeCreateRecord(CreateMode createMode, String serverPath, byte[] data, List<ACL> acl, long ttl) {
+ Record record;
+ if (createMode.isTTL()) {
+ CreateTTLRequest request = new CreateTTLRequest();
+ request.setData(data);
+ request.setFlags(createMode.toFlag());
+ request.setPath(serverPath);
+ request.setAcl(acl);
+ request.setTtl(ttl);
+ record = request;
+ } else {
+ CreateRequest request = new CreateRequest();
+ request.setData(data);
+ request.setFlags(createMode.toFlag());
+ request.setPath(serverPath);
+ request.setAcl(acl);
+ record = request;
+ }
+ return record;
+ }
+
+ /**
+ * The asynchronous version of create.
+ *
+ * @see #create(String, byte[], List, CreateMode)
+ */
+ public void create(final String path, byte data[], List<ACL> acl,
+ CreateMode createMode, StringCallback cb, Object ctx)
+ {
+ final String clientPath = path;
+ PathUtils.validatePath(clientPath, createMode.isSequential());
+ EphemeralType.validateTTL(createMode, -1);
+
+ final String serverPath = prependChroot(clientPath);
+
+ RequestHeader h = new RequestHeader();
+ h.setType(createMode.isContainer() ? ZooDefs.OpCode.createContainer : ZooDefs.OpCode.create);
+ CreateRequest request = new CreateRequest();
+ CreateResponse response = new CreateResponse();
+ ReplyHeader r = new ReplyHeader();
+ request.setData(data);
+ request.setFlags(createMode.toFlag());
+ request.setPath(serverPath);
+ request.setAcl(acl);
+ cnxn.queuePacket(h, r, request, response, cb, clientPath,
+ serverPath, ctx, null);
+ }
+
+ /**
+ * The asynchronous version of create.
+ *
+ * @see #create(String, byte[], List, CreateMode, Stat)
+ */
+ public void create(final String path, byte data[], List<ACL> acl,
+ CreateMode createMode, Create2Callback cb, Object ctx)
+ {
+ create(path, data, acl, createMode, cb, ctx, -1);
+ }
+
+ /**
+ * The asynchronous version of create with ttl.
+ *
+ * @see #create(String, byte[], List, CreateMode, Stat, long)
+ */
+ public void create(final String path, byte data[], List<ACL> acl,
+ CreateMode createMode, Create2Callback cb, Object ctx, long ttl)
+ {
+ final String clientPath = path;
+ PathUtils.validatePath(clientPath, createMode.isSequential());
+ EphemeralType.validateTTL(createMode, ttl);
+
+ final String serverPath = prependChroot(clientPath);
+
+ RequestHeader h = new RequestHeader();
+ setCreateHeader(createMode, h);
+ ReplyHeader r = new ReplyHeader();
+ Create2Response response = new Create2Response();
+ Record record = makeCreateRecord(createMode, serverPath, data, acl, ttl);
+ cnxn.queuePacket(h, r, record, response, cb, clientPath,
+ serverPath, ctx, null);
+ }
+
+ /**
+ * Delete the node with the given path. The call will succeed if such a node
+ * exists, and the given version matches the node's version (if the given
+ * version is -1, it matches any node's versions).
+ * <p>
+ * A KeeperException with error code KeeperException.NoNode will be thrown
+ * if the nodes does not exist.
+ * <p>
+ * A KeeperException with error code KeeperException.BadVersion will be
+ * thrown if the given version does not match the node's version.
+ * <p>
+ * A KeeperException with error code KeeperException.NotEmpty will be thrown
+ * if the node has children.
+ * <p>
+ * This operation, if successful, will trigger all the watches on the node
+ * of the given path left by exists API calls, and the watches on the parent
+ * node left by getChildren API calls.
+ *
+ * @param path
+ * the path of the node to be deleted.
+ * @param version
+ * the expected node version.
+ * @throws InterruptedException IF the server transaction is interrupted
+ * @throws KeeperException If the server signals an error with a non-zero
+ * return code.
+ * @throws IllegalArgumentException if an invalid path is specified
+ */
+ public void delete(final String path, int version)
+ throws InterruptedException, KeeperException
+ {
+ final String clientPath = path;
+ PathUtils.validatePath(clientPath);
+
+ final String serverPath;
+
+ // maintain semantics even in chroot case
+ // specifically - root cannot be deleted
+ // I think this makes sense even in chroot case.
+ if (clientPath.equals("/")) {
+ // a bit of a hack, but delete(/) will never succeed and ensures
+ // that the same semantics are maintained
+ serverPath = clientPath;
+ } else {
+ serverPath = prependChroot(clientPath);
+ }
+
+ RequestHeader h = new RequestHeader();
+ h.setType(ZooDefs.OpCode.delete);
+ DeleteRequest request = new DeleteRequest();
+ request.setPath(serverPath);
+ request.setVersion(version);
+ ReplyHeader r = cnxn.submitRequest(h, request, null, null);
+ if (r.getErr() != 0) {
+ throw KeeperException.create(KeeperException.Code.get(r.getErr()),
+ clientPath);
+ }
+ }
+
+ /**
+ * Executes multiple ZooKeeper operations or none of them.
+ * <p>
+ * On success, a list of results is returned.
+ * On failure, an exception is raised which contains partial results and
+ * error details, see {@link KeeperException#getResults}
+ * <p>
+ * Note: The maximum allowable size of all of the data arrays in all of
+ * the setData operations in this single request is typically 1 MB
+ * (1,048,576 bytes). This limit is specified on the server via
+ * <a href="http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#Unsafe+Options">jute.maxbuffer</a>.
+ * Requests larger than this will cause a KeeperException to be
+ * thrown.
+ *
+ * @param ops An iterable that contains the operations to be done.
+ * These should be created using the factory methods on {@link Op}.
+ * @return A list of results, one for each input Op, the order of
+ * which exactly matches the order of the <code>ops</code> input
+ * operations.
+ * @throws InterruptedException If the operation was interrupted.
+ * The operation may or may not have succeeded, but will not have
+ * partially succeeded if this exception is thrown.
+ * @throws KeeperException If the operation could not be completed
+ * due to some error in doing one of the specified ops.
+ * @throws IllegalArgumentException if an invalid path is specified
+ *
+ * @since 3.4.0
+ */
+ public List<OpResult> multi(Iterable<Op> ops) throws InterruptedException, KeeperException {
+ for (Op op : ops) {
+ op.validate();
+ }
+ return multiInternal(generateMultiTransaction(ops));
+ }
+
+ /**
+ * The asynchronous version of multi.
+ *
+ * @see #multi(Iterable)
+ */
+ public void multi(Iterable<Op> ops, MultiCallback cb, Object ctx) {
+ List<OpResult> results = validatePath(ops);
+ if (results.size() > 0) {
+ cb.processResult(KeeperException.Code.BADARGUMENTS.intValue(),
+ null, ctx, results);
+ return;
+ }
+ multiInternal(generateMultiTransaction(ops), cb, ctx);
+ }
+
+ private List<OpResult> validatePath(Iterable<Op> ops) {
+ List<OpResult> results = new ArrayList<OpResult>();
+ boolean error = false;
+ for (Op op : ops) {
+ try {
+ op.validate();
+ } catch (IllegalArgumentException iae) {
+ LOG.error("IllegalArgumentException: " + iae.getMessage());
+ ErrorResult err = new ErrorResult(
+ KeeperException.Code.BADARGUMENTS.intValue());
+ results.add(err);
+ error = true;
+ continue;
+ } catch (KeeperException ke) {
+ LOG.error("KeeperException: " + ke.getMessage());
+ ErrorResult err = new ErrorResult(ke.code().intValue());
+ results.add(err);
+ error = true;
+ continue;
+ }
+ ErrorResult err = new ErrorResult(
+ KeeperException.Code.RUNTIMEINCONSISTENCY.intValue());
+ results.add(err);
+ }
+ if (false == error) {
+ results.clear();
+ }
+ return results;
+ }
+
+ private MultiTransactionRecord generateMultiTransaction(Iterable<Op> ops) {
+ // reconstructing transaction with the chroot prefix
+ List<Op> transaction = new ArrayList<Op>();
+ for (Op op : ops) {
+ transaction.add(withRootPrefix(op));
+ }
+ return new MultiTransactionRecord(transaction);
+ }
+
+ private Op withRootPrefix(Op op) {
+ if (null != op.getPath()) {
+ final String serverPath = prependChroot(op.getPath());
+ if (!op.getPath().equals(serverPath)) {
+ return op.withChroot(serverPath);
+ }
+ }
+ return op;
+ }
+
+ protected void multiInternal(MultiTransactionRecord request, MultiCallback cb, Object ctx) {
+ RequestHeader h = new RequestHeader();
+ h.setType(ZooDefs.OpCode.multi);
+ MultiResponse response = new MultiResponse();
+ cnxn.queuePacket(h, new ReplyHeader(), request, response, cb, null, null, ctx, null);
+ }
+
+ protected List<OpResult> multiInternal(MultiTransactionRecord request)
+ throws InterruptedException, KeeperException {
+ RequestHeader h = new RequestHeader();
+ h.setType(ZooDefs.OpCode.multi);
+ MultiResponse response = new MultiResponse();
+ ReplyHeader r = cnxn.submitRequest(h, request, response, null);
+ if (r.getErr() != 0) {
+ throw KeeperException.create(KeeperException.Code.get(r.getErr()));
+ }
+
+ List<OpResult> results = response.getResultList();
+
+ ErrorResult fatalError = null;
+ for (OpResult result : results) {
+ if (result instanceof ErrorResult && ((ErrorResult)result).getErr() != KeeperException.Code.OK.intValue()) {
+ fatalError = (ErrorResult) result;
+ break;
+ }
+ }
+
+ if (fatalError != null) {
+ KeeperException ex = KeeperException.create(KeeperException.Code.get(fatalError.getErr()));
+ ex.setMultiResults(results);
+ throw ex;
+ }
+
+ return results;
+ }
+
+ /**
+ * A Transaction is a thin wrapper on the {@link #multi} method
+ * which provides a builder object that can be used to construct
+ * and commit an atomic set of operations.
+ *
+ * @since 3.4.0
+ *
+ * @return a Transaction builder object
+ */
+ public Transaction transaction() {
+ return new Transaction(this);
+ }
+
+ /**
+ * The asynchronous version of delete.
+ *
+ * @see #delete(String, int)
+ */
+ public void delete(final String path, int version, VoidCallback cb,
+ Object ctx)
+ {
+ final String clientPath = path;
+ PathUtils.validatePath(clientPath);
+
+ final String serverPath;
+
+ // maintain semantics even in chroot case
+ // specifically - root cannot be deleted
+ // I think this makes sense even in chroot case.
+ if (clientPath.equals("/")) {
+ // a bit of a hack, but delete(/) will never succeed and ensures
+ // that the same semantics are maintained
+ serverPath = clientPath;
+ } else {
+ serverPath = prependChroot(clientPath);
+ }
+
+ RequestHeader h = new RequestHeader();
+ h.setType(ZooDefs.OpCode.delete);
+ DeleteRequest request = new DeleteRequest();
+ request.setPath(serverPath);
+ request.setVersion(version);
+ cnxn.queuePacket(h, new ReplyHeader(), request, null, cb, clientPath,
+ serverPath, ctx, null);
+ }
+
+ /**
+ * Return the stat of the node of the given path. Return null if no such a
+ * node exists.
+ * <p>
+ * If the watch is non-null and the call is successful (no exception is thrown),
+ * a watch will be left on the node with the given path. The watch will be
+ * triggered by a successful operation that creates/delete the node or sets
+ * the data on the node.
+ *
+ * @param path the node path
+ * @param watcher explicit watcher
+ * @return the stat of the node of the given path; return null if no such a
+ * node exists.
+ * @throws KeeperException If the server signals an error
+ * @throws InterruptedException If the server transaction is interrupted.
+ * @throws IllegalArgumentException if an invalid path is specified
+ */
+ public Stat exists(final String path, Watcher watcher)
+ throws KeeperException, InterruptedException
+ {
+ final String clientPath = path;
+ PathUtils.validatePath(clientPath);
+
+ // the watch contains the un-chroot path
+ WatchRegistration wcb = null;
+ if (watcher != null) {
+ wcb = new ExistsWatchRegistration(watcher, clientPath);
+ }
+
+ final String serverPath = prependChroot(clientPath);
+
+ RequestHeader h = new RequestHeader();
+ h.setType(ZooDefs.OpCode.exists);
+ ExistsRequest request = new ExistsRequest();
+ request.setPath(serverPath);
+ request.setWatch(watcher != null);
+ SetDataResponse response = new SetDataResponse();
+ ReplyHeader r = cnxn.submitRequest(h, request, response, wcb);
+ if (r.getErr() != 0) {
+ if (r.getErr() == KeeperException.Code.NONODE.intValue()) {
+ return null;
+ }
+ throw KeeperException.create(KeeperException.Code.get(r.getErr()),
+ clientPath);
+ }
+
+ return response.getStat().getCzxid() == -1 ? null : response.getStat();
+ }
+
+ /**
+ * Return the stat of the node of the given path. Return null if no such a
+ * node exists.
+ * <p>
+ * If the watch is true and the call is successful (no exception is thrown),
+ * a watch will be left on the node with the given path. The watch will be
+ * triggered by a successful operation that creates/delete the node or sets
+ * the data on the node.
+ *
+ * @param path
+ * the node path
+ * @param watch
+ * whether need to watch this node
+ * @return the stat of the node of the given path; return null if no such a
+ * node exists.
+ * @throws KeeperException If the server signals an error
+ * @throws InterruptedException If the server transaction is interrupted.
+ */
+ public Stat exists(String path, boolean watch) throws KeeperException,
+ InterruptedException
+ {
+ return exists(path, watch ? watchManager.defaultWatcher : null);
+ }
+
+ /**
+ * The asynchronous version of exists.
+ *
+ * @see #exists(String, Watcher)
+ */
+ public void exists(final String path, Watcher watcher,
+ StatCallback cb, Object ctx)
+ {
+ final String clientPath = path;
+ PathUtils.validatePath(clientPath);
+
+ // the watch contains the un-chroot path
+ WatchRegistration wcb = null;
+ if (watcher != null) {
+ wcb = new ExistsWatchRegistration(watcher, clientPath);
+ }
+
+ final String serverPath = prependChroot(clientPath);
+
+ RequestHeader h = new RequestHeader();
+ h.setType(ZooDefs.OpCode.exists);
+ ExistsRequest request = new ExistsRequest();
+ request.setPath(serverPath);
+ request.setWatch(watcher != null);
+ SetDataResponse response = new SetDataResponse();
+ cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
+ clientPath, serverPath, ctx, wcb);
+ }
+
+ /**
+ * The asynchronous version of exists.
+ *
+ * @see #exists(String, boolean)
+ */
+ public void exists(String path, boolean watch, StatCallback cb, Object ctx) {
+ exists(path, watch ? watchManager.defaultWatcher : null, cb, ctx);
+ }
+
+ /**
+ * Return the data and the stat of the node of the given path.
+ * <p>
+ * If the watch is non-null and the call is successful (no exception is
+ * thrown), a watch will be left on the node with the given path. The watch
+ * will be triggered by a successful operation that sets data on the node, or
+ * deletes the node.
+ * <p>
+ * A KeeperException with error code KeeperException.NoNode will be thrown
+ * if no node with the given path exists.
+ *
+ * @param path the given path
+ * @param watcher explicit watcher
+ * @param stat the stat of the node
+ * @return the data of the node
+ * @throws KeeperException If the server signals an error with a non-zero error code
+ * @throws InterruptedException If the server transaction is interrupted.
+ * @throws IllegalArgumentException if an invalid path is specified
+ */
+ public byte[] getData(final String path, Watcher watcher, Stat stat)
+ throws KeeperException, InterruptedException
+ {
+ final String clientPath = path;
+ PathUtils.validatePath(clientPath);
+
+ // the watch contains the un-chroot path
+ WatchRegistration wcb = null;
+ if (watcher != null) {
+ wcb = new DataWatchRegistration(watcher, clientPath);
+ }
+
+ final String serverPath = prependChroot(clientPath);
+
+ RequestHeader h = new RequestHeader();
+ h.setType(ZooDefs.OpCode.getData);
+ GetDataRequest request = new GetDataRequest();
+ request.setPath(serverPath);
+ request.setWatch(watcher != null);
+ GetDataResponse response = new GetDataResponse();
+ ReplyHeader r = cnxn.submitRequest(h, request, response, wcb);
+ if (r.getErr() != 0) {
+ throw KeeperException.create(KeeperException.Code.get(r.getErr()),
+ clientPath);
+ }
+ if (stat != null) {
+ DataTree.copyStat(response.getStat(), stat);
+ }
+ return response.getData();
+ }
+
+ /**
+ * Return the data and the stat of the node of the given path.
+ * <p>
+ * If the watch is true and the call is successful (no exception is
+ * thrown), a watch will be left on the node with the given path. The watch
+ * will be triggered by a successful operation that sets data on the node, or
+ * deletes the node.
+ * <p>
+ * A KeeperException with error code KeeperException.NoNode will be thrown
+ * if no node with the given path exists.
+ *
+ * @param path the given path
+ * @param watch whether need to watch this node
+ * @param stat the stat of the node
+ * @return the data of the node
+ * @throws KeeperException If the server signals an error with a non-zero error code
+ * @throws InterruptedException If the server transaction is interrupted.
+ */
+ public byte[] getData(String path, boolean watch, Stat stat)
+ throws KeeperException, InterruptedException {
+ return getData(path, watch ? watchManager.defaultWatcher : null, stat);
+ }
+
+ /**
+ * The asynchronous version of getData.
+ *
+ * @see #getData(String, Watcher, Stat)
+ */
+ public void getData(final String path, Watcher watcher,
+ DataCallback cb, Object ctx)
+ {
+ final String clientPath = path;
+ PathUtils.validatePath(clientPath);
+
+ // the watch contains the un-chroot path
+ WatchRegistration wcb = null;
+ if (watcher != null) {
+ wcb = new DataWatchRegistration(watcher, clientPath);
+ }
+
+ final String serverPath = prependChroot(clientPath);
+
+ RequestHeader h = new RequestHeader();
+ h.setType(ZooDefs.OpCode.getData);
+ GetDataRequest request = new GetDataRequest();
+ request.setPath(serverPath);
+ request.setWatch(watcher != null);
+ GetDataResponse response = new GetDataResponse();
+ cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
+ clientPath, serverPath, ctx, wcb);
+ }
+
+ /**
+ * The asynchronous version of getData.
+ *
+ * @see #getData(String, boolean, Stat)
+ */
+ public void getData(String path, boolean watch, DataCallback cb, Object ctx) {
+ getData(path, watch ? watchManager.defaultWatcher : null, cb, ctx);
+ }
+
+ /**
+ * Return the last committed configuration (as known to the server to which the client is connected)
+ * and the stat of the configuration.
+ * <p>
+ * If the watch is non-null and the call is successful (no exception is
+ * thrown), a watch will be left on the configuration node (ZooDefs.CONFIG_NODE). The watch
+ * will be triggered by a successful reconfig operation
+ * <p>
+ * A KeeperException with error code KeeperException.NoNode will be thrown
+ * if the configuration node doesn't exists.
+ *
+ * @param watcher explicit watcher
+ * @param stat the stat of the configuration node ZooDefs.CONFIG_NODE
+ * @return configuration data stored in ZooDefs.CONFIG_NODE
+ * @throws KeeperException If the server signals an error with a non-zero error code
+ * @throws InterruptedException If the server transaction is interrupted.
+ */
+ public byte[] getConfig(Watcher watcher, Stat stat)
+ throws KeeperException, InterruptedException
+ {
+ final String configZnode = ZooDefs.CONFIG_NODE;
+
+ // the watch contains the un-chroot path
+ WatchRegistration wcb = null;
+ if (watcher != null) {
+ wcb = new DataWatchRegistration(watcher, configZnode);
+ }
+
+ RequestHeader h = new RequestHeader();
+ h.setType(ZooDefs.OpCode.getData);
+ GetDataRequest request = new GetDataRequest();
+ request.setPath(configZnode);
+ request.setWatch(watcher != null);
+ GetDataResponse response = new GetDataResponse();
+ ReplyHeader r = cnxn.submitRequest(h, request, response, wcb);
+ if (r.getErr() != 0) {
+ throw KeeperException.create(KeeperException.Code.get(r.getErr()),
+ configZnode);
+ }
+ if (stat != null) {
+ DataTree.copyStat(response.getStat(), stat);
+ }
+ return response.getData();
+ }
+
+ /**
+ * The asynchronous version of getConfig.
+ *
+ * @see #getConfig(Watcher, Stat)
+ */
+ public void getConfig(Watcher watcher,
+ DataCallback cb, Object ctx)
+ {
+ final String configZnode = ZooDefs.CONFIG_NODE;
+
+ // the watch contains the un-chroot path
+ WatchRegistration wcb = null;
+ if (watcher != null) {
+ wcb = new DataWatchRegistration(watcher, configZnode);
+ }
+
+ RequestHeader h = new RequestHeader();
+ h.setType(ZooDefs.OpCode.getData);
+ GetDataRequest request = new GetDataRequest();
+ request.setPath(configZnode);
+ request.setWatch(watcher != null);
+ GetDataResponse response = new GetDataResponse();
+ cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
+ configZnode, configZnode, ctx, wcb);
+ }
+
+
+ /**
+ * Return the last committed configuration (as known to the server to which the client is connected)
+ * and the stat of the configuration.
+ * <p>
+ * If the watch is true and the call is successful (no exception is
+ * thrown), a watch will be left on the configuration node (ZooDefs.CONFIG_NODE). The watch
+ * will be triggered by a successful reconfig operation
+ * <p>
+ * A KeeperException with error code KeeperException.NoNode will be thrown
+ * if no node with the given path exists.
+ *
+ * @param watch whether need to watch this node
+ * @param stat the stat of the configuration node ZooDefs.CONFIG_NODE
+ * @return configuration data stored in ZooDefs.CONFIG_NODE
+ * @throws KeeperException If the server signals an error with a non-zero error code
+ * @throws InterruptedException If the server transaction is interrupted.
+ */
+ public byte[] getConfig(boolean watch, Stat stat)
+ throws KeeperException, InterruptedException {
+ return getConfig(watch ? watchManager.defaultWatcher : null, stat);
+ }
+
+ /**
+ * The Asynchronous version of getConfig.
+ *
+ * @see #getData(String, boolean, Stat)
+ */
+ public void getConfig(boolean watch, DataCallback cb, Object ctx) {
+ getConfig(watch ? watchManager.defaultWatcher : null, cb, ctx);
+ }
+
+ /**
+ * @deprecated instead use the reconfigure() methods instead in {@link org.apache.zookeeper.admin.ZooKeeperAdmin}
+ */
+ @Deprecated
+ public byte[] reconfig(String joiningServers, String leavingServers, String newMembers, long fromConfig, Stat stat) throws KeeperException, InterruptedException {
+ return internalReconfig(joiningServers, leavingServers, newMembers, fromConfig, stat);
+ }
+
+ /**
+ * @deprecated instead use the reconfigure() methods instead in {@link org.apache.zookeeper.admin.ZooKeeperAdmin}
+ */
+ @Deprecated
+ public byte[] reconfig(List<String> joiningServers, List<String> leavingServers, List<String> newMembers, long fromConfig, Stat stat) throws KeeperException, InterruptedException {
+ return internalReconfig(joiningServers, leavingServers, newMembers, fromConfig, stat);
+ }
+
+ /**
+ * @deprecated instead use the reconfigure() methods instead in {@link org.apache.zookeeper.admin.ZooKeeperAdmin}
+ */
+ @Deprecated
+ public void reconfig(String joiningServers, String leavingServers,
+ String newMembers, long fromConfig, DataCallback cb, Object ctx) {
+ internalReconfig(joiningServers, leavingServers, newMembers, fromConfig, cb, ctx);
+ }
+
+ /**
+ * @deprecated instead use the reconfigure() methods instead in {@link org.apache.zookeeper.admin.ZooKeeperAdmin}
+ */
+ @Deprecated
+ public void reconfig(List<String> joiningServers,
+ List<String> leavingServers, List<String> newMembers, long fromConfig,
+ DataCallback cb, Object ctx) {
+ internalReconfig(joiningServers, leavingServers, newMembers, fromConfig, cb, ctx);
+ }
+
+ /**
+ * Set the data for the node of the given path if such a node exists and the
+ * given version matches the version of the node (if the given version is
+ * -1, it matches any node's versions). Return the stat of the node.
+ * <p>
+ * This operation, if successful, will trigger all the watches on the node
+ * of the given path left by getData calls.
+ * <p>
+ * A KeeperException with error code KeeperException.NoNode will be thrown
+ * if no node with the given path exists.
+ * <p>
+ * A KeeperException with error code KeeperException.BadVersion will be
+ * thrown if the given version does not match the node's version.
+ * <p>
+ * The maximum allowable size of the data array is 1 MB (1,048,576 bytes).
+ * Arrays larger than this will cause a KeeperException to be thrown.
+ *
+ * @param path
+ * the path of the node
+ * @param data
+ * the data to set
+ * @param version
+ * the expected matching version
+ * @return the state of the node
+ * @throws InterruptedException If the server transaction is interrupted.
+ * @throws KeeperException If the server signals an error with a non-zero error code.
+ * @throws IllegalArgumentException if an invalid path is specified
+ */
+ public Stat setData(final String path, byte data[], int version)
+ throws KeeperException, InterruptedException
+ {
+ final String clientPath = path;
+ PathUtils.validatePath(clientPath);
+
+ final String serverPath = prependChroot(clientPath);
+
+ RequestHeader h = new RequestHeader();
+ h.setType(ZooDefs.OpCode.setData);
+ SetDataRequest request = new SetDataRequest();
+ request.setPath(serverPath);
+ request.setData(data);
+ request.setVersion(version);
+ SetDataResponse response = new SetDataResponse();
+ ReplyHeader r = cnxn.submitRequest(h, request, response, null);
+ if (r.getErr() != 0) {
+ throw KeeperException.create(KeeperException.Code.get(r.getErr()),
+ clientPath);
+ }
+ return response.getStat();
+ }
+
+ /**
+ * The asynchronous version of setData.
+ *
+ * @see #setData(String, byte[], int)
+ */
+ public void setData(final String path, byte data[], int version,
+ StatCallback cb, Object ctx)
+ {
+ final String clientPath = path;
+ PathUtils.validatePath(clientPath);
+
+ final String serverPath = prependChroot(clientPath);
+
+ RequestHeader h = new RequestHeader();
+ h.setType(ZooDefs.OpCode.setData);
+ SetDataRequest request = new SetDataRequest();
+ request.setPath(serverPath);
+ request.setData(data);
+ request.setVersion(version);
+ SetDataResponse response = new SetDataResponse();
+ cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
+ clientPath, serverPath, ctx, null);
+ }
+
+ /**
+ * Return the ACL and stat of the node of the given path.
+ * <p>
+ * A KeeperException with error code KeeperException.NoNode will be thrown
+ * if no node with the given path exists.
+ *
+ * @param path
+ * the given path for the node
+ * @param stat
+ * the stat of the node will be copied to this parameter if
+ * not null.
+ * @return the ACL array of the given node.
+ * @throws InterruptedException If the server transaction is interrupted.
+ * @throws KeeperException If the server signals an error with a non-zero error code.
+ * @throws IllegalArgumentException if an invalid path is specified
+ */
+ public List<ACL> getACL(final String path, Stat stat)
+ throws KeeperException, InterruptedException
+ {
+ final String clientPath = path;
+ PathUtils.validatePath(clientPath);
+
+ final String serverPath = prependChroot(clientPath);
+
+ RequestHeader h = new RequestHeader();
+ h.setType(ZooDefs.OpCode.getACL);
+ GetACLRequest request = new GetACLRequest();
+ request.setPath(serverPath);
+ GetACLResponse response = new GetACLResponse();
+ ReplyHeader r = cnxn.submitRequest(h, request, response, null);
+ if (r.getErr() != 0) {
+ throw KeeperException.create(KeeperException.Code.get(r.getErr()),
+ clientPath);
+ }
+ if (stat != null) {
+ DataTree.copyStat(response.getStat(), stat);
+ }
+ return response.getAcl();
+ }
+
+ /**
+ * The asynchronous version of getACL.
+ *
+ * @see #getACL(String, Stat)
+ */
+ public void getACL(final String path, Stat stat, ACLCallback cb,
+ Object ctx)
+ {
+ final String clientPath = path;
+ PathUtils.validatePath(clientPath);
+
+ final String serverPath = prependChroot(clientPath);
+
+ RequestHeader h = new RequestHeader();
+ h.setType(ZooDefs.OpCode.getACL);
+ GetACLRequest request = new GetACLRequest();
+ request.setPath(serverPath);
+ GetACLResponse response = new GetACLResponse();
+ cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
+ clientPath, serverPath, ctx, null);
+ }
+
+ /**
+ * Set the ACL for the node of the given path if such a node exists and the
+ * given aclVersion matches the acl version of the node. Return the stat of the
+ * node.
+ * <p>
+ * A KeeperException with error code KeeperException.NoNode will be thrown
+ * if no node with the given path exists.
+ * <p>
+ * A KeeperException with error code KeeperException.BadVersion will be
+ * thrown if the given aclVersion does not match the node's aclVersion.
+ *
+ * @param path the given path for the node
+ * @param acl the given acl for the node
+ * @param aclVersion the given acl version of the node
+ * @return the stat of the node.
+ * @throws InterruptedException If the server transaction is interrupted.
+ * @throws KeeperException If the server signals an error with a non-zero error code.
+ * @throws org.apache.zookeeper.KeeperException.InvalidACLException If the acl is invalide.
+ * @throws IllegalArgumentException if an invalid path is specified
+ */
+ public Stat setACL(final String path, List<ACL> acl, int aclVersion)
+ throws KeeperException, InterruptedException
+ {
+ final String clientPath = path;
+ PathUtils.validatePath(clientPath);
+
+ final String serverPath = prependChroot(clientPath);
+
+ RequestHeader h = new RequestHeader();
+ h.setType(ZooDefs.OpCode.setACL);
+ SetACLRequest request = new SetACLRequest();
+ request.setPath(serverPath);
+ if (acl != null && acl.size() == 0) {
+ throw new KeeperException.InvalidACLException(clientPath);
+ }
+ request.setAcl(acl);
+ request.setVersion(aclVersion);
+ SetACLResponse response = new SetACLResponse();
+ ReplyHeader r = cnxn.submitRequest(h, request, response, null);
+ if (r.getErr() != 0) {
+ throw KeeperException.create(KeeperException.Code.get(r.getErr()),
+ clientPath);
+ }
+ return response.getStat();
+ }
+
+ /**
+ * The asynchronous version of setACL.
+ *
+ * @see #setACL(String, List, int)
+ */
+ public void setACL(final String path, List<ACL> acl, int version,
+ StatCallback cb, Object ctx)
+ {
+ final String clientPath = path;
+ PathUtils.validatePath(clientPath);
+
+ final String serverPath = prependChroot(clientPath);
+
+ RequestHeader h = new RequestHeader();
+ h.setType(ZooDefs.OpCode.setACL);
+ SetACLRequest request = new SetACLRequest();
+ request.setPath(serverPath);
+ request.setAcl(acl);
+ request.setVersion(version);
+ SetACLResponse response = new SetACLResponse();
+ cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
+ clientPath, serverPath, ctx, null);
+ }
+
+ /**
+ * Return the list of the children of the node of the given path.
+ * <p>
+ * If the watch is non-null and the call is successful (no exception is thrown),
+ * a watch will be left on the node with the given path. The watch will be
+ * triggered by a successful operation that deletes the node of the given
+ * path or creates/delete a child under the node.
+ * <p>
+ * The list of children returned is not sorted and no guarantee is provided
+ * as to its natural or lexical order.
+ * <p>
+ * A KeeperException with error code KeeperException.NoNode will be thrown
+ * if no node with the given path exists.
+ *
+ * @param path
+ * @param watcher explicit watcher
+ * @return an unordered array of children of the node with the given path
+ * @throws InterruptedException If the server transaction is interrupted.
+ * @throws KeeperException If the server signals an error with a non-zero error code.
+ * @throws IllegalArgumentException if an invalid path is specified
+ */
+ public List<String> getChildren(final String path, Watcher watcher)
+ throws KeeperException, InterruptedException
+ {
+ final String clientPath = path;
+ PathUtils.validatePath(clientPath);
+
+ // the watch contains the un-chroot path
+ WatchRegistration wcb = null;
+ if (watcher != null) {
+ wcb = new ChildWatchRegistration(watcher, clientPath);
+ }
+
+ final String serverPath = prependChroot(clientPath);
+
+ RequestHeader h = new RequestHeader();
+ h.setType(ZooDefs.OpCode.getChildren);
+ GetChildrenRequest request = new GetChildrenRequest();
+ request.setPath(serverPath);
+ request.setWatch(watcher != null);
+ GetChildrenResponse response = new GetChildrenResponse();
+ ReplyHeader r = cnxn.submitRequest(h, request, response, wcb);
+ if (r.getErr() != 0) {
+ throw KeeperException.create(KeeperException.Code.get(r.getErr()),
+ clientPath);
+ }
+ return response.getChildren();
+ }
+
+ /**
+ * Return the list of the children of the node of the given path.
+ * <p>
+ * If the watch is true and the call is successful (no exception is thrown),
+ * a watch will be left on the node with the given path. The watch will be
+ * triggered by a successful operation that deletes the node of the given
+ * path or creates/delete a child under the node.
+ * <p>
+ * The list of children returned is not sorted and no guarantee is provided
+ * as to its natural or lexical order.
+ * <p>
+ * A KeeperException with error code KeeperException.NoNode will be thrown
+ * if no node with the given path exists.
+ *
+ * @param path
+ * @param watch
+ * @return an unordered array of children of the node with the given path
+ * @throws InterruptedException If the server transaction is interrupted.
+ * @throws KeeperException If the server signals an error with a non-zero error code.
+ */
+ public List<String> getChildren(String path, boolean watch)
+ throws KeeperException, InterruptedException {
+ return getChildren(path, watch ? watchManager.defaultWatcher : null);
+ }
+
+ /**
+ * The asynchronous version of getChildren.
+ *
+ * @see #getChildren(String, Watcher)
+ */
+ public void getChildren(final String path, Watcher watcher,
+ ChildrenCallback cb, Object ctx)
+ {
+ final String clientPath = path;
+ PathUtils.validatePath(clientPath);
+
+ // the watch contains the un-chroot path
+ WatchRegistration wcb = null;
+ if (watcher != null) {
+ wcb = new ChildWatchRegistration(watcher, clientPath);
+ }
+
+ final String serverPath = prependChroot(clientPath);
+
+ RequestHeader h = new RequestHeader();
+ h.setType(ZooDefs.OpCode.getChildren);
+ GetChildrenRequest request = new GetChildrenRequest();
+ request.setPath(serverPath);
+ request.setWatch(watcher != null);
+ GetChildrenResponse response = new GetChildrenResponse();
+ cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
+ clientPath, serverPath, ctx, wcb);
+ }
+
+ /**
+ * The asynchronous version of getChildren.
+ *
+ * @see #getChildren(String, boolean)
+ */
+ public void getChildren(String path, boolean watch, ChildrenCallback cb,
+ Object ctx)
+ {
+ getChildren(path, watch ? watchManager.defaultWatcher : null, cb, ctx);
+ }
+
+ /**
+ * For the given znode path return the stat and children list.
+ * <p>
+ * If the watch is non-null and the call is successful (no exception is thrown),
+ * a watch will be left on the node with the given path. The watch will be
+ * triggered by a successful operation that deletes the node of the given
+ * path or creates/delete a child under the node.
+ * <p>
+ * The list of children returned is not sorted and no guarantee is provided
+ * as to its natural or lexical order.
+ * <p>
+ * A KeeperException with error code KeeperException.NoNode will be thrown
+ * if no node with the given path exists.
+ *
+ * @since 3.3.0
+ *
+ * @param path
+ * @param watcher explicit watcher
+ * @param stat stat of the znode designated by path
+ * @return an unordered array of children of the node with the given path
+ * @throws InterruptedException If the server transaction is interrupted.
+ * @throws KeeperException If the server signals an error with a non-zero error code.
+ * @throws IllegalArgumentException if an invalid path is specified
+ */
+ public List<String> getChildren(final String path, Watcher watcher,
+ Stat stat)
+ throws KeeperException, InterruptedException
+ {
+ final String clientPath = path;
+ PathUtils.validatePath(clientPath);
+
+ // the watch contains the un-chroot path
+ WatchRegistration wcb = null;
+ if (watcher != null) {
+ wcb = new ChildWatchRegistration(watcher, clientPath);
+ }
+
+ final String serverPath = prependChroot(clientPath);
+
+ RequestHeader h = new RequestHeader();
+ h.setType(ZooDefs.OpCode.getChildren2);
+ GetChildren2Request request = new GetChildren2Request();
+ request.setPath(serverPath);
+ request.setWatch(watcher != null);
+ GetChildren2Response response = new GetChildren2Response();
+ ReplyHeader r = cnxn.submitRequest(h, request, response, wcb);
+ if (r.getErr() != 0) {
+ throw KeeperException.create(KeeperException.Code.get(r.getErr()),
+ clientPath);
+ }
+ if (stat != null) {
+ DataTree.copyStat(response.getStat(), stat);
+ }
+ return response.getChildren();
+ }
+
+ /**
+ * For the given znode path return the stat and children list.
+ * <p>
+ * If the watch is true and the call is successful (no exception is thrown),
+ * a watch will be left on the node with the given path. The watch will be
+ * triggered by a successful operation that deletes the node of the given
+ * path or creates/delete a child under the node.
+ * <p>
+ * The list of children returned is not sorted and no guarantee is provided
+ * as to its natural or lexical order.
+ * <p>
+ * A KeeperException with error code KeeperException.NoNode will be thrown
+ * if no node with the given path exists.
+ *
+ * @since 3.3.0
+ *
+ * @param path
+ * @param watch
+ * @param stat stat of the znode designated by path
+ * @return an unordered array of children of the node with the given path
+ * @throws InterruptedException If the server transaction is interrupted.
+ * @throws KeeperException If the server signals an error with a non-zero
+ * error code.
+ */
+ public List<String> getChildren(String path, boolean watch, Stat stat)
+ throws KeeperException, InterruptedException {
+ return getChildren(path, watch ? watchManager.defaultWatcher : null,
+ stat);
+ }
+
+ /**
+ * The asynchronous version of getChildren.
+ *
+ * @since 3.3.0
+ *
+ * @see #getChildren(String, Watcher, Stat)
+ */
+ public void getChildren(final String path, Watcher watcher,
+ Children2Callback cb, Object ctx)
+ {
+ final String clientPath = path;
+ PathUtils.validatePath(clientPath);
+
+ // the watch contains the un-chroot path
+ WatchRegistration wcb = null;
+ if (watcher != null) {
+ wcb = new ChildWatchRegistration(watcher, clientPath);
+ }
+
+ final String serverPath = prependChroot(clientPath);
+
+ RequestHeader h = new RequestHeader();
+ h.setType(ZooDefs.OpCode.getChildren2);
+ GetChildren2Request request = new GetChildren2Request();
+ request.setPath(serverPath);
+ request.setWatch(watcher != null);
+ GetChildren2Response response = new GetChildren2Response();
+ cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
+ clientPath, serverPath, ctx, wcb);
+ }
+
+ /**
+ * The asynchronous version of getChildren.
+ *
+ * @since 3.3.0
+ *
+ * @see #getChildren(String, boolean, Stat)
+ */
+ public void getChildren(String path, boolean watch, Children2Callback cb,
+ Object ctx)
+ {
+ getChildren(path, watch ? watchManager.defaultWatcher : null, cb, ctx);
+ }
+
+ /**
+ * Asynchronous sync. Flushes channel between process and leader.
+ * @param path
+ * @param cb a handler for the callback
+ * @param ctx context to be provided to the callback
+ * @throws IllegalArgumentException if an invalid path is specified
+ */
+ public void sync(final String path, VoidCallback cb, Object ctx){
+ final String clientPath = path;
+ PathUtils.validatePath(clientPath);
+
+ final String serverPath = prependChroot(clientPath);
+
+ RequestHeader h = new RequestHeader();
+ h.setType(ZooDefs.OpCode.sync);
+ SyncRequest request = new SyncRequest();
+ SyncResponse response = new SyncResponse();
+ request.setPath(serverPath);
+ cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
+ clientPath, serverPath, ctx, null);
+ }
+
+ /**
+ * For the given znode path, removes the specified watcher of given
+ * watcherType.
+ *
+ * <p>
+ * Watcher shouldn't be null. A successful call guarantees that, the
+ * removed watcher won't be triggered.
+ * </p>
+ *
+ * @param path
+ * - the path of the node
+ * @param watcher
+ * - a concrete watcher
+ * @param watcherType
+ * - the type of watcher to be removed
+ * @param local
+ * - whether the watcher can be removed locally when there is no
+ * server connection
+ * @throws InterruptedException
+ * if the server transaction is interrupted.
+ * @throws KeeperException.NoWatcherException
+ * if no watcher exists that match the specified parameters
+ * @throws KeeperException
+ * if the server signals an error with a non-zero error code.
+ * @throws IllegalArgumentException
+ * if any of the following is true:
+ * <ul>
+ * <li> {@code path} is invalid
+ * <li> {@code watcher} is null
+ * </ul>
+ *
+ * @since 3.5.0
+ */
+ public void removeWatches(String path, Watcher watcher,
+ WatcherType watcherType, boolean local)
+ throws InterruptedException, KeeperException {
+ validateWatcher(watcher);
+ removeWatches(ZooDefs.OpCode.checkWatches, path, watcher,
+ watcherType, local);
+ }
+
+ /**
+ * The asynchronous version of removeWatches.
+ *
+ * @see #removeWatches
+ */
+ public void removeWatches(String path, Watcher watcher,
+ WatcherType watcherType, boolean local, VoidCallback cb, Object ctx) {
+ validateWatcher(watcher);
+ removeWatches(ZooDefs.OpCode.checkWatches, path, watcher,
+ watcherType, local, cb, ctx);
+ }
+
+ /**
+ * For the given znode path, removes all the registered watchers of given
+ * watcherType.
+ *
+ * <p>
+ * A successful call guarantees that, the removed watchers won't be
+ * triggered.
+ * </p>
+ *
+ * @param path
+ * - the path of the node
+ * @param watcherType
+ * - the type of watcher to be removed
+ * @param local
+ * - whether watches can be removed locally when there is no
+ * server connection
+ * @throws InterruptedException
+ * if the server transaction is interrupted.
+ * @throws KeeperException.NoWatcherException
+ * if no watcher exists that match the specified parameters
+ * @throws KeeperException
+ * if the server signals an error with a non-zero error code.
+ * @throws IllegalArgumentException
+ * if an invalid {@code path} is specified
+ *
+ * @since 3.5.0
+ */
+ public void removeAllWatches(String path, WatcherType watcherType,
+ boolean local) throws InterruptedException, KeeperException {
+
+ removeWatches(ZooDefs.OpCode.removeWatches, path, null, watcherType,
+ local);
+ }
+
+ /**
+ * The asynchronous version of removeAllWatches.
+ *
+ * @see #removeAllWatches
+ */
+ public void removeAllWatches(String path, WatcherType watcherType,
+ boolean local, VoidCallback cb, Object ctx) {
+
+ removeWatches(ZooDefs.OpCode.removeWatches, path, null,
+ watcherType, local, cb, ctx);
+ }
+
+ private void validateWatcher(Watcher watcher) {
+ if (watcher == null) {
+ throw new IllegalArgumentException(
+ "Invalid Watcher, shouldn't be null!");
+ }
+ }
+
+ private void removeWatches(int opCode, String path, Watcher watcher,
+ WatcherType watcherType, boolean local)
+ throws InterruptedException, KeeperException {
+ PathUtils.validatePath(path);
+ final String clientPath = path;
+ final String serverPath = prependChroot(clientPath);
+ WatchDeregistration wcb = new WatchDeregistration(clientPath, watcher,
+ watcherType, local, watchManager);
+
+ RequestHeader h = new RequestHeader();
+ h.setType(opCode);
+ Record request = getRemoveWatchesRequest(opCode, watcherType,
+ serverPath);
+
+ ReplyHeader r = cnxn.submitRequest(h, request, null, null, wcb);
+ if (r.getErr() != 0) {
+ throw KeeperException.create(KeeperException.Code.get(r.getErr()),
+ clientPath);
+ }
+ }
+
+ private void removeWatches(int opCode, String path, Watcher watcher,
+ WatcherType watcherType, boolean local, VoidCallback cb, Object ctx) {
+ PathUtils.validatePath(path);
+ final String clientPath = path;
+ final String serverPath = prependChroot(clientPath);
+ WatchDeregistration wcb = new WatchDeregistration(clientPath, watcher,
+ watcherType, local, watchManager);
+
+ RequestHeader h = new RequestHeader();
+ h.setType(opCode);
+ Record request = getRemoveWatchesRequest(opCode, watcherType,
+ serverPath);
+
+ cnxn.queuePacket(h, new ReplyHeader(), request, null, cb, clientPath,
+ serverPath, ctx, null, wcb);
+ }
+
+ private Record getRemoveWatchesRequest(int opCode, WatcherType watcherType,
+ final String serverPath) {
+ Record request = null;
+ switch (opCode) {
+ case ZooDefs.OpCode.checkWatches:
+ CheckWatchesRequest chkReq = new CheckWatchesRequest();
+ chkReq.setPath(serverPath);
+ chkReq.setType(watcherType.getIntValue());
+ request = chkReq;
+ break;
+ case ZooDefs.OpCode.removeWatches:
+ RemoveWatchesRequest rmReq = new RemoveWatchesRequest();
+ rmReq.setPath(serverPath);
+ rmReq.setType(watcherType.getIntValue());
+ request = rmReq;
+ break;
+ default:
+ LOG.warn("unknown type " + opCode);
+ break;
+ }
+ return request;
+ }
+
+ public States getState() {
+ return cnxn.getState();
+ }
+
+ /**
+ * String representation of this ZooKeeper client. Suitable for things
+ * like logging.
+ *
+ * Do NOT count on the format of this string, it may change without
+ * warning.
+ *
+ * @since 3.3.0
+ */
+ @Override
+ public String toString() {
+ States state = getState();
+ return ("State:" + state.toString()
+ + (state.isConnected() ?
+ " Timeout:" + getSessionTimeout() + " " :
+ " ")
+ + cnxn);
+ }
+
+ /*
+ * Methods to aid in testing follow.
+ *
+ * THESE METHODS ARE EXPECTED TO BE USED FOR TESTING ONLY!!!
+ */
+
+ /**
+ * Wait up to wait milliseconds for the underlying threads to shutdown.
+ * THIS METHOD IS EXPECTED TO BE USED FOR TESTING ONLY!!!
+ *
+ * @since 3.3.0
+ *
+ * @param wait max wait in milliseconds
+ * @return true iff all threads are shutdown, otw false
+ */
+ protected boolean testableWaitForShutdown(int wait)
+ throws InterruptedException
+ {
+ cnxn.sendThread.join(wait);
+ if (cnxn.sendThread.isAlive()) return false;
+ cnxn.eventThread.join(wait);
+ if (cnxn.eventThread.isAlive()) return false;
+ return true;
+ }
+
+ /**
+ * Returns the address to which the socket is connected. Useful for testing
+ * against an ensemble - test client may need to know which server
+ * to shutdown if interested in verifying that the code handles
+ * disconnection/reconnection correctly.
+ * THIS METHOD IS EXPECTED TO BE USED FOR TESTING ONLY!!!
+ *
+ * @since 3.3.0
+ *
+ * @return ip address of the remote side of the connection or null if
+ * not connected
+ */
+ protected SocketAddress testableRemoteSocketAddress() {
+ return cnxn.sendThread.getClientCnxnSocket().getRemoteSocketAddress();
+ }
+
+ /**
+ * Returns the local address to which the socket is bound.
+ * THIS METHOD IS EXPECTED TO BE USED FOR TESTING ONLY!!!
+ *
+ * @since 3.3.0
+ *
+ * @return ip address of the remote side of the connection or null if
+ * not connected
+ */
+ protected SocketAddress testableLocalSocketAddress() {
+ return cnxn.sendThread.getClientCnxnSocket().getLocalSocketAddress();
+ }
+
+ private ClientCnxnSocket getClientCnxnSocket() throws IOException {
+ String clientCnxnSocketName = getClientConfig().getProperty(
+ ZKClientConfig.ZOOKEEPER_CLIENT_CNXN_SOCKET);
+ if (clientCnxnSocketName == null) {
+ clientCnxnSocketName = ClientCnxnSocketNIO.class.getName();
+ }
+ try {
+ Constructor<?> clientCxnConstructor = Class.forName(clientCnxnSocketName).getDeclaredConstructor(ZKClientConfig.class);
+ ClientCnxnSocket clientCxnSocket = (ClientCnxnSocket) clientCxnConstructor.newInstance(getClientConfig());
+ return clientCxnSocket;
+ } catch (Exception e) {
+ IOException ioe = new IOException("Couldn't instantiate "
+ + clientCnxnSocketName);
+ ioe.initCause(e);
+ throw ioe;
+ }
+ }
+
+ protected byte[] internalReconfig(String joiningServers, String leavingServers, String newMembers, long fromConfig, Stat stat) throws KeeperException, InterruptedException {
+ RequestHeader h = new RequestHeader();
+ h.setType(ZooDefs.OpCode.reconfig);
+ ReconfigRequest request = new ReconfigRequest(joiningServers, leavingServers, newMembers, fromConfig);
+ GetDataResponse response = new GetDataResponse();
+ ReplyHeader r = cnxn.submitRequest(h, request, response, null);
+ if (r.getErr() != 0) {
+ throw KeeperException.create(KeeperException.Code.get(r.getErr()), "");
+ }
+ if (stat != null) {
+ DataTree.copyStat(response.getStat(), stat);
Review comment:
*NULL_DEREFERENCE:* object returned by `getStat(response)` could be null and is dereferenced by call to `copyStat(...)` at line 3084.
(at-me [in a reply](https://docs.muse.dev/docs/talk-to-muse/) with `help` or `ignore`)
##########
File path: zookeeper-server/src/main/java/org/apache/zookeeper/server/LogFormatter.java
##########
@@ -0,0 +1,106 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.server;
+
+import java.io.EOFException;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.text.DateFormat;
+import java.util.Date;
+import java.util.zip.Adler32;
+import java.util.zip.Checksum;
+
+import org.apache.jute.BinaryInputArchive;
+import org.apache.jute.Record;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.zookeeper.server.persistence.FileHeader;
+import org.apache.zookeeper.server.persistence.FileTxnLog;
+import org.apache.zookeeper.server.util.SerializeUtils;
+import org.apache.zookeeper.txn.TxnHeader;
+
+@InterfaceAudience.Public
+public class LogFormatter {
+ private static final Logger LOG = LoggerFactory.getLogger(LogFormatter.class);
+
+ /**
+ * @param args
+ */
+ public static void main(String[] args) throws Exception {
+ if (args.length != 1) {
+ System.err.println("USAGE: LogFormatter log_file");
+ System.exit(2);
+ }
+ FileInputStream fis = new FileInputStream(args[0]);
+ BinaryInputArchive logStream = BinaryInputArchive.getArchive(fis);
+ FileHeader fhdr = new FileHeader();
+ fhdr.deserialize(logStream, "fileheader");
+
+ if (fhdr.getMagic() != FileTxnLog.TXNLOG_MAGIC) {
+ System.err.println("Invalid magic number for " + args[0]);
+ System.exit(2);
+ }
+ System.out.println("ZooKeeper Transactional Log File with dbid "
+ + fhdr.getDbid() + " txnlog format version "
+ + fhdr.getVersion());
+
+ int count = 0;
+ while (true) {
+ long crcValue;
+ byte[] bytes;
+ try {
+ crcValue = logStream.readLong("crcvalue");
+
+ bytes = logStream.readBuffer("txnEntry");
+ } catch (EOFException e) {
+ System.out.println("EOF reached after " + count + " txns.");
+ return;
+ }
+ if (bytes.length == 0) {
Review comment:
*NULL_DEREFERENCE:* object `bytes` last assigned on line 71 could be null and is dereferenced at line 76.
(at-me [in a reply](https://docs.muse.dev/docs/talk-to-muse/) with `help` or `ignore`)
##########
File path: zookeeper-server/src/main/java/org/apache/zookeeper/server/PrepRequestProcessor.java
##########
@@ -0,0 +1,1012 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.server;
+
+import org.apache.jute.BinaryOutputArchive;
+import org.apache.jute.Record;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.KeeperException.BadArgumentsException;
+import org.apache.zookeeper.KeeperException.Code;
+import org.apache.zookeeper.MultiTransactionRecord;
+import org.apache.zookeeper.Op;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.ZooDefs.OpCode;
+import org.apache.zookeeper.common.PathUtils;
+import org.apache.zookeeper.common.StringUtils;
+import org.apache.zookeeper.common.Time;
+import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.data.Id;
+import org.apache.zookeeper.data.StatPersisted;
+import org.apache.zookeeper.proto.CheckVersionRequest;
+import org.apache.zookeeper.proto.CreateRequest;
+import org.apache.zookeeper.proto.CreateTTLRequest;
+import org.apache.zookeeper.proto.DeleteRequest;
+import org.apache.zookeeper.proto.ReconfigRequest;
+import org.apache.zookeeper.proto.SetACLRequest;
+import org.apache.zookeeper.proto.SetDataRequest;
+import org.apache.zookeeper.server.ZooKeeperServer.ChangeRecord;
+import org.apache.zookeeper.server.auth.AuthenticationProvider;
+import org.apache.zookeeper.server.auth.ProviderRegistry;
+import org.apache.zookeeper.server.quorum.Leader.XidRolloverException;
+import org.apache.zookeeper.server.quorum.LeaderZooKeeperServer;
+import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
+import org.apache.zookeeper.server.quorum.QuorumPeerConfig;
+import org.apache.zookeeper.server.quorum.QuorumPeerConfig.ConfigException;
+import org.apache.zookeeper.server.quorum.flexible.QuorumMaj;
+import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
+import org.apache.zookeeper.txn.CheckVersionTxn;
+import org.apache.zookeeper.txn.CreateContainerTxn;
+import org.apache.zookeeper.txn.CreateSessionTxn;
+import org.apache.zookeeper.txn.CreateTTLTxn;
+import org.apache.zookeeper.txn.CreateTxn;
+import org.apache.zookeeper.txn.DeleteTxn;
+import org.apache.zookeeper.txn.ErrorTxn;
+import org.apache.zookeeper.txn.MultiTxn;
+import org.apache.zookeeper.txn.SetACLTxn;
+import org.apache.zookeeper.txn.SetDataTxn;
+import org.apache.zookeeper.txn.Txn;
+import org.apache.zookeeper.txn.TxnHeader;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.StringReader;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.LinkedBlockingQueue;
+
+/**
+ * This request processor is generally at the start of a RequestProcessor
+ * change. It sets up any transactions associated with requests that change the
+ * state of the system. It counts on ZooKeeperServer to update
+ * outstandingRequests, so that it can take into account transactions that are
+ * in the queue to be applied when generating a transaction.
+ */
+public class PrepRequestProcessor extends ZooKeeperCriticalThread implements
+ RequestProcessor {
+ private static final Logger LOG = LoggerFactory.getLogger(PrepRequestProcessor.class);
+
+ static boolean skipACL;
+ static {
+ skipACL = System.getProperty("zookeeper.skipACL", "no").equals("yes");
+ if (skipACL) {
+ LOG.info("zookeeper.skipACL==\"yes\", ACL checks will be skipped");
+ }
+ }
+
+ /**
+ * this is only for testing purposes.
+ * should never be used otherwise
+ */
+ private static boolean failCreate = false;
+
+ LinkedBlockingQueue<Request> submittedRequests = new LinkedBlockingQueue<Request>();
+
+ private final RequestProcessor nextProcessor;
+
+ ZooKeeperServer zks;
+
+ public PrepRequestProcessor(ZooKeeperServer zks,
+ RequestProcessor nextProcessor) {
+ super("ProcessThread(sid:" + zks.getServerId() + " cport:"
+ + zks.getClientPort() + "):", zks.getZooKeeperServerListener());
+ this.nextProcessor = nextProcessor;
+ this.zks = zks;
+ }
+
+ /**
+ * method for tests to set failCreate
+ * @param b
+ */
+ public static void setFailCreate(boolean b) {
+ failCreate = b;
+ }
+ @Override
+ public void run() {
+ try {
+ while (true) {
+ Request request = submittedRequests.take();
+ long traceMask = ZooTrace.CLIENT_REQUEST_TRACE_MASK;
+ if (request.type == OpCode.ping) {
+ traceMask = ZooTrace.CLIENT_PING_TRACE_MASK;
+ }
+ if (LOG.isTraceEnabled()) {
+ ZooTrace.logRequest(LOG, traceMask, 'P', request, "");
+ }
+ if (Request.requestOfDeath == request) {
+ break;
+ }
+ pRequest(request);
+ }
+ } catch (RequestProcessorException e) {
+ if (e.getCause() instanceof XidRolloverException) {
+ LOG.info(e.getCause().getMessage());
+ }
+ handleException(this.getName(), e);
+ } catch (Exception e) {
+ handleException(this.getName(), e);
+ }
+ LOG.info("PrepRequestProcessor exited loop!");
+ }
+
+ private ChangeRecord getRecordForPath(String path) throws KeeperException.NoNodeException {
+ ChangeRecord lastChange = null;
+ synchronized (zks.outstandingChanges) {
+ lastChange = zks.outstandingChangesForPath.get(path);
+ if (lastChange == null) {
+ DataNode n = zks.getZKDatabase().getNode(path);
+ if (n != null) {
+ Set<String> children;
+ synchronized(n) {
+ children = n.getChildren();
+ }
+ lastChange = new ChangeRecord(-1, path, n.stat, children.size(),
+ zks.getZKDatabase().aclForNode(n));
+ }
+ }
+ }
+ if (lastChange == null || lastChange.stat == null) {
+ throw new KeeperException.NoNodeException(path);
+ }
+ return lastChange;
+ }
+
+ private ChangeRecord getOutstandingChange(String path) {
+ synchronized (zks.outstandingChanges) {
+ return zks.outstandingChangesForPath.get(path);
+ }
+ }
+
+ private void addChangeRecord(ChangeRecord c) {
+ synchronized (zks.outstandingChanges) {
+ zks.outstandingChanges.add(c);
+ zks.outstandingChangesForPath.put(c.path, c);
+ }
+ }
+
+ /**
+ * Grab current pending change records for each op in a multi-op.
+ *
+ * This is used inside MultiOp error code path to rollback in the event
+ * of a failed multi-op.
+ *
+ * @param multiRequest
+ * @return a map that contains previously existed records that probably need to be
+ * rolled back in any failure.
+ */
+ private Map<String, ChangeRecord> getPendingChanges(MultiTransactionRecord multiRequest) {
+ HashMap<String, ChangeRecord> pendingChangeRecords = new HashMap<String, ChangeRecord>();
+
+ for (Op op : multiRequest) {
+ String path = op.getPath();
+ ChangeRecord cr = getOutstandingChange(path);
+ // only previously existing records need to be rolled back.
+ if (cr != null) {
+ pendingChangeRecords.put(path, cr);
+ }
+
+ /*
+ * ZOOKEEPER-1624 - We need to store for parent's ChangeRecord
+ * of the parent node of a request. So that if this is a
+ * sequential node creation request, rollbackPendingChanges()
+ * can restore previous parent's ChangeRecord correctly.
+ *
+ * Otherwise, sequential node name generation will be incorrect
+ * for a subsequent request.
+ */
+ int lastSlash = path.lastIndexOf('/');
+ if (lastSlash == -1 || path.indexOf('\0') != -1) {
+ continue;
+ }
+ String parentPath = path.substring(0, lastSlash);
+ ChangeRecord parentCr = getOutstandingChange(parentPath);
+ if (parentCr != null) {
+ pendingChangeRecords.put(parentPath, parentCr);
+ }
+ }
+
+ return pendingChangeRecords;
+ }
+
+ /**
+ * Rollback pending changes records from a failed multi-op.
+ *
+ * If a multi-op fails, we can't leave any invalid change records we created
+ * around. We also need to restore their prior value (if any) if their prior
+ * value is still valid.
+ *
+ * @param zxid
+ * @param pendingChangeRecords
+ */
+ void rollbackPendingChanges(long zxid, Map<String, ChangeRecord>pendingChangeRecords) {
+ synchronized (zks.outstandingChanges) {
+ // Grab a list iterator starting at the END of the list so we can iterate in reverse
+ Iterator<ChangeRecord> iter = zks.outstandingChanges.descendingIterator();
+ while (iter.hasNext()) {
+ ChangeRecord c = iter.next();
+ if (c.zxid == zxid) {
+ iter.remove();
+ // Remove all outstanding changes for paths of this multi.
+ // Previous records will be added back later.
+ zks.outstandingChangesForPath.remove(c.path);
+ } else {
+ break;
+ }
+ }
+
+ // we don't need to roll back any records because there is nothing left.
+ if (zks.outstandingChanges.isEmpty()) {
+ return;
+ }
+
+ long firstZxid = zks.outstandingChanges.peek().zxid;
+
+ for (ChangeRecord c : pendingChangeRecords.values()) {
+ // Don't apply any prior change records less than firstZxid.
+ // Note that previous outstanding requests might have been removed
+ // once they are completed.
+ if (c.zxid < firstZxid) {
+ continue;
+ }
+
+ // add previously existing records back.
+ zks.outstandingChangesForPath.put(c.path, c);
+ }
+ }
+ }
+
+ /**
+ * Grant or deny authorization to an operation on a node as a function of:
+ *
+ * @param zks: not used.
+ * @param acl: set of ACLs for the node
+ * @param perm: the permission that the client is requesting
+ * @param ids: the credentials supplied by the client
+ */
+ static void checkACL(ZooKeeperServer zks, List<ACL> acl, int perm,
+ List<Id> ids) throws KeeperException.NoAuthException {
+ if (skipACL) {
+ return;
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Permission requested: {} ", perm);
+ LOG.debug("ACLs for node: {}", acl);
+ LOG.debug("Client credentials: {}", ids);
+ }
+ if (acl == null || acl.size() == 0) {
+ return;
+ }
+ for (Id authId : ids) {
+ if (authId.getScheme().equals("super")) {
+ return;
+ }
+ }
+ for (ACL a : acl) {
+ Id id = a.getId();
+ if ((a.getPerms() & perm) != 0) {
+ if (id.getScheme().equals("world")
+ && id.getId().equals("anyone")) {
+ return;
+ }
+ AuthenticationProvider ap = ProviderRegistry.getProvider(id
+ .getScheme());
+ if (ap != null) {
+ for (Id authId : ids) {
+ if (authId.getScheme().equals(id.getScheme())
+ && ap.matches(authId.getId(), id.getId())) {
+ return;
+ }
+ }
+ }
+ }
+ }
+ throw new KeeperException.NoAuthException();
+ }
+
+ /**
+ * Performs basic validation of a path for a create request.
+ * Throws if the path is not valid and returns the parent path.
+ * @throws BadArgumentsException
+ */
+ private String validatePathForCreate(String path, long sessionId)
+ throws BadArgumentsException {
+ int lastSlash = path.lastIndexOf('/');
+ if (lastSlash == -1 || path.indexOf('\0') != -1 || failCreate) {
+ LOG.info("Invalid path %s with session 0x%s",
+ path, Long.toHexString(sessionId));
+ throw new KeeperException.BadArgumentsException(path);
+ }
+ return path.substring(0, lastSlash);
+ }
+
+ /**
+ * This method will be called inside the ProcessRequestThread, which is a
+ * singleton, so there will be a single thread calling this code.
+ *
+ * @param type
+ * @param zxid
+ * @param request
+ * @param record
+ */
+ protected void pRequest2Txn(int type, long zxid, Request request,
+ Record record, boolean deserialize)
+ throws KeeperException, IOException, RequestProcessorException
+ {
+ request.setHdr(new TxnHeader(request.sessionId, request.cxid, zxid,
+ Time.currentWallTime(), type));
+
+ switch (type) {
+ case OpCode.create:
+ case OpCode.create2:
+ case OpCode.createTTL:
+ case OpCode.createContainer: {
+ pRequest2TxnCreate(type, request, record, deserialize);
+ break;
+ }
+ case OpCode.deleteContainer: {
+ String path = new String(request.request.array());
+ String parentPath = getParentPathAndValidate(path);
+ ChangeRecord parentRecord = getRecordForPath(parentPath);
+ ChangeRecord nodeRecord = getRecordForPath(path);
+ if (nodeRecord.childCount > 0) {
+ throw new KeeperException.NotEmptyException(path);
+ }
+ if (EphemeralType.get(nodeRecord.stat.getEphemeralOwner()) == EphemeralType.NORMAL) {
+ throw new KeeperException.BadVersionException(path);
+ }
+ request.setTxn(new DeleteTxn(path));
+ parentRecord = parentRecord.duplicate(request.getHdr().getZxid());
+ parentRecord.childCount--;
+ addChangeRecord(parentRecord);
+ addChangeRecord(new ChangeRecord(request.getHdr().getZxid(), path, null, -1, null));
+ break;
+ }
+ case OpCode.delete:
+ zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
+ DeleteRequest deleteRequest = (DeleteRequest)record;
+ if(deserialize)
+ ByteBufferInputStream.byteBuffer2Record(request.request, deleteRequest);
+ String path = deleteRequest.getPath();
+ String parentPath = getParentPathAndValidate(path);
+ ChangeRecord parentRecord = getRecordForPath(parentPath);
+ ChangeRecord nodeRecord = getRecordForPath(path);
+ checkACL(zks, parentRecord.acl, ZooDefs.Perms.DELETE, request.authInfo);
+ checkAndIncVersion(nodeRecord.stat.getVersion(), deleteRequest.getVersion(), path);
+ if (nodeRecord.childCount > 0) {
+ throw new KeeperException.NotEmptyException(path);
+ }
+ request.setTxn(new DeleteTxn(path));
+ parentRecord = parentRecord.duplicate(request.getHdr().getZxid());
+ parentRecord.childCount--;
+ addChangeRecord(parentRecord);
+ addChangeRecord(new ChangeRecord(request.getHdr().getZxid(), path, null, -1, null));
+ break;
+ case OpCode.setData:
+ zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
+ SetDataRequest setDataRequest = (SetDataRequest)record;
+ if(deserialize)
+ ByteBufferInputStream.byteBuffer2Record(request.request, setDataRequest);
+ path = setDataRequest.getPath();
+ validatePath(path, request.sessionId);
+ nodeRecord = getRecordForPath(path);
+ checkACL(zks, nodeRecord.acl, ZooDefs.Perms.WRITE, request.authInfo);
+ int newVersion = checkAndIncVersion(nodeRecord.stat.getVersion(), setDataRequest.getVersion(), path);
+ request.setTxn(new SetDataTxn(path, setDataRequest.getData(), newVersion));
+ nodeRecord = nodeRecord.duplicate(request.getHdr().getZxid());
+ nodeRecord.stat.setVersion(newVersion);
+ addChangeRecord(nodeRecord);
+ break;
+ case OpCode.reconfig:
+ if (!QuorumPeerConfig.isReconfigEnabled()) {
+ LOG.error("Reconfig operation requested but reconfig feature is disabled.");
+ throw new KeeperException.ReconfigDisabledException();
+ }
+
+ if (skipACL) {
+ LOG.warn("skipACL is set, reconfig operation will skip ACL checks!");
+ }
+
+ zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
+ ReconfigRequest reconfigRequest = (ReconfigRequest)record;
+ LeaderZooKeeperServer lzks;
+ try {
+ lzks = (LeaderZooKeeperServer)zks;
+ } catch (ClassCastException e) {
+ // standalone mode - reconfiguration currently not supported
+ throw new KeeperException.UnimplementedException();
+ }
+ QuorumVerifier lastSeenQV = lzks.self.getLastSeenQuorumVerifier();
+ // check that there's no reconfig in progress
+ if (lastSeenQV.getVersion()!=lzks.self.getQuorumVerifier().getVersion()) {
+ throw new KeeperException.ReconfigInProgress();
+ }
+ long configId = reconfigRequest.getCurConfigId();
+
+ if (configId != -1 && configId!=lzks.self.getLastSeenQuorumVerifier().getVersion()){
+ String msg = "Reconfiguration from version " + configId + " failed -- last seen version is " +
+ lzks.self.getLastSeenQuorumVerifier().getVersion();
+ throw new KeeperException.BadVersionException(msg);
+ }
+
+ String newMembers = reconfigRequest.getNewMembers();
+
+ if (newMembers != null) { //non-incremental membership change
+ LOG.info("Non-incremental reconfig");
+
+ // Input may be delimited by either commas or newlines so convert to common newline separated format
+ newMembers = newMembers.replaceAll(",", "\n");
+
+ try{
+ Properties props = new Properties();
+ props.load(new StringReader(newMembers));
+ request.qv = QuorumPeerConfig.parseDynamicConfig(props, lzks.self.getElectionType(), true, false);
+ request.qv.setVersion(request.getHdr().getZxid());
+ } catch (IOException | ConfigException e) {
+ throw new KeeperException.BadArgumentsException(e.getMessage());
+ }
+ } else { //incremental change - must be a majority quorum system
+ LOG.info("Incremental reconfig");
+
+ List<String> joiningServers = null;
+ String joiningServersString = reconfigRequest.getJoiningServers();
+ if (joiningServersString != null)
+ {
+ joiningServers = StringUtils.split(joiningServersString,",");
+ }
+
+ List<String> leavingServers = null;
+ String leavingServersString = reconfigRequest.getLeavingServers();
+ if (leavingServersString != null)
+ {
+ leavingServers = StringUtils.split(leavingServersString, ",");
+ }
+
+ if (!(lastSeenQV instanceof QuorumMaj)) {
+ String msg = "Incremental reconfiguration requested but last configuration seen has a non-majority quorum system";
+ LOG.warn(msg);
+ throw new KeeperException.BadArgumentsException(msg);
+ }
+ Map<Long, QuorumServer> nextServers = new HashMap<Long, QuorumServer>(lastSeenQV.getAllMembers());
+ try {
+ if (leavingServers != null) {
+ for (String leaving: leavingServers){
+ long sid = Long.parseLong(leaving);
+ nextServers.remove(sid);
+ }
+ }
+ if (joiningServers != null) {
+ for (String joiner: joiningServers){
+ // joiner should have the following format: server.x = server_spec;client_spec
+ String[] parts = StringUtils.split(joiner, "=").toArray(new String[0]);
+ if (parts.length != 2) {
+ throw new KeeperException.BadArgumentsException("Wrong format of server string");
+ }
+ // extract server id x from first part of joiner: server.x
+ Long sid = Long.parseLong(parts[0].substring(parts[0].lastIndexOf('.') + 1));
+ QuorumServer qs = new QuorumServer(sid, parts[1]);
+ if (qs.clientAddr == null || qs.electionAddr == null || qs.addr == null) {
+ throw new KeeperException.BadArgumentsException("Wrong format of server string - each server should have 3 ports specified");
+ }
+
+ // check duplication of addresses and ports
+ for (QuorumServer nqs: nextServers.values()) {
+ if (qs.id == nqs.id) {
+ continue;
+ }
+ qs.checkAddressDuplicate(nqs);
+ }
+
+ nextServers.remove(qs.id);
+ nextServers.put(qs.id, qs);
+ }
+ }
+ } catch (ConfigException e){
+ throw new KeeperException.BadArgumentsException("Reconfiguration failed");
+ }
+ request.qv = new QuorumMaj(nextServers);
+ request.qv.setVersion(request.getHdr().getZxid());
+ }
+ if (QuorumPeerConfig.isStandaloneEnabled() && request.qv.getVotingMembers().size() < 2) {
+ String msg = "Reconfig failed - new configuration must include at least 2 followers";
+ LOG.warn(msg);
+ throw new KeeperException.BadArgumentsException(msg);
+ } else if (request.qv.getVotingMembers().size() < 1) {
+ String msg = "Reconfig failed - new configuration must include at least 1 follower";
+ LOG.warn(msg);
+ throw new KeeperException.BadArgumentsException(msg);
+ }
+
+ if (!lzks.getLeader().isQuorumSynced(request.qv)) {
+ String msg2 = "Reconfig failed - there must be a connected and synced quorum in new configuration";
+ LOG.warn(msg2);
+ throw new KeeperException.NewConfigNoQuorum();
+ }
+
+ nodeRecord = getRecordForPath(ZooDefs.CONFIG_NODE);
+ checkACL(zks, nodeRecord.acl, ZooDefs.Perms.WRITE, request.authInfo);
+ request.setTxn(new SetDataTxn(ZooDefs.CONFIG_NODE, request.qv.toString().getBytes(), -1));
+ nodeRecord = nodeRecord.duplicate(request.getHdr().getZxid());
+ nodeRecord.stat.setVersion(-1);
+ addChangeRecord(nodeRecord);
+ break;
+ case OpCode.setACL:
+ zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
+ SetACLRequest setAclRequest = (SetACLRequest)record;
+ if(deserialize)
+ ByteBufferInputStream.byteBuffer2Record(request.request, setAclRequest);
+ path = setAclRequest.getPath();
+ validatePath(path, request.sessionId);
+ List<ACL> listACL = fixupACL(path, request.authInfo, setAclRequest.getAcl());
+ nodeRecord = getRecordForPath(path);
+ checkACL(zks, nodeRecord.acl, ZooDefs.Perms.ADMIN, request.authInfo);
+ newVersion = checkAndIncVersion(nodeRecord.stat.getAversion(), setAclRequest.getVersion(), path);
+ request.setTxn(new SetACLTxn(path, listACL, newVersion));
+ nodeRecord = nodeRecord.duplicate(request.getHdr().getZxid());
+ nodeRecord.stat.setAversion(newVersion);
+ addChangeRecord(nodeRecord);
+ break;
+ case OpCode.createSession:
+ request.request.rewind();
+ int to = request.request.getInt();
+ request.setTxn(new CreateSessionTxn(to));
+ request.request.rewind();
+ if (request.isLocalSession()) {
+ // This will add to local session tracker if it is enabled
+ zks.sessionTracker.addSession(request.sessionId, to);
+ } else {
+ // Explicitly add to global session if the flag is not set
+ zks.sessionTracker.addGlobalSession(request.sessionId, to);
+ }
+ zks.setOwner(request.sessionId, request.getOwner());
+ break;
+ case OpCode.closeSession:
+ // We don't want to do this check since the session expiration thread
+ // queues up this operation without being the session owner.
+ // this request is the last of the session so it should be ok
+ //zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
+ Set<String> es = zks.getZKDatabase()
+ .getEphemerals(request.sessionId);
+ synchronized (zks.outstandingChanges) {
+ for (ChangeRecord c : zks.outstandingChanges) {
+ if (c.stat == null) {
+ // Doing a delete
+ es.remove(c.path);
+ } else if (c.stat.getEphemeralOwner() == request.sessionId) {
+ es.add(c.path);
+ }
+ }
+ for (String path2Delete : es) {
+ addChangeRecord(new ChangeRecord(request.getHdr().getZxid(), path2Delete, null, 0, null));
+ }
+
+ zks.sessionTracker.setSessionClosing(request.sessionId);
+ }
+ break;
+ case OpCode.check:
+ zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
+ CheckVersionRequest checkVersionRequest = (CheckVersionRequest)record;
+ if(deserialize)
+ ByteBufferInputStream.byteBuffer2Record(request.request, checkVersionRequest);
+ path = checkVersionRequest.getPath();
+ validatePath(path, request.sessionId);
+ nodeRecord = getRecordForPath(path);
+ checkACL(zks, nodeRecord.acl, ZooDefs.Perms.READ, request.authInfo);
+ request.setTxn(new CheckVersionTxn(path, checkAndIncVersion(nodeRecord.stat.getVersion(),
+ checkVersionRequest.getVersion(), path)));
+ break;
+ default:
+ LOG.warn("unknown type " + type);
+ break;
+ }
+ }
+
+ private void pRequest2TxnCreate(int type, Request request, Record record, boolean deserialize) throws IOException, KeeperException {
+ if (deserialize) {
+ ByteBufferInputStream.byteBuffer2Record(request.request, record);
+ }
+
+ int flags;
+ String path;
+ List<ACL> acl;
+ byte[] data;
+ long ttl;
+ if (type == OpCode.createTTL) {
+ CreateTTLRequest createTtlRequest = (CreateTTLRequest)record;
+ flags = createTtlRequest.getFlags();
+ path = createTtlRequest.getPath();
+ acl = createTtlRequest.getAcl();
+ data = createTtlRequest.getData();
+ ttl = createTtlRequest.getTtl();
+ } else {
+ CreateRequest createRequest = (CreateRequest)record;
+ flags = createRequest.getFlags();
+ path = createRequest.getPath();
+ acl = createRequest.getAcl();
+ data = createRequest.getData();
+ ttl = -1;
+ }
+ CreateMode createMode = CreateMode.fromFlag(flags);
+ validateCreateRequest(path, createMode, request, ttl);
+ String parentPath = validatePathForCreate(path, request.sessionId);
+
+ List<ACL> listACL = fixupACL(path, request.authInfo, acl);
+ ChangeRecord parentRecord = getRecordForPath(parentPath);
+
+ checkACL(zks, parentRecord.acl, ZooDefs.Perms.CREATE, request.authInfo);
+ int parentCVersion = parentRecord.stat.getCversion();
+ if (createMode.isSequential()) {
+ path = path + String.format(Locale.ENGLISH, "%010d", parentCVersion);
+ }
+ validatePath(path, request.sessionId);
+ try {
+ if (getRecordForPath(path) != null) {
+ throw new KeeperException.NodeExistsException(path);
+ }
+ } catch (KeeperException.NoNodeException e) {
+ // ignore this one
+ }
+ boolean ephemeralParent = EphemeralType.get(parentRecord.stat.getEphemeralOwner()) == EphemeralType.NORMAL;
+ if (ephemeralParent) {
+ throw new KeeperException.NoChildrenForEphemeralsException(path);
+ }
+ int newCversion = parentRecord.stat.getCversion()+1;
+ if (type == OpCode.createContainer) {
+ request.setTxn(new CreateContainerTxn(path, data, listACL, newCversion));
+ } else if (type == OpCode.createTTL) {
+ request.setTxn(new CreateTTLTxn(path, data, listACL, newCversion, ttl));
+ } else {
+ request.setTxn(new CreateTxn(path, data, listACL, createMode.isEphemeral(),
+ newCversion));
+ }
+ StatPersisted s = new StatPersisted();
+ if (createMode.isEphemeral()) {
+ s.setEphemeralOwner(request.sessionId);
+ }
+ parentRecord = parentRecord.duplicate(request.getHdr().getZxid());
+ parentRecord.childCount++;
+ parentRecord.stat.setCversion(newCversion);
+ addChangeRecord(parentRecord);
+ addChangeRecord(new ChangeRecord(request.getHdr().getZxid(), path, s, 0, listACL));
+ }
+
+ private void validatePath(String path, long sessionId) throws BadArgumentsException {
+ try {
+ PathUtils.validatePath(path);
+ } catch(IllegalArgumentException ie) {
+ LOG.info("Invalid path {} with session 0x{}, reason: {}",
+ path, Long.toHexString(sessionId), ie.getMessage());
+ throw new BadArgumentsException(path);
+ }
+ }
+
+ private String getParentPathAndValidate(String path)
+ throws BadArgumentsException {
+ int lastSlash = path.lastIndexOf('/');
+ if (lastSlash == -1 || path.indexOf('\0') != -1
+ || zks.getZKDatabase().isSpecialPath(path)) {
+ throw new BadArgumentsException(path);
+ }
+ return path.substring(0, lastSlash);
+ }
+
+ private static int checkAndIncVersion(int currentVersion, int expectedVersion, String path)
+ throws KeeperException.BadVersionException {
+ if (expectedVersion != -1 && expectedVersion != currentVersion) {
+ throw new KeeperException.BadVersionException(path);
+ }
+ return currentVersion + 1;
+ }
+
+ /**
+ * This method will be called inside the ProcessRequestThread, which is a
+ * singleton, so there will be a single thread calling this code.
+ *
+ * @param request
+ */
+ protected void pRequest(Request request) throws RequestProcessorException {
+ // LOG.info("Prep>>> cxid = " + request.cxid + " type = " +
+ // request.type + " id = 0x" + Long.toHexString(request.sessionId));
+ request.setHdr(null);
+ request.setTxn(null);
+
+ try {
+ switch (request.type) {
+ case OpCode.createContainer:
+ case OpCode.create:
+ case OpCode.create2:
+ CreateRequest create2Request = new CreateRequest();
+ pRequest2Txn(request.type, zks.getNextZxid(), request, create2Request, true);
+ break;
+ case OpCode.createTTL:
+ CreateTTLRequest createTtlRequest = new CreateTTLRequest();
+ pRequest2Txn(request.type, zks.getNextZxid(), request, createTtlRequest, true);
+ break;
+ case OpCode.deleteContainer:
+ case OpCode.delete:
+ DeleteRequest deleteRequest = new DeleteRequest();
+ pRequest2Txn(request.type, zks.getNextZxid(), request, deleteRequest, true);
+ break;
+ case OpCode.setData:
+ SetDataRequest setDataRequest = new SetDataRequest();
+ pRequest2Txn(request.type, zks.getNextZxid(), request, setDataRequest, true);
+ break;
+ case OpCode.reconfig:
+ ReconfigRequest reconfigRequest = new ReconfigRequest();
+ ByteBufferInputStream.byteBuffer2Record(request.request, reconfigRequest);
+ pRequest2Txn(request.type, zks.getNextZxid(), request, reconfigRequest, true);
+ break;
+ case OpCode.setACL:
+ SetACLRequest setAclRequest = new SetACLRequest();
+ pRequest2Txn(request.type, zks.getNextZxid(), request, setAclRequest, true);
+ break;
+ case OpCode.check:
+ CheckVersionRequest checkRequest = new CheckVersionRequest();
+ pRequest2Txn(request.type, zks.getNextZxid(), request, checkRequest, true);
+ break;
+ case OpCode.multi:
+ MultiTransactionRecord multiRequest = new MultiTransactionRecord();
+ try {
+ ByteBufferInputStream.byteBuffer2Record(request.request, multiRequest);
+ } catch(IOException e) {
+ request.setHdr(new TxnHeader(request.sessionId, request.cxid, zks.getNextZxid(),
+ Time.currentWallTime(), OpCode.multi));
+ throw e;
+ }
+ List<Txn> txns = new ArrayList<Txn>();
+ //Each op in a multi-op must have the same zxid!
+ long zxid = zks.getNextZxid();
+ KeeperException ke = null;
+
+ //Store off current pending change records in case we need to rollback
+ Map<String, ChangeRecord> pendingChanges = getPendingChanges(multiRequest);
+
+ for(Op op: multiRequest) {
+ Record subrequest = op.toRequestRecord();
+ int type;
+ Record txn;
+
+ /* If we've already failed one of the ops, don't bother
+ * trying the rest as we know it's going to fail and it
+ * would be confusing in the logfiles.
+ */
+ if (ke != null) {
+ type = OpCode.error;
+ txn = new ErrorTxn(Code.RUNTIMEINCONSISTENCY.intValue());
+ }
+
+ /* Prep the request and convert to a Txn */
+ else {
+ try {
+ pRequest2Txn(op.getType(), zxid, request, subrequest, false);
+ type = request.getHdr().getType();
Review comment:
*NULL_DEREFERENCE:* object returned by `request.getHdr()` could be null and is dereferenced at line 807.
(at-me [in a reply](https://docs.muse.dev/docs/talk-to-muse/) with `help` or `ignore`)
##########
File path: zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServer.java
##########
@@ -0,0 +1,1255 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.server;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.PrintWriter;
+import java.nio.ByteBuffer;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.security.sasl.SaslException;
+
+import org.apache.jute.BinaryInputArchive;
+import org.apache.jute.BinaryOutputArchive;
+import org.apache.jute.Record;
+import org.apache.zookeeper.Environment;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.KeeperException.Code;
+import org.apache.zookeeper.KeeperException.SessionExpiredException;
+import org.apache.zookeeper.ZooDefs.OpCode;
+import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.data.Id;
+import org.apache.zookeeper.data.StatPersisted;
+import org.apache.zookeeper.jmx.MBeanRegistry;
+import org.apache.zookeeper.proto.AuthPacket;
+import org.apache.zookeeper.proto.ConnectRequest;
+import org.apache.zookeeper.proto.ConnectResponse;
+import org.apache.zookeeper.proto.GetSASLRequest;
+import org.apache.zookeeper.proto.ReplyHeader;
+import org.apache.zookeeper.proto.RequestHeader;
+import org.apache.zookeeper.proto.SetSASLResponse;
+import org.apache.zookeeper.server.DataTree.ProcessTxnResult;
+import org.apache.zookeeper.server.RequestProcessor.RequestProcessorException;
+import org.apache.zookeeper.server.ServerCnxn.CloseRequestException;
+import org.apache.zookeeper.server.SessionTracker.Session;
+import org.apache.zookeeper.server.SessionTracker.SessionExpirer;
+import org.apache.zookeeper.server.auth.AuthenticationProvider;
+import org.apache.zookeeper.server.auth.ProviderRegistry;
+import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
+import org.apache.zookeeper.server.quorum.ReadOnlyZooKeeperServer;
+import org.apache.zookeeper.txn.CreateSessionTxn;
+import org.apache.zookeeper.txn.TxnHeader;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * This class implements a simple standalone ZooKeeperServer. It sets up the
+ * following chain of RequestProcessors to process requests:
+ * PrepRequestProcessor -> SyncRequestProcessor -> FinalRequestProcessor
+ */
+public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
+ protected static final Logger LOG;
+
+ static {
+ LOG = LoggerFactory.getLogger(ZooKeeperServer.class);
+
+ Environment.logEnv("Server environment:", LOG);
+ }
+
+ protected ZooKeeperServerBean jmxServerBean;
+ protected DataTreeBean jmxDataTreeBean;
+
+ public static final int DEFAULT_TICK_TIME = 3000;
+ protected int tickTime = DEFAULT_TICK_TIME;
+ /** value of -1 indicates unset, use default */
+ protected int minSessionTimeout = -1;
+ /** value of -1 indicates unset, use default */
+ protected int maxSessionTimeout = -1;
+ protected SessionTracker sessionTracker;
+ private FileTxnSnapLog txnLogFactory = null;
+ private ZKDatabase zkDb;
+ private final AtomicLong hzxid = new AtomicLong(0);
+ public final static Exception ok = new Exception("No prob");
+ protected RequestProcessor firstProcessor;
+ protected volatile State state = State.INITIAL;
+
+ protected enum State {
+ INITIAL, RUNNING, SHUTDOWN, ERROR
+ }
+
+ /**
+ * This is the secret that we use to generate passwords. For the moment,
+ * it's more of a checksum that's used in reconnection, which carries no
+ * security weight, and is treated internally as if it carries no
+ * security weight.
+ */
+ static final private long superSecret = 0XB3415C00L;
+
+ private final AtomicInteger requestsInProcess = new AtomicInteger(0);
+ final Deque<ChangeRecord> outstandingChanges = new ArrayDeque<>();
+ // this data structure must be accessed under the outstandingChanges lock
+ final HashMap<String, ChangeRecord> outstandingChangesForPath =
+ new HashMap<String, ChangeRecord>();
+
+ protected ServerCnxnFactory serverCnxnFactory;
+ protected ServerCnxnFactory secureServerCnxnFactory;
+
+ private final ServerStats serverStats;
+ private final ZooKeeperServerListener listener;
+ private ZooKeeperServerShutdownHandler zkShutdownHandler;
+ private volatile int createSessionTrackerServerId = 1;
+
+ void removeCnxn(ServerCnxn cnxn) {
+ zkDb.removeCnxn(cnxn);
+ }
+
+ /**
+ * Creates a ZooKeeperServer instance. Nothing is setup, use the setX
+ * methods to prepare the instance (eg datadir, datalogdir, ticktime,
+ * builder, etc...)
+ *
+ * @throws IOException
+ */
+ public ZooKeeperServer() {
+ serverStats = new ServerStats(this);
+ listener = new ZooKeeperServerListenerImpl(this);
+ }
+
+ /**
+ * Creates a ZooKeeperServer instance. It sets everything up, but doesn't
+ * actually start listening for clients until run() is invoked.
+ *
+ * @param dataDir the directory to put the data
+ */
+ public ZooKeeperServer(FileTxnSnapLog txnLogFactory, int tickTime,
+ int minSessionTimeout, int maxSessionTimeout, ZKDatabase zkDb) {
+ serverStats = new ServerStats(this);
+ this.txnLogFactory = txnLogFactory;
+ this.txnLogFactory.setServerStats(this.serverStats);
+ this.zkDb = zkDb;
+ this.tickTime = tickTime;
+ setMinSessionTimeout(minSessionTimeout);
+ setMaxSessionTimeout(maxSessionTimeout);
+ listener = new ZooKeeperServerListenerImpl(this);
+ LOG.info("Created server with tickTime " + tickTime
+ + " minSessionTimeout " + getMinSessionTimeout()
+ + " maxSessionTimeout " + getMaxSessionTimeout()
+ + " datadir " + txnLogFactory.getDataDir()
+ + " snapdir " + txnLogFactory.getSnapDir());
+ }
+
+ /**
+ * creates a zookeeperserver instance.
+ * @param txnLogFactory the file transaction snapshot logging class
+ * @param tickTime the ticktime for the server
+ * @throws IOException
+ */
+ public ZooKeeperServer(FileTxnSnapLog txnLogFactory, int tickTime)
+ throws IOException {
+ this(txnLogFactory, tickTime, -1, -1, new ZKDatabase(txnLogFactory));
+ }
+
+ public ServerStats serverStats() {
+ return serverStats;
+ }
+
+ public void dumpConf(PrintWriter pwriter) {
+ pwriter.print("clientPort=");
+ pwriter.println(getClientPort());
+ pwriter.print("secureClientPort=");
+ pwriter.println(getSecureClientPort());
+ pwriter.print("dataDir=");
+ pwriter.println(zkDb.snapLog.getSnapDir().getAbsolutePath());
+ pwriter.print("dataDirSize=");
+ pwriter.println(getDataDirSize());
+ pwriter.print("dataLogDir=");
+ pwriter.println(zkDb.snapLog.getDataDir().getAbsolutePath());
+ pwriter.print("dataLogSize=");
+ pwriter.println(getLogDirSize());
+ pwriter.print("tickTime=");
+ pwriter.println(getTickTime());
+ pwriter.print("maxClientCnxns=");
+ pwriter.println(getMaxClientCnxnsPerHost());
+ pwriter.print("minSessionTimeout=");
+ pwriter.println(getMinSessionTimeout());
+ pwriter.print("maxSessionTimeout=");
+ pwriter.println(getMaxSessionTimeout());
+
+ pwriter.print("serverId=");
+ pwriter.println(getServerId());
+ }
+
+ public ZooKeeperServerConf getConf() {
+ return new ZooKeeperServerConf
+ (getClientPort(),
+ zkDb.snapLog.getSnapDir().getAbsolutePath(),
+ zkDb.snapLog.getDataDir().getAbsolutePath(),
+ getTickTime(),
+ getMaxClientCnxnsPerHost(),
+ getMinSessionTimeout(),
+ getMaxSessionTimeout(),
+ getServerId());
+ }
+
+ /**
+ * This constructor is for backward compatibility with the existing unit
+ * test code.
+ * It defaults to FileLogProvider persistence provider.
+ */
+ public ZooKeeperServer(File snapDir, File logDir, int tickTime)
+ throws IOException {
+ this( new FileTxnSnapLog(snapDir, logDir),
+ tickTime);
+ }
+
+ /**
+ * Default constructor, relies on the config for its argument values
+ *
+ * @throws IOException
+ */
+ public ZooKeeperServer(FileTxnSnapLog txnLogFactory)
+ throws IOException
+ {
+ this(txnLogFactory, DEFAULT_TICK_TIME, -1, -1, new ZKDatabase(txnLogFactory));
+ }
+
+ /**
+ * get the zookeeper database for this server
+ * @return the zookeeper database for this server
+ */
+ public ZKDatabase getZKDatabase() {
+ return this.zkDb;
+ }
+
+ /**
+ * set the zkdatabase for this zookeeper server
+ * @param zkDb
+ */
+ public void setZKDatabase(ZKDatabase zkDb) {
+ this.zkDb = zkDb;
+ }
+
+ /**
+ * Restore sessions and data
+ */
+ public void loadData() throws IOException, InterruptedException {
+ /*
+ * When a new leader starts executing Leader#lead, it
+ * invokes this method. The database, however, has been
+ * initialized before running leader election so that
+ * the server could pick its zxid for its initial vote.
+ * It does it by invoking QuorumPeer#getLastLoggedZxid.
+ * Consequently, we don't need to initialize it once more
+ * and avoid the penalty of loading it a second time. Not
+ * reloading it is particularly important for applications
+ * that host a large database.
+ *
+ * The following if block checks whether the database has
+ * been initialized or not. Note that this method is
+ * invoked by at least one other method:
+ * ZooKeeperServer#startdata.
+ *
+ * See ZOOKEEPER-1642 for more detail.
+ */
+ if(zkDb.isInitialized()){
+ setZxid(zkDb.getDataTreeLastProcessedZxid());
+ }
+ else {
+ setZxid(zkDb.loadDataBase());
+ }
+
+ // Clean up dead sessions
+ LinkedList<Long> deadSessions = new LinkedList<Long>();
+ for (Long session : zkDb.getSessions()) {
+ if (zkDb.getSessionWithTimeOuts().get(session) == null) {
+ deadSessions.add(session);
+ }
+ }
+
+ for (long session : deadSessions) {
+ // XXX: Is lastProcessedZxid really the best thing to use?
+ killSession(session, zkDb.getDataTreeLastProcessedZxid());
+ }
+
+ // Make a clean snapshot
+ takeSnapshot();
+ }
+
+ public void takeSnapshot(){
+ try {
+ txnLogFactory.save(zkDb.getDataTree(), zkDb.getSessionWithTimeOuts());
+ } catch (IOException e) {
+ LOG.error("Severe unrecoverable error, exiting", e);
+ // This is a severe error that we cannot recover from,
+ // so we need to exit
+ System.exit(10);
+ }
+ }
+
+ @Override
+ public long getDataDirSize() {
+ if (zkDb == null) {
+ return 0L;
+ }
+ File path = zkDb.snapLog.getDataDir();
+ return getDirSize(path);
+ }
+
+ @Override
+ public long getLogDirSize() {
+ if (zkDb == null) {
+ return 0L;
+ }
+ File path = zkDb.snapLog.getSnapDir();
+ return getDirSize(path);
+ }
+
+ private long getDirSize(File file) {
+ long size = 0L;
+ if (file.isDirectory()) {
+ File[] files = file.listFiles();
+ if (files != null) {
+ for (File f : files) {
+ size += getDirSize(f);
+ }
+ }
+ } else {
+ size = file.length();
+ }
+ return size;
+ }
+
+ public long getZxid() {
+ return hzxid.get();
+ }
+
+ public SessionTracker getSessionTracker() {
+ return sessionTracker;
+ }
+
+ long getNextZxid() {
+ return hzxid.incrementAndGet();
+ }
+
+ public void setZxid(long zxid) {
+ hzxid.set(zxid);
+ }
+
+ private void close(long sessionId) {
+ Request si = new Request(null, sessionId, 0, OpCode.closeSession, null, null);
+ setLocalSessionFlag(si);
+ submitRequest(si);
+ }
+
+ public void closeSession(long sessionId) {
+ LOG.info("Closing session 0x" + Long.toHexString(sessionId));
+
+ // we do not want to wait for a session close. send it as soon as we
+ // detect it!
+ close(sessionId);
+ }
+
+ protected void killSession(long sessionId, long zxid) {
+ zkDb.killSession(sessionId, zxid);
+ if (LOG.isTraceEnabled()) {
+ ZooTrace.logTraceMessage(LOG, ZooTrace.SESSION_TRACE_MASK,
+ "ZooKeeperServer --- killSession: 0x"
+ + Long.toHexString(sessionId));
+ }
+ if (sessionTracker != null) {
+ sessionTracker.removeSession(sessionId);
+ }
+ }
+
+ public void expire(Session session) {
+ long sessionId = session.getSessionId();
+ LOG.info("Expiring session 0x" + Long.toHexString(sessionId)
+ + ", timeout of " + session.getTimeout() + "ms exceeded");
+ close(sessionId);
+ }
+
+ public static class MissingSessionException extends IOException {
+ private static final long serialVersionUID = 7467414635467261007L;
+
+ public MissingSessionException(String msg) {
+ super(msg);
+ }
+ }
+
+ void touch(ServerCnxn cnxn) throws MissingSessionException {
+ if (cnxn == null) {
+ return;
+ }
+ long id = cnxn.getSessionId();
+ int to = cnxn.getSessionTimeout();
+ if (!sessionTracker.touchSession(id, to)) {
+ throw new MissingSessionException(
+ "No session with sessionid 0x" + Long.toHexString(id)
+ + " exists, probably expired and removed");
+ }
+ }
+
+ protected void registerJMX() {
+ // register with JMX
+ try {
+ jmxServerBean = new ZooKeeperServerBean(this);
+ MBeanRegistry.getInstance().register(jmxServerBean, null);
+
+ try {
+ jmxDataTreeBean = new DataTreeBean(zkDb.getDataTree());
+ MBeanRegistry.getInstance().register(jmxDataTreeBean, jmxServerBean);
+ } catch (Exception e) {
+ LOG.warn("Failed to register with JMX", e);
+ jmxDataTreeBean = null;
+ }
+ } catch (Exception e) {
+ LOG.warn("Failed to register with JMX", e);
+ jmxServerBean = null;
+ }
+ }
+
+ public void startdata()
+ throws IOException, InterruptedException {
+ //check to see if zkDb is not null
+ if (zkDb == null) {
+ zkDb = new ZKDatabase(this.txnLogFactory);
+ }
+ if (!zkDb.isInitialized()) {
+ loadData();
+ }
+ }
+
+ public synchronized void startup() {
+ if (sessionTracker == null) {
+ createSessionTracker();
+ }
+ startSessionTracker();
+ setupRequestProcessors();
+
+ registerJMX();
+
+ setState(State.RUNNING);
+ notifyAll();
+ }
+
+ protected void setupRequestProcessors() {
+ RequestProcessor finalProcessor = new FinalRequestProcessor(this);
+ RequestProcessor syncProcessor = new SyncRequestProcessor(this,
+ finalProcessor);
+ ((SyncRequestProcessor)syncProcessor).start();
+ firstProcessor = new PrepRequestProcessor(this, syncProcessor);
+ ((PrepRequestProcessor)firstProcessor).start();
+ }
+
+ public ZooKeeperServerListener getZooKeeperServerListener() {
+ return listener;
+ }
+
+ /**
+ * Change the server ID used by {@link #createSessionTracker()}. Must be called prior to
+ * {@link #startup()} being called
+ *
+ * @param newId ID to use
+ */
+ public void setCreateSessionTrackerServerId(int newId) {
+ createSessionTrackerServerId = newId;
+ }
+
+ protected void createSessionTracker() {
+ sessionTracker = new SessionTrackerImpl(this, zkDb.getSessionWithTimeOuts(),
+ tickTime, createSessionTrackerServerId, getZooKeeperServerListener());
+ }
+
+ protected void startSessionTracker() {
+ ((SessionTrackerImpl)sessionTracker).start();
+ }
+
+ /**
+ * Sets the state of ZooKeeper server. After changing the state, it notifies
+ * the server state change to a registered shutdown handler, if any.
+ * <p>
+ * The following are the server state transitions:
+ * <li>During startup the server will be in the INITIAL state.</li>
+ * <li>After successfully starting, the server sets the state to RUNNING.
+ * </li>
+ * <li>The server transitions to the ERROR state if it hits an internal
+ * error. {@link ZooKeeperServerListenerImpl} notifies any critical resource
+ * error events, e.g., SyncRequestProcessor not being able to write a txn to
+ * disk.</li>
+ * <li>During shutdown the server sets the state to SHUTDOWN, which
+ * corresponds to the server not running.</li>
+ *
+ * @param state new server state.
+ */
+ protected void setState(State state) {
+ this.state = state;
+ // Notify server state changes to the registered shutdown handler, if any.
+ if (zkShutdownHandler != null) {
+ zkShutdownHandler.handle(state);
+ } else {
+ LOG.debug("ZKShutdownHandler is not registered, so ZooKeeper server "
+ + "won't take any action on ERROR or SHUTDOWN server state changes");
+ }
+ }
+
+ /**
+ * This can be used while shutting down the server to see whether the server
+ * is already shutdown or not.
+ *
+ * @return true if the server is running or server hits an error, false
+ * otherwise.
+ */
+ protected boolean canShutdown() {
+ return state == State.RUNNING || state == State.ERROR;
+ }
+
+ /**
+ * @return true if the server is running, false otherwise.
+ */
+ public boolean isRunning() {
+ return state == State.RUNNING;
+ }
+
+ public void shutdown() {
+ shutdown(false);
+ }
+
+ /**
+ * Shut down the server instance
+ * @param fullyShutDown true if another server using the same database will not replace this one in the same process
+ */
+ public synchronized void shutdown(boolean fullyShutDown) {
+ if (!canShutdown()) {
+ LOG.debug("ZooKeeper server is not running, so not proceeding to shutdown!");
+ return;
+ }
+ LOG.info("shutting down");
+
+ // new RuntimeException("Calling shutdown").printStackTrace();
+ setState(State.SHUTDOWN);
+ // Since sessionTracker and syncThreads poll we just have to
+ // set running to false and they will detect it during the poll
+ // interval.
+ if (sessionTracker != null) {
+ sessionTracker.shutdown();
+ }
+ if (firstProcessor != null) {
+ firstProcessor.shutdown();
+ }
+
+ if (zkDb != null) {
+ if (fullyShutDown) {
+ zkDb.clear();
+ } else {
+ // else there is no need to clear the database
+ // * When a new quorum is established we can still apply the diff
+ // on top of the same zkDb data
+ // * If we fetch a new snapshot from leader, the zkDb will be
+ // cleared anyway before loading the snapshot
+ try {
+ //This will fast forward the database to the latest recorded transactions
+ zkDb.fastForwardDataBase();
+ } catch (IOException e) {
+ LOG.error("Error updating DB", e);
+ zkDb.clear();
+ }
+ }
+ }
+
+ unregisterJMX();
+ }
+
+ protected void unregisterJMX() {
+ // unregister from JMX
+ try {
+ if (jmxDataTreeBean != null) {
+ MBeanRegistry.getInstance().unregister(jmxDataTreeBean);
+ }
+ } catch (Exception e) {
+ LOG.warn("Failed to unregister with JMX", e);
+ }
+ try {
+ if (jmxServerBean != null) {
+ MBeanRegistry.getInstance().unregister(jmxServerBean);
+ }
+ } catch (Exception e) {
+ LOG.warn("Failed to unregister with JMX", e);
+ }
+ jmxServerBean = null;
+ jmxDataTreeBean = null;
+ }
+
+ public void incInProcess() {
+ requestsInProcess.incrementAndGet();
+ }
+
+ public void decInProcess() {
+ requestsInProcess.decrementAndGet();
+ }
+
+ public int getInProcess() {
+ return requestsInProcess.get();
+ }
+
+ /**
+ * This structure is used to facilitate information sharing between PrepRP
+ * and FinalRP.
+ */
+ static class ChangeRecord {
+ ChangeRecord(long zxid, String path, StatPersisted stat, int childCount,
+ List<ACL> acl) {
+ this.zxid = zxid;
+ this.path = path;
+ this.stat = stat;
+ this.childCount = childCount;
+ this.acl = acl;
+ }
+
+ long zxid;
+
+ String path;
+
+ StatPersisted stat; /* Make sure to create a new object when changing */
+
+ int childCount;
+
+ List<ACL> acl; /* Make sure to create a new object when changing */
+
+ ChangeRecord duplicate(long zxid) {
+ StatPersisted stat = new StatPersisted();
+ if (this.stat != null) {
+ DataTree.copyStatPersisted(this.stat, stat);
+ }
+ return new ChangeRecord(zxid, path, stat, childCount,
+ acl == null ? new ArrayList<ACL>() : new ArrayList<ACL>(acl));
+ }
+ }
+
+ byte[] generatePasswd(long id) {
+ Random r = new Random(id ^ superSecret);
+ byte p[] = new byte[16];
+ r.nextBytes(p);
+ return p;
+ }
+
+ protected boolean checkPasswd(long sessionId, byte[] passwd) {
+ return sessionId != 0
+ && Arrays.equals(passwd, generatePasswd(sessionId));
+ }
+
+ long createSession(ServerCnxn cnxn, byte passwd[], int timeout) {
+ if (passwd == null) {
+ // Possible since it's just deserialized from a packet on the wire.
+ passwd = new byte[0];
+ }
+ long sessionId = sessionTracker.createSession(timeout);
+ Random r = new Random(sessionId ^ superSecret);
+ r.nextBytes(passwd);
+ ByteBuffer to = ByteBuffer.allocate(4);
+ to.putInt(timeout);
+ cnxn.setSessionId(sessionId);
+ Request si = new Request(cnxn, sessionId, 0, OpCode.createSession, to, null);
+ setLocalSessionFlag(si);
+ submitRequest(si);
+ return sessionId;
+ }
+
+ /**
+ * set the owner of this session as owner
+ * @param id the session id
+ * @param owner the owner of the session
+ * @throws SessionExpiredException
+ */
+ public void setOwner(long id, Object owner) throws SessionExpiredException {
+ sessionTracker.setOwner(id, owner);
+ }
+
+ protected void revalidateSession(ServerCnxn cnxn, long sessionId,
+ int sessionTimeout) throws IOException {
+ boolean rc = sessionTracker.touchSession(sessionId, sessionTimeout);
+ if (LOG.isTraceEnabled()) {
+ ZooTrace.logTraceMessage(LOG, ZooTrace.SESSION_TRACE_MASK,
+ "Session 0x" + Long.toHexString(sessionId) +
+ " is valid: " + rc);
+ }
+ finishSessionInit(cnxn, rc);
+ }
+
+ public void reopenSession(ServerCnxn cnxn, long sessionId, byte[] passwd,
+ int sessionTimeout) throws IOException {
+ if (checkPasswd(sessionId, passwd)) {
+ revalidateSession(cnxn, sessionId, sessionTimeout);
+ } else {
+ LOG.warn("Incorrect password from " + cnxn.getRemoteSocketAddress()
+ + " for session 0x" + Long.toHexString(sessionId));
+ finishSessionInit(cnxn, false);
+ }
+ }
+
+ public void finishSessionInit(ServerCnxn cnxn, boolean valid) {
+ // register with JMX
+ try {
+ if (valid) {
+ if (serverCnxnFactory != null && serverCnxnFactory.cnxns.contains(cnxn)) {
+ serverCnxnFactory.registerConnection(cnxn);
+ } else if (secureServerCnxnFactory != null && secureServerCnxnFactory.cnxns.contains(cnxn)) {
+ secureServerCnxnFactory.registerConnection(cnxn);
+ }
+ }
+ } catch (Exception e) {
+ LOG.warn("Failed to register with JMX", e);
+ }
+
+ try {
+ ConnectResponse rsp = new ConnectResponse(0, valid ? cnxn.getSessionTimeout()
+ : 0, valid ? cnxn.getSessionId() : 0, // send 0 if session is no
+ // longer valid
+ valid ? generatePasswd(cnxn.getSessionId()) : new byte[16]);
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ BinaryOutputArchive bos = BinaryOutputArchive.getArchive(baos);
+ bos.writeInt(-1, "len");
+ rsp.serialize(bos, "connect");
+ if (!cnxn.isOldClient) {
+ bos.writeBool(
+ this instanceof ReadOnlyZooKeeperServer, "readOnly");
+ }
+ baos.close();
+ ByteBuffer bb = ByteBuffer.wrap(baos.toByteArray());
+ bb.putInt(bb.remaining() - 4).rewind();
+ cnxn.sendBuffer(bb);
+
+ if (valid) {
+ LOG.debug("Established session 0x"
+ + Long.toHexString(cnxn.getSessionId())
+ + " with negotiated timeout " + cnxn.getSessionTimeout()
+ + " for client "
+ + cnxn.getRemoteSocketAddress());
+ cnxn.enableRecv();
+ } else {
+
+ LOG.info("Invalid session 0x"
+ + Long.toHexString(cnxn.getSessionId())
+ + " for client "
+ + cnxn.getRemoteSocketAddress()
+ + ", probably expired");
+ cnxn.sendBuffer(ServerCnxnFactory.closeConn);
+ }
+
+ } catch (Exception e) {
+ LOG.warn("Exception while establishing session, closing", e);
+ cnxn.close();
+ }
+ }
+
+ public void closeSession(ServerCnxn cnxn, RequestHeader requestHeader) {
+ closeSession(cnxn.getSessionId());
+ }
+
+ public long getServerId() {
+ return 0;
+ }
+
+ /**
+ * If the underlying Zookeeper server support local session, this method
+ * will set a isLocalSession to true if a request is associated with
+ * a local session.
+ *
+ * @param si
+ */
+ protected void setLocalSessionFlag(Request si) {
+ }
+
+ public void submitRequest(Request si) {
+ if (firstProcessor == null) {
+ synchronized (this) {
+ try {
+ // Since all requests are passed to the request
+ // processor it should wait for setting up the request
+ // processor chain. The state will be updated to RUNNING
+ // after the setup.
+ while (state == State.INITIAL) {
+ wait(1000);
+ }
+ } catch (InterruptedException e) {
+ LOG.warn("Unexpected interruption", e);
+ }
+ if (firstProcessor == null || state != State.RUNNING) {
+ throw new RuntimeException("Not started");
+ }
+ }
+ }
+ try {
+ touch(si.cnxn);
+ boolean validpacket = Request.isValid(si.type);
+ if (validpacket) {
+ firstProcessor.processRequest(si);
+ if (si.cnxn != null) {
+ incInProcess();
+ }
+ } else {
+ LOG.warn("Received packet at server of unknown type " + si.type);
+ new UnimplementedRequestProcessor().processRequest(si);
+ }
+ } catch (MissingSessionException e) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Dropping request: " + e.getMessage());
+ }
+ } catch (RequestProcessorException e) {
+ LOG.error("Unable to process request:" + e.getMessage(), e);
+ }
+ }
+
+ public static int getSnapCount() {
+ String sc = System.getProperty("zookeeper.snapCount");
+ try {
+ int snapCount = Integer.parseInt(sc);
+
+ // snapCount must be 2 or more. See org.apache.zookeeper.server.SyncRequestProcessor
+ if( snapCount < 2 ) {
+ LOG.warn("SnapCount should be 2 or more. Now, snapCount is reset to 2");
+ snapCount = 2;
+ }
+ return snapCount;
+ } catch (Exception e) {
+ return 100000;
+ }
+ }
+
+ public int getGlobalOutstandingLimit() {
+ String sc = System.getProperty("zookeeper.globalOutstandingLimit");
+ int limit;
+ try {
+ limit = Integer.parseInt(sc);
+ } catch (Exception e) {
+ limit = 1000;
+ }
+ return limit;
+ }
+
+ public void setServerCnxnFactory(ServerCnxnFactory factory) {
+ serverCnxnFactory = factory;
+ }
+
+ public ServerCnxnFactory getServerCnxnFactory() {
+ return serverCnxnFactory;
+ }
+
+ public ServerCnxnFactory getSecureServerCnxnFactory() {
+ return secureServerCnxnFactory;
+ }
+
+ public void setSecureServerCnxnFactory(ServerCnxnFactory factory) {
+ secureServerCnxnFactory = factory;
+ }
+
+ /**
+ * return the last proceesed id from the
+ * datatree
+ */
+ public long getLastProcessedZxid() {
+ return zkDb.getDataTreeLastProcessedZxid();
+ }
+
+ /**
+ * return the outstanding requests
+ * in the queue, which havent been
+ * processed yet
+ */
+ public long getOutstandingRequests() {
+ return getInProcess();
+ }
+
+ /**
+ * return the total number of client connections that are alive
+ * to this server
+ */
+ public int getNumAliveConnections() {
+ int numAliveConnections = 0;
+
+ if (serverCnxnFactory != null) {
+ numAliveConnections += serverCnxnFactory.getNumAliveConnections();
+ }
+
+ if (secureServerCnxnFactory != null) {
+ numAliveConnections += secureServerCnxnFactory.getNumAliveConnections();
+ }
+
+ return numAliveConnections;
+ }
+
+ /**
+ * trunccate the log to get in sync with others
+ * if in a quorum
+ * @param zxid the zxid that it needs to get in sync
+ * with others
+ * @throws IOException
+ */
+ public void truncateLog(long zxid) throws IOException {
+ this.zkDb.truncateLog(zxid);
+ }
+
+ public int getTickTime() {
+ return tickTime;
+ }
+
+ public void setTickTime(int tickTime) {
+ LOG.info("tickTime set to " + tickTime);
+ this.tickTime = tickTime;
+ }
+
+ public int getMinSessionTimeout() {
+ return minSessionTimeout;
+ }
+
+ public void setMinSessionTimeout(int min) {
+ this.minSessionTimeout = min == -1 ? tickTime * 2 : min;
+ LOG.info("minSessionTimeout set to {}", this.minSessionTimeout);
+ }
+
+ public int getMaxSessionTimeout() {
+ return maxSessionTimeout;
+ }
+
+ public void setMaxSessionTimeout(int max) {
+ this.maxSessionTimeout = max == -1 ? tickTime * 20 : max;
+ LOG.info("maxSessionTimeout set to {}", this.maxSessionTimeout);
+ }
+
+ public int getClientPort() {
+ return serverCnxnFactory != null ? serverCnxnFactory.getLocalPort() : -1;
+ }
+
+ public int getSecureClientPort() {
+ return secureServerCnxnFactory != null ? secureServerCnxnFactory.getLocalPort() : -1;
+ }
+
+ /** Maximum number of connections allowed from particular host (ip) */
+ public int getMaxClientCnxnsPerHost() {
+ if (serverCnxnFactory != null) {
+ return serverCnxnFactory.getMaxClientCnxnsPerHost();
+ }
+ if (secureServerCnxnFactory != null) {
+ return secureServerCnxnFactory.getMaxClientCnxnsPerHost();
+ }
+ return -1;
+ }
+
+ public void setTxnLogFactory(FileTxnSnapLog txnLog) {
+ this.txnLogFactory = txnLog;
+ }
+
+ public FileTxnSnapLog getTxnLogFactory() {
+ return this.txnLogFactory;
+ }
+
+ /**
+ * Returns the elapsed sync of time of transaction log in milliseconds.
+ */
+ public long getTxnLogElapsedSyncTime() {
+ return txnLogFactory.getTxnLogElapsedSyncTime();
+ }
+
+ public String getState() {
+ return "standalone";
+ }
+
+ public void dumpEphemerals(PrintWriter pwriter) {
+ zkDb.dumpEphemerals(pwriter);
+ }
+
+ public Map<Long, Set<String>> getEphemerals() {
+ return zkDb.getEphemerals();
+ }
+
+ public void processConnectRequest(ServerCnxn cnxn, ByteBuffer incomingBuffer) throws IOException {
+ BinaryInputArchive bia = BinaryInputArchive.getArchive(new ByteBufferInputStream(incomingBuffer));
+ ConnectRequest connReq = new ConnectRequest();
+ connReq.deserialize(bia, "connect");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Session establishment request from client "
+ + cnxn.getRemoteSocketAddress()
+ + " client's lastZxid is 0x"
+ + Long.toHexString(connReq.getLastZxidSeen()));
+ }
+ boolean readOnly = false;
+ try {
+ readOnly = bia.readBool("readOnly");
+ cnxn.isOldClient = false;
+ } catch (IOException e) {
+ // this is ok -- just a packet from an old client which
+ // doesn't contain readOnly field
+ LOG.warn("Connection request from old client "
+ + cnxn.getRemoteSocketAddress()
+ + "; will be dropped if server is in r-o mode");
+ }
+ if (!readOnly && this instanceof ReadOnlyZooKeeperServer) {
+ String msg = "Refusing session request for not-read-only client "
+ + cnxn.getRemoteSocketAddress();
+ LOG.info(msg);
+ throw new CloseRequestException(msg);
+ }
+ if (connReq.getLastZxidSeen() > zkDb.dataTree.lastProcessedZxid) {
+ String msg = "Refusing session request for client "
+ + cnxn.getRemoteSocketAddress()
+ + " as it has seen zxid 0x"
+ + Long.toHexString(connReq.getLastZxidSeen())
+ + " our last zxid is 0x"
+ + Long.toHexString(getZKDatabase().getDataTreeLastProcessedZxid())
+ + " client must try another server";
+
+ LOG.info(msg);
+ throw new CloseRequestException(msg);
+ }
+ int sessionTimeout = connReq.getTimeOut();
+ byte passwd[] = connReq.getPasswd();
+ int minSessionTimeout = getMinSessionTimeout();
+ if (sessionTimeout < minSessionTimeout) {
+ sessionTimeout = minSessionTimeout;
+ }
+ int maxSessionTimeout = getMaxSessionTimeout();
+ if (sessionTimeout > maxSessionTimeout) {
+ sessionTimeout = maxSessionTimeout;
+ }
+ cnxn.setSessionTimeout(sessionTimeout);
+ // We don't want to receive any packets until we are sure that the
+ // session is setup
+ cnxn.disableRecv();
+ long sessionId = connReq.getSessionId();
+ if (sessionId == 0) {
+ long id = createSession(cnxn, passwd, sessionTimeout);
+ LOG.debug("Client attempting to establish new session:" +
+ " session = 0x{}, zxid = 0x{}, timeout = {}, address = {}",
+ Long.toHexString(id),
+ Long.toHexString(connReq.getLastZxidSeen()),
+ connReq.getTimeOut(),
+ cnxn.getRemoteSocketAddress());
+ } else {
+ long clientSessionId = connReq.getSessionId();
+ LOG.debug("Client attempting to renew session:" +
+ " session = 0x{}, zxid = 0x{}, timeout = {}, address = {}",
+ Long.toHexString(clientSessionId),
+ Long.toHexString(connReq.getLastZxidSeen()),
+ connReq.getTimeOut(),
+ cnxn.getRemoteSocketAddress());
+ if (serverCnxnFactory != null) {
+ serverCnxnFactory.closeSession(sessionId);
+ }
+ if (secureServerCnxnFactory != null) {
+ secureServerCnxnFactory.closeSession(sessionId);
+ }
+ cnxn.setSessionId(sessionId);
+ reopenSession(cnxn, sessionId, passwd, sessionTimeout);
+ }
+ }
+
+ public boolean shouldThrottle(long outStandingCount) {
+ if (getGlobalOutstandingLimit() < getInProcess()) {
+ return outStandingCount > 0;
+ }
+ return false;
+ }
+
+ public void processPacket(ServerCnxn cnxn, ByteBuffer incomingBuffer) throws IOException {
+ // We have the request, now process and setup for next
+ InputStream bais = new ByteBufferInputStream(incomingBuffer);
+ BinaryInputArchive bia = BinaryInputArchive.getArchive(bais);
+ RequestHeader h = new RequestHeader();
+ h.deserialize(bia, "header");
+ // Through the magic of byte buffers, txn will not be
+ // pointing
+ // to the start of the txn
+ incomingBuffer = incomingBuffer.slice();
+ if (h.getType() == OpCode.auth) {
+ LOG.info("got auth packet " + cnxn.getRemoteSocketAddress());
+ AuthPacket authPacket = new AuthPacket();
+ ByteBufferInputStream.byteBuffer2Record(incomingBuffer, authPacket);
+ String scheme = authPacket.getScheme();
+ AuthenticationProvider ap = ProviderRegistry.getProvider(scheme);
+ Code authReturn = KeeperException.Code.AUTHFAILED;
+ if(ap != null) {
+ try {
+ authReturn = ap.handleAuthentication(cnxn, authPacket.getAuth());
+ } catch(RuntimeException e) {
+ LOG.warn("Caught runtime exception from AuthenticationProvider: " + scheme + " due to " + e);
+ authReturn = KeeperException.Code.AUTHFAILED;
+ }
+ }
+ if (authReturn == KeeperException.Code.OK) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Authentication succeeded for scheme: " + scheme);
+ }
+ LOG.info("auth success " + cnxn.getRemoteSocketAddress());
+ ReplyHeader rh = new ReplyHeader(h.getXid(), 0,
+ KeeperException.Code.OK.intValue());
+ cnxn.sendResponse(rh, null, null);
+ } else {
+ if (ap == null) {
+ LOG.warn("No authentication provider for scheme: "
+ + scheme + " has "
+ + ProviderRegistry.listProviders());
+ } else {
+ LOG.warn("Authentication failed for scheme: " + scheme);
+ }
+ // send a response...
+ ReplyHeader rh = new ReplyHeader(h.getXid(), 0,
+ KeeperException.Code.AUTHFAILED.intValue());
+ cnxn.sendResponse(rh, null, null);
+ // ... and close connection
+ cnxn.sendBuffer(ServerCnxnFactory.closeConn);
+ cnxn.disableRecv();
+ }
+ return;
+ } else {
+ if (h.getType() == OpCode.sasl) {
+ Record rsp = processSasl(incomingBuffer,cnxn);
+ ReplyHeader rh = new ReplyHeader(h.getXid(), 0, KeeperException.Code.OK.intValue());
+ cnxn.sendResponse(rh,rsp, "response"); // not sure about 3rd arg..what is it?
+ return;
+ }
+ else {
+ Request si = new Request(cnxn, cnxn.getSessionId(), h.getXid(),
+ h.getType(), incomingBuffer, cnxn.getAuthInfo());
+ si.setOwner(ServerCnxn.me);
+ // Always treat packet from the client as a possible
+ // local request.
+ setLocalSessionFlag(si);
+ submitRequest(si);
+ }
+ }
+ cnxn.incrOutstandingRequests(h);
+ }
+
+ private Record processSasl(ByteBuffer incomingBuffer, ServerCnxn cnxn) throws IOException {
+ LOG.debug("Responding to client SASL token.");
+ GetSASLRequest clientTokenRecord = new GetSASLRequest();
+ ByteBufferInputStream.byteBuffer2Record(incomingBuffer,clientTokenRecord);
+ byte[] clientToken = clientTokenRecord.getToken();
+ LOG.debug("Size of client SASL token: " + clientToken.length);
Review comment:
*NULL_DEREFERENCE:* object `clientToken` last assigned on line 1158 could be null and is dereferenced at line 1159.
(at-me [in a reply](https://docs.muse.dev/docs/talk-to-muse/) with `help` or `ignore`)
##########
File path: zookeeper-server/src/main/java/org/apache/zookeeper/server/persistence/FileTxnLog.java
##########
@@ -0,0 +1,746 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.zookeeper.server.persistence;
+
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
+import java.io.Closeable;
+import java.io.EOFException;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.FilterInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.RandomAccessFile;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.zip.Adler32;
+import java.util.zip.Checksum;
+
+import org.apache.jute.BinaryInputArchive;
+import org.apache.jute.BinaryOutputArchive;
+import org.apache.jute.InputArchive;
+import org.apache.jute.OutputArchive;
+import org.apache.jute.Record;
+import org.apache.zookeeper.server.ServerStats;
+import org.apache.zookeeper.server.util.SerializeUtils;
+import org.apache.zookeeper.txn.TxnHeader;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This class implements the TxnLog interface. It provides api's
+ * to access the txnlogs and add entries to it.
+ * <p>
+ * The format of a Transactional log is as follows:
+ * <blockquote><pre>
+ * LogFile:
+ * FileHeader TxnList ZeroPad
+ *
+ * FileHeader: {
+ * magic 4bytes (ZKLG)
+ * version 4bytes
+ * dbid 8bytes
+ * }
+ *
+ * TxnList:
+ * Txn || Txn TxnList
+ *
+ * Txn:
+ * checksum Txnlen TxnHeader Record 0x42
+ *
+ * checksum: 8bytes Adler32 is currently used
+ * calculated across payload -- Txnlen, TxnHeader, Record and 0x42
+ *
+ * Txnlen:
+ * len 4bytes
+ *
+ * TxnHeader: {
+ * sessionid 8bytes
+ * cxid 4bytes
+ * zxid 8bytes
+ * time 8bytes
+ * type 4bytes
+ * }
+ *
+ * Record:
+ * See Jute definition file for details on the various record types
+ *
+ * ZeroPad:
+ * 0 padded to EOF (filled during preallocation stage)
+ * </pre></blockquote>
+ */
+public class FileTxnLog implements TxnLog, Closeable {
+
+ private static final Logger LOG;
+
+ public final static int TXNLOG_MAGIC =
+ ByteBuffer.wrap("ZKLG".getBytes()).getInt();
+
+ public final static int VERSION = 2;
+
+ public static final String LOG_FILE_PREFIX = "log";
+
+ static final String FSYNC_WARNING_THRESHOLD_MS_PROPERTY = "fsync.warningthresholdms";
+ static final String ZOOKEEPER_FSYNC_WARNING_THRESHOLD_MS_PROPERTY = "zookeeper." + FSYNC_WARNING_THRESHOLD_MS_PROPERTY;
+
+ /** Maximum time we allow for elapsed fsync before WARNing */
+ private final static long fsyncWarningThresholdMS;
+
+ static {
+ LOG = LoggerFactory.getLogger(FileTxnLog.class);
+
+ /** Local variable to read fsync.warningthresholdms into */
+ Long fsyncWarningThreshold;
+ if ((fsyncWarningThreshold = Long.getLong(ZOOKEEPER_FSYNC_WARNING_THRESHOLD_MS_PROPERTY)) == null)
+ fsyncWarningThreshold = Long.getLong(FSYNC_WARNING_THRESHOLD_MS_PROPERTY, 1000);
+ fsyncWarningThresholdMS = fsyncWarningThreshold;
+ }
+
+ long lastZxidSeen;
+ volatile BufferedOutputStream logStream = null;
+ volatile OutputArchive oa;
+ volatile FileOutputStream fos = null;
+
+ File logDir;
+ private final boolean forceSync = !System.getProperty("zookeeper.forceSync", "yes").equals("no");
+ long dbId;
+ private LinkedList<FileOutputStream> streamsToFlush =
+ new LinkedList<FileOutputStream>();
+ File logFileWrite = null;
+ private FilePadding filePadding = new FilePadding();
+
+ private ServerStats serverStats;
+
+ private volatile long syncElapsedMS = -1L;
+
+ /**
+ * constructor for FileTxnLog. Take the directory
+ * where the txnlogs are stored
+ * @param logDir the directory where the txnlogs are stored
+ */
+ public FileTxnLog(File logDir) {
+ this.logDir = logDir;
+ }
+
+ /**
+ * method to allow setting preallocate size
+ * of log file to pad the file.
+ * @param size the size to set to in bytes
+ */
+ public static void setPreallocSize(long size) {
+ FilePadding.setPreallocSize(size);
+ }
+
+ /**
+ * Setter for ServerStats to monitor fsync threshold exceed
+ * @param serverStats used to update fsyncThresholdExceedCount
+ */
+ @Override
+ public synchronized void setServerStats(ServerStats serverStats) {
+ this.serverStats = serverStats;
+ }
+
+ /**
+ * creates a checksum algorithm to be used
+ * @return the checksum used for this txnlog
+ */
+ protected Checksum makeChecksumAlgorithm(){
+ return new Adler32();
+ }
+
+ /**
+ * rollover the current log file to a new one.
+ * @throws IOException
+ */
+ public synchronized void rollLog() throws IOException {
+ if (logStream != null) {
+ this.logStream.flush();
+ this.logStream = null;
+ oa = null;
+ }
+ }
+
+ /**
+ * close all the open file handles
+ * @throws IOException
+ */
+ public synchronized void close() throws IOException {
+ if (logStream != null) {
+ logStream.close();
+ }
+ for (FileOutputStream log : streamsToFlush) {
+ log.close();
+ }
+ }
+
+ /**
+ * append an entry to the transaction log
+ * @param hdr the header of the transaction
+ * @param txn the transaction part of the entry
+ * returns true iff something appended, otw false
+ */
+ public synchronized boolean append(TxnHeader hdr, Record txn)
+ throws IOException
+ {
+ if (hdr == null) {
+ return false;
+ }
+ if (hdr.getZxid() <= lastZxidSeen) {
+ LOG.warn("Current zxid " + hdr.getZxid()
+ + " is <= " + lastZxidSeen + " for "
+ + hdr.getType());
+ } else {
+ lastZxidSeen = hdr.getZxid();
+ }
+ if (logStream==null) {
+ if(LOG.isInfoEnabled()){
+ LOG.info("Creating new log file: " + Util.makeLogName(hdr.getZxid()));
+ }
+
+ logFileWrite = new File(logDir, Util.makeLogName(hdr.getZxid()));
+ fos = new FileOutputStream(logFileWrite);
+ logStream=new BufferedOutputStream(fos);
+ oa = BinaryOutputArchive.getArchive(logStream);
+ FileHeader fhdr = new FileHeader(TXNLOG_MAGIC,VERSION, dbId);
+ fhdr.serialize(oa, "fileheader");
+ // Make sure that the magic number is written before padding.
+ logStream.flush();
+ filePadding.setCurrentSize(fos.getChannel().position());
+ streamsToFlush.add(fos);
+ }
+ filePadding.padFile(fos.getChannel());
+ byte[] buf = Util.marshallTxnEntry(hdr, txn);
+ if (buf == null || buf.length == 0) {
+ throw new IOException("Faulty serialization for header " +
+ "and txn");
+ }
+ Checksum crc = makeChecksumAlgorithm();
+ crc.update(buf, 0, buf.length);
+ oa.writeLong(crc.getValue(), "txnEntryCRC");
+ Util.writeTxnBytes(oa, buf);
+
+ return true;
+ }
+
+ /**
+ * Find the log file that starts at, or just before, the snapshot. Return
+ * this and all subsequent logs. Results are ordered by zxid of file,
+ * ascending order.
+ * @param logDirList array of files
+ * @param snapshotZxid return files at, or before this zxid
+ * @return
+ */
+ public static File[] getLogFiles(File[] logDirList,long snapshotZxid) {
+ List<File> files = Util.sortDataDir(logDirList, LOG_FILE_PREFIX, true);
+ long logZxid = 0;
+ // Find the log file that starts before or at the same time as the
+ // zxid of the snapshot
+ for (File f : files) {
+ long fzxid = Util.getZxidFromName(f.getName(), LOG_FILE_PREFIX);
+ if (fzxid > snapshotZxid) {
+ continue;
+ }
+ // the files
+ // are sorted with zxid's
+ if (fzxid > logZxid) {
+ logZxid = fzxid;
+ }
+ }
+ List<File> v=new ArrayList<File>(5);
+ for (File f : files) {
+ long fzxid = Util.getZxidFromName(f.getName(), LOG_FILE_PREFIX);
+ if (fzxid < logZxid) {
+ continue;
+ }
+ v.add(f);
+ }
+ return v.toArray(new File[0]);
+
+ }
+
+ /**
+ * get the last zxid that was logged in the transaction logs
+ * @return the last zxid logged in the transaction logs
+ */
+ public long getLastLoggedZxid() {
+ File[] files = getLogFiles(logDir.listFiles(), 0);
+ long maxLog=files.length>0?
+ Util.getZxidFromName(files[files.length-1].getName(),LOG_FILE_PREFIX):-1;
+
+ // if a log file is more recent we must scan it to find
+ // the highest zxid
+ long zxid = maxLog;
+ TxnIterator itr = null;
+ try {
+ FileTxnLog txn = new FileTxnLog(logDir);
+ itr = txn.read(maxLog);
+ while (true) {
+ if(!itr.next())
+ break;
+ TxnHeader hdr = itr.getHeader();
+ zxid = hdr.getZxid();
+ }
+ } catch (IOException e) {
+ LOG.warn("Unexpected exception", e);
+ } finally {
+ close(itr);
+ }
+ return zxid;
+ }
+
+ private void close(TxnIterator itr) {
+ if (itr != null) {
+ try {
+ itr.close();
+ } catch (IOException ioe) {
+ LOG.warn("Error closing file iterator", ioe);
+ }
+ }
+ }
+
+ /**
+ * commit the logs. make sure that everything hits the
+ * disk
+ */
+ public synchronized void commit() throws IOException {
+ if (logStream != null) {
+ logStream.flush();
+ }
+ for (FileOutputStream log : streamsToFlush) {
+ log.flush();
+ if (forceSync) {
+ long startSyncNS = System.nanoTime();
+
+ FileChannel channel = log.getChannel();
+ channel.force(false);
+
+ syncElapsedMS = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startSyncNS);
+ if (syncElapsedMS > fsyncWarningThresholdMS) {
+ if(serverStats != null) {
+ serverStats.incrementFsyncThresholdExceedCount();
+ }
+ LOG.warn("fsync-ing the write ahead log in "
+ + Thread.currentThread().getName()
+ + " took " + syncElapsedMS
+ + "ms which will adversely effect operation latency. "
+ + "File size is " + channel.size() + " bytes. "
+ + "See the ZooKeeper troubleshooting guide");
+ }
+ }
+ }
+ while (streamsToFlush.size() > 1) {
+ streamsToFlush.removeFirst().close();
+ }
+ }
+
+ /**
+ *
+ * @return elapsed sync time of transaction log in milliseconds
+ */
+ public long getTxnLogSyncElapsedTime() {
+ return syncElapsedMS;
+ }
+
+ /**
+ * start reading all the transactions from the given zxid
+ * @param zxid the zxid to start reading transactions from
+ * @return returns an iterator to iterate through the transaction
+ * logs
+ */
+ public TxnIterator read(long zxid) throws IOException {
+ return read(zxid, true);
+ }
+
+ /**
+ * start reading all the transactions from the given zxid.
+ *
+ * @param zxid the zxid to start reading transactions from
+ * @param fastForward true if the iterator should be fast forwarded to point
+ * to the txn of a given zxid, else the iterator will point to the
+ * starting txn of a txnlog that may contain txn of a given zxid
+ * @return returns an iterator to iterate through the transaction logs
+ */
+ public TxnIterator read(long zxid, boolean fastForward) throws IOException {
+ return new FileTxnIterator(logDir, zxid, fastForward);
+ }
+
+ /**
+ * truncate the current transaction logs
+ * @param zxid the zxid to truncate the logs to
+ * @return true if successful false if not
+ */
+ public boolean truncate(long zxid) throws IOException {
+ FileTxnIterator itr = null;
+ try {
+ itr = new FileTxnIterator(this.logDir, zxid);
+ PositionInputStream input = itr.inputStream;
+ if(input == null) {
+ throw new IOException("No log files found to truncate! This could " +
+ "happen if you still have snapshots from an old setup or " +
+ "log files were deleted accidentally or dataLogDir was changed in zoo.cfg.");
+ }
+ long pos = input.getPosition();
+ // now, truncate at the current position
+ RandomAccessFile raf=new RandomAccessFile(itr.logFile,"rw");
+ raf.setLength(pos);
+ raf.close();
+ while(itr.goToNextLog()) {
+ if (!itr.logFile.delete()) {
+ LOG.warn("Unable to truncate {}", itr.logFile);
+ }
+ }
+ } finally {
+ close(itr);
+ }
+ return true;
+ }
+
+ /**
+ * read the header of the transaction file
+ * @param file the transaction file to read
+ * @return header that was read from the file
+ * @throws IOException
+ */
+ private static FileHeader readHeader(File file) throws IOException {
+ InputStream is =null;
+ try {
+ is = new BufferedInputStream(new FileInputStream(file));
+ InputArchive ia=BinaryInputArchive.getArchive(is);
+ FileHeader hdr = new FileHeader();
+ hdr.deserialize(ia, "fileheader");
+ return hdr;
+ } finally {
+ try {
+ if (is != null) is.close();
+ } catch (IOException e) {
+ LOG.warn("Ignoring exception during close", e);
+ }
+ }
+ }
+
+ /**
+ * the dbid of this transaction database
+ * @return the dbid of this database
+ */
+ public long getDbId() throws IOException {
+ FileTxnIterator itr = new FileTxnIterator(logDir, 0);
+ FileHeader fh=readHeader(itr.logFile);
+ itr.close();
+ if(fh==null)
+ throw new IOException("Unsupported Format.");
+ return fh.getDbid();
+ }
+
+ /**
+ * the forceSync value. true if forceSync is enabled, false otherwise.
+ * @return the forceSync value
+ */
+ public boolean isForceSync() {
+ return forceSync;
+ }
+
+ /**
+ * a class that keeps track of the position
+ * in the input stream. The position points to offset
+ * that has been consumed by the applications. It can
+ * wrap buffered input streams to provide the right offset
+ * for the application.
+ */
+ static class PositionInputStream extends FilterInputStream {
+ long position;
+ protected PositionInputStream(InputStream in) {
+ super(in);
+ position = 0;
+ }
+
+ @Override
+ public int read() throws IOException {
+ int rc = super.read();
+ if (rc > -1) {
+ position++;
+ }
+ return rc;
+ }
+
+ public int read(byte[] b) throws IOException {
+ int rc = super.read(b);
+ if (rc > 0) {
+ position += rc;
+ }
+ return rc;
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException {
+ int rc = super.read(b, off, len);
+ if (rc > 0) {
+ position += rc;
+ }
+ return rc;
+ }
+
+ @Override
+ public long skip(long n) throws IOException {
+ long rc = super.skip(n);
+ if (rc > 0) {
+ position += rc;
+ }
+ return rc;
+ }
+ public long getPosition() {
+ return position;
+ }
+
+ @Override
+ public boolean markSupported() {
+ return false;
+ }
+
+ @Override
+ public void mark(int readLimit) {
+ throw new UnsupportedOperationException("mark");
+ }
+
+ @Override
+ public void reset() {
+ throw new UnsupportedOperationException("reset");
+ }
+ }
+
+ /**
+ * this class implements the txnlog iterator interface
+ * which is used for reading the transaction logs
+ */
+ public static class FileTxnIterator implements TxnLog.TxnIterator {
+ File logDir;
+ long zxid;
+ TxnHeader hdr;
+ Record record;
+ File logFile;
+ InputArchive ia;
+ static final String CRC_ERROR="CRC check failed";
+
+ PositionInputStream inputStream=null;
+ //stored files is the list of files greater than
+ //the zxid we are looking for.
+ private ArrayList<File> storedFiles;
+
+ /**
+ * create an iterator over a transaction database directory
+ * @param logDir the transaction database directory
+ * @param zxid the zxid to start reading from
+ * @param fastForward true if the iterator should be fast forwarded to
+ * point to the txn of a given zxid, else the iterator will
+ * point to the starting txn of a txnlog that may contain txn of
+ * a given zxid
+ * @throws IOException
+ */
+ public FileTxnIterator(File logDir, long zxid, boolean fastForward)
+ throws IOException {
+ this.logDir = logDir;
+ this.zxid = zxid;
+ init();
+
+ if (fastForward && hdr != null) {
+ while (hdr.getZxid() < zxid) {
+ if (!next())
Review comment:
*NULL_DEREFERENCE:* object `FileTxnLog$FileTxnIterator.inputStream` last assigned on line 544 could be null and is dereferenced by call to `next()` at line 567.
(at-me [in a reply](https://docs.muse.dev/docs/talk-to-muse/) with `help` or `ignore`)
##########
File path: zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/QuorumPeerConfig.java
##########
@@ -0,0 +1,823 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.server.quorum;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.StringReader;
+import java.io.Writer;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Map.Entry;
+
+import org.apache.yetus.audience.InterfaceAudience;
+import org.apache.zookeeper.common.ClientX509Util;
+import org.apache.zookeeper.common.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.slf4j.MDC;
+import org.apache.zookeeper.common.AtomicFileWritingIdiom;
+import org.apache.zookeeper.common.AtomicFileWritingIdiom.OutputStreamStatement;
+import org.apache.zookeeper.common.AtomicFileWritingIdiom.WriterStatement;
+import org.apache.zookeeper.common.PathUtils;
+import org.apache.zookeeper.server.ZooKeeperServer;
+import org.apache.zookeeper.server.quorum.QuorumPeer.LearnerType;
+import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
+import org.apache.zookeeper.server.quorum.auth.QuorumAuth;
+import org.apache.zookeeper.server.quorum.flexible.QuorumHierarchical;
+import org.apache.zookeeper.server.quorum.flexible.QuorumMaj;
+import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
+import org.apache.zookeeper.server.util.VerifyingFileFactory;
+
+import static org.apache.zookeeper.common.NetUtils.formatInetAddr;
+
+@InterfaceAudience.Public
+public class QuorumPeerConfig {
+ private static final Logger LOG = LoggerFactory.getLogger(QuorumPeerConfig.class);
+ private static final int UNSET_SERVERID = -1;
+ public static final String nextDynamicConfigFileSuffix = ".dynamic.next";
+
+ private static boolean standaloneEnabled = true;
+ private static boolean reconfigEnabled = false;
+
+ protected InetSocketAddress clientPortAddress;
+ protected InetSocketAddress secureClientPortAddress;
+ protected boolean sslQuorum = false;
+ protected boolean shouldUsePortUnification = false;
+ protected boolean sslQuorumReloadCertFiles = false;
+ protected File dataDir;
+ protected File dataLogDir;
+ protected String dynamicConfigFileStr = null;
+ protected String configFileStr = null;
+ protected int tickTime = ZooKeeperServer.DEFAULT_TICK_TIME;
+ protected int maxClientCnxns = 60;
+ /** defaults to -1 if not set explicitly */
+ protected int minSessionTimeout = -1;
+ /** defaults to -1 if not set explicitly */
+ protected int maxSessionTimeout = -1;
+ protected boolean localSessionsEnabled = false;
+ protected boolean localSessionsUpgradingEnabled = false;
+
+ protected int initLimit;
+ protected int syncLimit;
+ protected int electionAlg = 3;
+ protected int electionPort = 2182;
+ protected boolean quorumListenOnAllIPs = false;
+
+ protected long serverId = UNSET_SERVERID;
+
+ protected QuorumVerifier quorumVerifier = null, lastSeenQuorumVerifier = null;
+ protected int snapRetainCount = 3;
+ protected int purgeInterval = 0;
+ protected boolean syncEnabled = true;
+
+ protected LearnerType peerType = LearnerType.PARTICIPANT;
+
+ /**
+ * Configurations for the quorumpeer-to-quorumpeer sasl authentication
+ */
+ protected boolean quorumServerRequireSasl = false;
+ protected boolean quorumLearnerRequireSasl = false;
+ protected boolean quorumEnableSasl = false;
+ protected String quorumServicePrincipal = QuorumAuth.QUORUM_KERBEROS_SERVICE_PRINCIPAL_DEFAULT_VALUE;
+ protected String quorumLearnerLoginContext = QuorumAuth.QUORUM_LEARNER_SASL_LOGIN_CONTEXT_DFAULT_VALUE;
+ protected String quorumServerLoginContext = QuorumAuth.QUORUM_SERVER_SASL_LOGIN_CONTEXT_DFAULT_VALUE;
+ protected int quorumCnxnThreadsSize;
+
+ /**
+ * Minimum snapshot retain count.
+ * @see org.apache.zookeeper.server.PurgeTxnLog#purge(File, File, int)
+ */
+ private final int MIN_SNAP_RETAIN_COUNT = 3;
+
+ @SuppressWarnings("serial")
+ public static class ConfigException extends Exception {
+ public ConfigException(String msg) {
+ super(msg);
+ }
+ public ConfigException(String msg, Exception e) {
+ super(msg, e);
+ }
+ }
+
+ /**
+ * Parse a ZooKeeper configuration file
+ * @param path the patch of the configuration file
+ * @throws ConfigException error processing configuration
+ */
+ public void parse(String path) throws ConfigException {
+ LOG.info("Reading configuration from: " + path);
+
+ try {
+ File configFile = (new VerifyingFileFactory.Builder(LOG)
+ .warnForRelativePath()
+ .failForNonExistingPath()
+ .build()).create(path);
+
+ Properties cfg = new Properties();
+ FileInputStream in = new FileInputStream(configFile);
+ try {
+ cfg.load(in);
+ configFileStr = path;
+ } finally {
+ in.close();
+ }
+ //解析配置文件
+ parseProperties(cfg);
+ } catch (IOException e) {
+ throw new ConfigException("Error processing " + path, e);
+ } catch (IllegalArgumentException e) {
+ throw new ConfigException("Error processing " + path, e);
+ }
+
+ if (dynamicConfigFileStr!=null) {
+ try {
+ Properties dynamicCfg = new Properties();
+ FileInputStream inConfig = new FileInputStream(dynamicConfigFileStr);
+ try {
+ dynamicCfg.load(inConfig);
+ if (dynamicCfg.getProperty("version") != null) {
+ throw new ConfigException("dynamic file shouldn't have version inside");
+ }
+
+ String version = getVersionFromFilename(dynamicConfigFileStr);
+ // If there isn't any version associated with the filename,
+ // the default version is 0.
+ if (version != null) {
+ dynamicCfg.setProperty("version", version);
+ }
+ } finally {
+ inConfig.close();
+ }
+ setupQuorumPeerConfig(dynamicCfg, false);
+
+ } catch (IOException e) {
+ throw new ConfigException("Error processing " + dynamicConfigFileStr, e);
+ } catch (IllegalArgumentException e) {
+ throw new ConfigException("Error processing " + dynamicConfigFileStr, e);
+ }
+ File nextDynamicConfigFile = new File(configFileStr + nextDynamicConfigFileSuffix);
+ if (nextDynamicConfigFile.exists()) {
+ try {
+ Properties dynamicConfigNextCfg = new Properties();
+ FileInputStream inConfigNext = new FileInputStream(nextDynamicConfigFile);
+ try {
+ dynamicConfigNextCfg.load(inConfigNext);
+ } finally {
+ inConfigNext.close();
+ }
+ boolean isHierarchical = false;
+ for (Entry<Object, Object> entry : dynamicConfigNextCfg.entrySet()) {
+ String key = entry.getKey().toString().trim();
+ if (key.startsWith("group") || key.startsWith("weight")) {
+ isHierarchical = true;
+ break;
+ }
+ }
+ lastSeenQuorumVerifier = createQuorumVerifier(dynamicConfigNextCfg, isHierarchical);
+ } catch (IOException e) {
+ LOG.warn("NextQuorumVerifier is initiated to null");
+ }
+ }
+ }
+ }
+
+ // This method gets the version from the end of dynamic file name.
+ // For example, "zoo.cfg.dynamic.0" returns initial version "0".
+ // "zoo.cfg.dynamic.1001" returns version of hex number "0x1001".
+ // If a dynamic file name doesn't have any version at the end of file,
+ // e.g. "zoo.cfg.dynamic", it returns null.
+ public static String getVersionFromFilename(String filename) {
+ int i = filename.lastIndexOf('.');
+ if(i < 0 || i >= filename.length())
+ return null;
+
+ String hexVersion = filename.substring(i + 1);
+ try {
+ long version = Long.parseLong(hexVersion, 16);
+ return Long.toHexString(version);
+ } catch (NumberFormatException e) {
+ return null;
+ }
+ }
+
+ /**
+ * Parse config from a Properties.
+ * @param zkProp Properties to parse from.
+ * @throws IOException
+ * @throws ConfigException
+ */
+ public void parseProperties(Properties zkProp)
+ throws IOException, ConfigException {
+ int clientPort = 0;
+ int secureClientPort = 0;
+ String clientPortAddress = null;
+ String secureClientPortAddress = null;
+ VerifyingFileFactory vff = new VerifyingFileFactory.Builder(LOG).warnForRelativePath().build();
+ for (Entry<Object, Object> entry : zkProp.entrySet()) {
+ String key = entry.getKey().toString().trim();
+ String value = entry.getValue().toString().trim();
+ if (key.equals("dataDir")) {
+ dataDir = vff.create(value);
+ } else if (key.equals("dataLogDir")) {
+ dataLogDir = vff.create(value);
+ } else if (key.equals("clientPort")) {
+ clientPort = Integer.parseInt(value);
+ } else if (key.equals("localSessionsEnabled")) {
+ localSessionsEnabled = Boolean.parseBoolean(value);
+ } else if (key.equals("localSessionsUpgradingEnabled")) {
+ localSessionsUpgradingEnabled = Boolean.parseBoolean(value);
+ } else if (key.equals("clientPortAddress")) {
+ clientPortAddress = value.trim();
+ } else if (key.equals("secureClientPort")) {
+ secureClientPort = Integer.parseInt(value);
+ } else if (key.equals("secureClientPortAddress")){
+ secureClientPortAddress = value.trim();
+ } else if (key.equals("tickTime")) {
+ tickTime = Integer.parseInt(value);
+ } else if (key.equals("maxClientCnxns")) {
+ maxClientCnxns = Integer.parseInt(value);
+ } else if (key.equals("minSessionTimeout")) {
+ minSessionTimeout = Integer.parseInt(value);
+ } else if (key.equals("maxSessionTimeout")) {
+ maxSessionTimeout = Integer.parseInt(value);
+ } else if (key.equals("initLimit")) {
+ initLimit = Integer.parseInt(value);
+ } else if (key.equals("syncLimit")) {
+ syncLimit = Integer.parseInt(value);
+ } else if (key.equals("electionAlg")) {
+ electionAlg = Integer.parseInt(value);
+ } else if (key.equals("quorumListenOnAllIPs")) {
+ quorumListenOnAllIPs = Boolean.parseBoolean(value);
+ } else if (key.equals("peerType")) {
+ if (value.toLowerCase().equals("observer")) {
+ peerType = LearnerType.OBSERVER;
+ } else if (value.toLowerCase().equals("participant")) {
+ peerType = LearnerType.PARTICIPANT;
+ } else
+ {
+ throw new ConfigException("Unrecognised peertype: " + value);
+ }
+ } else if (key.equals( "syncEnabled" )) {
+ syncEnabled = Boolean.parseBoolean(value);
+ } else if (key.equals("dynamicConfigFile")){
+ dynamicConfigFileStr = value;
+ } else if (key.equals("autopurge.snapRetainCount")) {
+ snapRetainCount = Integer.parseInt(value);
+ } else if (key.equals("autopurge.purgeInterval")) {
+ purgeInterval = Integer.parseInt(value);
+ } else if (key.equals("standaloneEnabled")) {
+ if (value.toLowerCase().equals("true")) {
+ setStandaloneEnabled(true);
+ } else if (value.toLowerCase().equals("false")) {
+ setStandaloneEnabled(false);
+ } else {
+ throw new ConfigException("Invalid option " + value + " for standalone mode. Choose 'true' or 'false.'");
+ }
+ } else if (key.equals("reconfigEnabled")) {
+ if (value.toLowerCase().equals("true")) {
+ setReconfigEnabled(true);
+ } else if (value.toLowerCase().equals("false")) {
+ setReconfigEnabled(false);
+ } else {
+ throw new ConfigException("Invalid option " + value + " for reconfigEnabled flag. Choose 'true' or 'false.'");
+ }
+ } else if (key.equals("sslQuorum")){
+ sslQuorum = Boolean.parseBoolean(value);
+ } else if (key.equals("portUnification")){
+ shouldUsePortUnification = Boolean.parseBoolean(value);
+ } else if (key.equals("sslQuorumReloadCertFiles")) {
+ sslQuorumReloadCertFiles = Boolean.parseBoolean(value);
+ } else if ((key.startsWith("server.") || key.startsWith("group") || key.startsWith("weight")) && zkProp.containsKey("dynamicConfigFile")) {
+ throw new ConfigException("parameter: " + key + " must be in a separate dynamic config file");
+ } else if (key.equals(QuorumAuth.QUORUM_SASL_AUTH_ENABLED)) {
+ quorumEnableSasl = Boolean.parseBoolean(value);
+ } else if (key.equals(QuorumAuth.QUORUM_SERVER_SASL_AUTH_REQUIRED)) {
+ quorumServerRequireSasl = Boolean.parseBoolean(value);
+ } else if (key.equals(QuorumAuth.QUORUM_LEARNER_SASL_AUTH_REQUIRED)) {
+ quorumLearnerRequireSasl = Boolean.parseBoolean(value);
+ } else if (key.equals(QuorumAuth.QUORUM_LEARNER_SASL_LOGIN_CONTEXT)) {
+ quorumLearnerLoginContext = value;
+ } else if (key.equals(QuorumAuth.QUORUM_SERVER_SASL_LOGIN_CONTEXT)) {
+ quorumServerLoginContext = value;
+ } else if (key.equals(QuorumAuth.QUORUM_KERBEROS_SERVICE_PRINCIPAL)) {
+ quorumServicePrincipal = value;
+ } else if (key.equals("quorum.cnxn.threads.size")) {
+ quorumCnxnThreadsSize = Integer.parseInt(value);
+ } else {
+ System.setProperty("zookeeper." + key, value);
+ }
+ }
+
+ if (!quorumEnableSasl && quorumServerRequireSasl) {
+ throw new IllegalArgumentException(
+ QuorumAuth.QUORUM_SASL_AUTH_ENABLED
+ + " is disabled, so cannot enable "
+ + QuorumAuth.QUORUM_SERVER_SASL_AUTH_REQUIRED);
+ }
+ if (!quorumEnableSasl && quorumLearnerRequireSasl) {
+ throw new IllegalArgumentException(
+ QuorumAuth.QUORUM_SASL_AUTH_ENABLED
+ + " is disabled, so cannot enable "
+ + QuorumAuth.QUORUM_LEARNER_SASL_AUTH_REQUIRED);
+ }
+ // If quorumpeer learner is not auth enabled then self won't be able to
+ // join quorum. So this condition is ensuring that the quorumpeer learner
+ // is also auth enabled while enabling quorum server require sasl.
+ if (!quorumLearnerRequireSasl && quorumServerRequireSasl) {
+ throw new IllegalArgumentException(
+ QuorumAuth.QUORUM_LEARNER_SASL_AUTH_REQUIRED
+ + " is disabled, so cannot enable "
+ + QuorumAuth.QUORUM_SERVER_SASL_AUTH_REQUIRED);
+ }
+
+ // Reset to MIN_SNAP_RETAIN_COUNT if invalid (less than 3)
+ // PurgeTxnLog.purge(File, File, int) will not allow to purge less
+ // than 3.
+ if (snapRetainCount < MIN_SNAP_RETAIN_COUNT) {
+ LOG.warn("Invalid autopurge.snapRetainCount: " + snapRetainCount
+ + ". Defaulting to " + MIN_SNAP_RETAIN_COUNT);
+ snapRetainCount = MIN_SNAP_RETAIN_COUNT;
+ }
+
+ if (dataDir == null) {
+ throw new IllegalArgumentException("dataDir is not set");
+ }
+ if (dataLogDir == null) {
+ dataLogDir = dataDir;
+ }
+
+ if (clientPort == 0) {
+ LOG.info("clientPort is not set");
+ if (clientPortAddress != null) {
+ throw new IllegalArgumentException("clientPortAddress is set but clientPort is not set");
+ }
+ } else if (clientPortAddress != null) {
+ this.clientPortAddress = new InetSocketAddress(
+ InetAddress.getByName(clientPortAddress), clientPort);
+ LOG.info("clientPortAddress is {}", formatInetAddr(this.clientPortAddress));
+ } else {
+ this.clientPortAddress = new InetSocketAddress(clientPort);
+ LOG.info("clientPortAddress is {}", formatInetAddr(this.clientPortAddress));
+ }
+
+ if (secureClientPort == 0) {
+ LOG.info("secureClientPort is not set");
+ if (secureClientPortAddress != null) {
+ throw new IllegalArgumentException("secureClientPortAddress is set but secureClientPort is not set");
+ }
+ } else if (secureClientPortAddress != null) {
+ this.secureClientPortAddress = new InetSocketAddress(
+ InetAddress.getByName(secureClientPortAddress), secureClientPort);
+ LOG.info("secureClientPortAddress is {}", formatInetAddr(this.secureClientPortAddress));
+ } else {
+ this.secureClientPortAddress = new InetSocketAddress(secureClientPort);
+ LOG.info("secureClientPortAddress is {}", formatInetAddr(this.secureClientPortAddress));
+ }
+ if (this.secureClientPortAddress != null) {
+ configureSSLAuth();
+ }
+
+ if (tickTime == 0) {
+ throw new IllegalArgumentException("tickTime is not set");
+ }
+
+ minSessionTimeout = minSessionTimeout == -1 ? tickTime * 2 : minSessionTimeout;
+ maxSessionTimeout = maxSessionTimeout == -1 ? tickTime * 20 : maxSessionTimeout;
+
+ if (minSessionTimeout > maxSessionTimeout) {
+ throw new IllegalArgumentException(
+ "minSessionTimeout must not be larger than maxSessionTimeout");
+ }
+
+ // backward compatibility - dynamic configuration in the same file as
+ // static configuration params see writeDynamicConfig()
+ if (dynamicConfigFileStr == null) {
+ setupQuorumPeerConfig(zkProp, true);
+ if (isDistributed() && isReconfigEnabled()) {
+ // we don't backup static config for standalone mode.
+ // we also don't backup if reconfig feature is disabled.
+ backupOldConfig();
+ }
+ }
+ }
+
+ /**
+ * Configure SSL authentication only if it is not configured.
+ *
+ * @throws ConfigException
+ * If authentication scheme is configured but authentication
+ * provider is not configured.
+ */
+ public static void configureSSLAuth() throws ConfigException {
+ try (ClientX509Util clientX509Util = new ClientX509Util()) {
+ String sslAuthProp = "zookeeper.authProvider." + System.getProperty(clientX509Util.getSslAuthProviderProperty(), "x509");
+ if (System.getProperty(sslAuthProp) == null) {
+ if ("zookeeper.authProvider.x509".equals(sslAuthProp)) {
+ System.setProperty("zookeeper.authProvider.x509",
+ "org.apache.zookeeper.server.auth.X509AuthenticationProvider");
+ } else {
+ throw new ConfigException("No auth provider configured for the SSL authentication scheme '"
+ + System.getProperty(clientX509Util.getSslAuthProviderProperty()) + "'.");
+ }
+ }
+ }
+ }
+
+ /**
+ * Backward compatibility -- It would backup static config file on bootup
+ * if users write dynamic configuration in "zoo.cfg".
+ */
+ private void backupOldConfig() throws IOException {
+ new AtomicFileWritingIdiom(new File(configFileStr + ".bak"), new OutputStreamStatement() {
+ @Override
+ public void write(OutputStream output) throws IOException {
+ InputStream input = null;
+ try {
+ input = new FileInputStream(new File(configFileStr));
+ byte[] buf = new byte[1024];
+ int bytesRead;
+ while ((bytesRead = input.read(buf)) > 0) {
+ output.write(buf, 0, bytesRead);
+ }
+ } finally {
+ if( input != null) {
+ input.close();
+ }
+ }
+ }
+ });
+ }
+
+ /**
+ * Writes dynamic configuration file
+ */
+ public static void writeDynamicConfig(final String dynamicConfigFilename,
+ final QuorumVerifier qv,
+ final boolean needKeepVersion)
+ throws IOException {
+
+ new AtomicFileWritingIdiom(new File(dynamicConfigFilename), new WriterStatement() {
+ @Override
+ public void write(Writer out) throws IOException {
+ Properties cfg = new Properties();
+ cfg.load( new StringReader(
+ qv.toString()));
+
+ List<String> servers = new ArrayList<String>();
+ for (Entry<Object, Object> entry : cfg.entrySet()) {
+ String key = entry.getKey().toString().trim();
+ if ( !needKeepVersion && key.startsWith("version"))
+ continue;
+
+ String value = entry.getValue().toString().trim();
+ servers.add(key
+ .concat("=")
+ .concat(value));
+ }
+
+ Collections.sort(servers);
+ out.write(StringUtils.joinStrings(servers, "\n"));
+ }
+ });
+ }
+
+ /**
+ * Edit static config file.
+ * If there are quorum information in static file, e.g. "server.X", "group",
+ * it will remove them.
+ * If it needs to erase client port information left by the old config,
+ * "eraseClientPortAddress" should be set true.
+ * It should also updates dynamic file pointer on reconfig.
+ */
+ public static void editStaticConfig(final String configFileStr,
+ final String dynamicFileStr,
+ final boolean eraseClientPortAddress)
+ throws IOException {
+ // Some tests may not have a static config file.
+ if (configFileStr == null)
+ return;
+
+ File configFile = (new VerifyingFileFactory.Builder(LOG)
+ .warnForRelativePath()
+ .failForNonExistingPath()
+ .build()).create(configFileStr);
+
+ final File dynamicFile = (new VerifyingFileFactory.Builder(LOG)
+ .warnForRelativePath()
+ .failForNonExistingPath()
+ .build()).create(dynamicFileStr);
+
+ final Properties cfg = new Properties();
+ FileInputStream in = new FileInputStream(configFile);
+ try {
+ cfg.load(in);
+ } finally {
+ in.close();
+ }
+
+ new AtomicFileWritingIdiom(new File(configFileStr), new WriterStatement() {
+ @Override
+ public void write(Writer out) throws IOException {
+ for (Entry<Object, Object> entry : cfg.entrySet()) {
+ String key = entry.getKey().toString().trim();
+
+ if (key.startsWith("server.")
+ || key.startsWith("group")
+ || key.startsWith("weight")
+ || key.startsWith("dynamicConfigFile")
+ || (eraseClientPortAddress
+ && (key.startsWith("clientPort")
+ || key.startsWith("clientPortAddress")))) {
+ // not writing them back to static file
+ continue;
+ }
+
+ String value = entry.getValue().toString().trim();
+ out.write(key.concat("=").concat(value).concat("\n"));
+ }
+
+ // updates the dynamic file pointer
+ String dynamicConfigFilePath = PathUtils.normalizeFileSystemPath(dynamicFile.getCanonicalPath());
+ out.write("dynamicConfigFile="
+ .concat(dynamicConfigFilePath)
+ .concat("\n"));
+ }
+ });
+ }
+
+
+ public static void deleteFile(String filename){
+ if (filename == null) return;
+ File f = new File(filename);
+ if (f.exists()) {
+ try{
+ f.delete();
+ } catch (Exception e) {
+ LOG.warn("deleting " + filename + " failed");
+ }
+ }
+ }
+
+
+ private static QuorumVerifier createQuorumVerifier(Properties dynamicConfigProp, boolean isHierarchical) throws ConfigException{
+ if(isHierarchical){
+ return new QuorumHierarchical(dynamicConfigProp);
+ } else {
+ /*
+ * The default QuorumVerifier is QuorumMaj
+ */
+ //LOG.info("Defaulting to majority quorums");
+ return new QuorumMaj(dynamicConfigProp);
+ }
+ }
+
+ void setupQuorumPeerConfig(Properties prop, boolean configBackwardCompatibilityMode)
+ throws IOException, ConfigException {
+ quorumVerifier = parseDynamicConfig(prop, electionAlg, true, configBackwardCompatibilityMode);
+ setupMyId();
+ setupClientPort();
+ setupPeerType();
+ checkValidity();
+ }
+
+ /**
+ * Parse dynamic configuration file and return
+ * quorumVerifier for new configuration.
+ * @param dynamicConfigProp Properties to parse from.
+ * @throws IOException
+ * @throws ConfigException
+ */
+ public static QuorumVerifier parseDynamicConfig(Properties dynamicConfigProp, int eAlg, boolean warnings,
+ boolean configBackwardCompatibilityMode) throws IOException, ConfigException {
+ boolean isHierarchical = false;
+ for (Entry<Object, Object> entry : dynamicConfigProp.entrySet()) {
+ String key = entry.getKey().toString().trim();
+ if (key.startsWith("group") || key.startsWith("weight")) {
+ isHierarchical = true;
+ } else if (!configBackwardCompatibilityMode && !key.startsWith("server.") && !key.equals("version")){
+ LOG.info(dynamicConfigProp.toString());
+ throw new ConfigException("Unrecognised parameter: " + key);
+ }
+ }
+
+ QuorumVerifier qv = createQuorumVerifier(dynamicConfigProp, isHierarchical);
+
+ int numParticipators = qv.getVotingMembers().size();
Review comment:
*NULL_DEREFERENCE:* object returned by `qv.getVotingMembers()` could be null and is dereferenced at line 632.
(at-me [in a reply](https://docs.muse.dev/docs/talk-to-muse/) with `help` or `ignore`)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org