You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zookeeper.apache.org by an...@apache.org on 2018/10/05 12:26:26 UTC
[33/51] [partial] zookeeper git commit: ZOOKEEPER-3032: MAVEN
MIGRATION - move java server, client
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/faa7cec7/src/java/main/org/apache/zookeeper/server/UnimplementedRequestProcessor.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/UnimplementedRequestProcessor.java b/src/java/main/org/apache/zookeeper/server/UnimplementedRequestProcessor.java
deleted file mode 100644
index aa58e63..0000000
--- a/src/java/main/org/apache/zookeeper/server/UnimplementedRequestProcessor.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/**
- * 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.IOException;
-
-import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.proto.ReplyHeader;
-
-/**
- * Manages the unknown requests (i.e. unknown OpCode), by:
- * - sending back the KeeperException.UnimplementedException() error code to the client
- * - closing the connection.
- */
-public class UnimplementedRequestProcessor implements RequestProcessor {
-
- public void processRequest(Request request) throws RequestProcessorException {
- KeeperException ke = new KeeperException.UnimplementedException();
- request.setException(ke);
- ReplyHeader rh = new ReplyHeader(request.cxid, request.zxid, ke.code().intValue());
- try {
- request.cnxn.sendResponse(rh, null, "response");
- } catch (IOException e) {
- throw new RequestProcessorException("Can't send the response", e);
- }
-
- request.cnxn.sendCloseSession();
- }
-
- public void shutdown() {
- }
-}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/faa7cec7/src/java/main/org/apache/zookeeper/server/WorkerService.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/WorkerService.java b/src/java/main/org/apache/zookeeper/server/WorkerService.java
deleted file mode 100644
index f6100d5..0000000
--- a/src/java/main/org/apache/zookeeper/server/WorkerService.java
+++ /dev/null
@@ -1,243 +0,0 @@
-/**
- * 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.util.ArrayList;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.zookeeper.common.Time;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * WorkerService is a worker thread pool for running tasks and is implemented
- * using one or more ExecutorServices. A WorkerService can support assignable
- * threads, which it does by creating N separate single thread ExecutorServices,
- * or non-assignable threads, which it does by creating a single N-thread
- * ExecutorService.
- * - NIOServerCnxnFactory uses a non-assignable WorkerService because the
- * socket IO requests are order independent and allowing the
- * ExecutorService to handle thread assignment gives optimal performance.
- * - CommitProcessor uses an assignable WorkerService because requests for
- * a given session must be processed in order.
- * ExecutorService provides queue management and thread restarting, so it's
- * useful even with a single thread.
- */
-public class WorkerService {
- private static final Logger LOG =
- LoggerFactory.getLogger(WorkerService.class);
-
- private final ArrayList<ExecutorService> workers =
- new ArrayList<ExecutorService>();
-
- private final String threadNamePrefix;
- private int numWorkerThreads;
- private boolean threadsAreAssignable;
- private long shutdownTimeoutMS = 5000;
-
- private volatile boolean stopped = true;
-
- /**
- * @param name worker threads are named <name>Thread-##
- * @param numThreads number of worker threads (0 - N)
- * If 0, scheduled work is run immediately by
- * the calling thread.
- * @param useAssignableThreads whether the worker threads should be
- * individually assignable or not
- */
- public WorkerService(String name, int numThreads,
- boolean useAssignableThreads) {
- this.threadNamePrefix = (name == null ? "" : name) + "Thread";
- this.numWorkerThreads = numThreads;
- this.threadsAreAssignable = useAssignableThreads;
- start();
- }
-
- /**
- * Callers should implement a class extending WorkRequest in order to
- * schedule work with the service.
- */
- public static abstract class WorkRequest {
- /**
- * Must be implemented. Is called when the work request is run.
- */
- public abstract void doWork() throws Exception;
-
- /**
- * (Optional) If implemented, is called if the service is stopped
- * or unable to schedule the request.
- */
- public void cleanup() {
- }
- }
-
- /**
- * Schedule work to be done. If a worker thread pool is not being
- * used, work is done directly by this thread. This schedule API is
- * for use with non-assignable WorkerServices. For assignable
- * WorkerServices, will always run on the first thread.
- */
- public void schedule(WorkRequest workRequest) {
- schedule(workRequest, 0);
- }
-
- /**
- * Schedule work to be done by the thread assigned to this id. Thread
- * assignment is a single mod operation on the number of threads. If a
- * worker thread pool is not being used, work is done directly by
- * this thread.
- */
- public void schedule(WorkRequest workRequest, long id) {
- if (stopped) {
- workRequest.cleanup();
- return;
- }
-
- ScheduledWorkRequest scheduledWorkRequest =
- new ScheduledWorkRequest(workRequest);
-
- // If we have a worker thread pool, use that; otherwise, do the work
- // directly.
- int size = workers.size();
- if (size > 0) {
- try {
- // make sure to map negative ids as well to [0, size-1]
- int workerNum = ((int) (id % size) + size) % size;
- ExecutorService worker = workers.get(workerNum);
- worker.execute(scheduledWorkRequest);
- } catch (RejectedExecutionException e) {
- LOG.warn("ExecutorService rejected execution", e);
- workRequest.cleanup();
- }
- } else {
- // When there is no worker thread pool, do the work directly
- // and wait for its completion
- scheduledWorkRequest.run();
- }
- }
-
- private class ScheduledWorkRequest implements Runnable {
- private final WorkRequest workRequest;
-
- ScheduledWorkRequest(WorkRequest workRequest) {
- this.workRequest = workRequest;
- }
-
- @Override
- public void run() {
- try {
- // Check if stopped while request was on queue
- if (stopped) {
- workRequest.cleanup();
- return;
- }
- workRequest.doWork();
- } catch (Exception e) {
- LOG.warn("Unexpected exception", e);
- workRequest.cleanup();
- }
- }
- }
-
- /**
- * ThreadFactory for the worker thread pool. We don't use the default
- * thread factory because (1) we want to give the worker threads easier
- * to identify names; and (2) we want to make the worker threads daemon
- * threads so they don't block the server from shutting down.
- */
- private static class DaemonThreadFactory implements ThreadFactory {
- final ThreadGroup group;
- final AtomicInteger threadNumber = new AtomicInteger(1);
- final String namePrefix;
-
- DaemonThreadFactory(String name) {
- this(name, 1);
- }
-
- DaemonThreadFactory(String name, int firstThreadNum) {
- threadNumber.set(firstThreadNum);
- SecurityManager s = System.getSecurityManager();
- group = (s != null)? s.getThreadGroup() :
- Thread.currentThread().getThreadGroup();
- namePrefix = name + "-";
- }
-
- public Thread newThread(Runnable r) {
- Thread t = new Thread(group, r,
- namePrefix + threadNumber.getAndIncrement(),
- 0);
- if (!t.isDaemon())
- t.setDaemon(true);
- if (t.getPriority() != Thread.NORM_PRIORITY)
- t.setPriority(Thread.NORM_PRIORITY);
- return t;
- }
- }
-
- public void start() {
- if (numWorkerThreads > 0) {
- if (threadsAreAssignable) {
- for(int i = 1; i <= numWorkerThreads; ++i) {
- workers.add(Executors.newFixedThreadPool(
- 1, new DaemonThreadFactory(threadNamePrefix, i)));
- }
- } else {
- workers.add(Executors.newFixedThreadPool(
- numWorkerThreads, new DaemonThreadFactory(threadNamePrefix)));
- }
- }
- stopped = false;
- }
-
- public void stop() {
- stopped = true;
-
- // Signal for graceful shutdown
- for(ExecutorService worker : workers) {
- worker.shutdown();
- }
- }
-
- public void join(long shutdownTimeoutMS) {
- // Give the worker threads time to finish executing
- long now = Time.currentElapsedTime();
- long endTime = now + shutdownTimeoutMS;
- for(ExecutorService worker : workers) {
- boolean terminated = false;
- while ((now = Time.currentElapsedTime()) <= endTime) {
- try {
- terminated = worker.awaitTermination(
- endTime - now, TimeUnit.MILLISECONDS);
- break;
- } catch (InterruptedException e) {
- // ignore
- }
- }
- if (!terminated) {
- // If we've timed out, do a hard shutdown
- worker.shutdownNow();
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/faa7cec7/src/java/main/org/apache/zookeeper/server/ZKDatabase.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/ZKDatabase.java b/src/java/main/org/apache/zookeeper/server/ZKDatabase.java
deleted file mode 100644
index 753461a..0000000
--- a/src/java/main/org/apache/zookeeper/server/ZKDatabase.java
+++ /dev/null
@@ -1,664 +0,0 @@
-/**
- * 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.File;
-import java.io.IOException;
-import java.io.PrintWriter;
-import java.util.Collection;
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;
-
-import org.apache.jute.InputArchive;
-import org.apache.jute.OutputArchive;
-import org.apache.jute.Record;
-import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.KeeperException.NoNodeException;
-import org.apache.zookeeper.Watcher;
-import org.apache.zookeeper.Watcher.WatcherType;
-import org.apache.zookeeper.ZooDefs;
-import org.apache.zookeeper.common.Time;
-import org.apache.zookeeper.data.ACL;
-import org.apache.zookeeper.data.Stat;
-import org.apache.zookeeper.server.DataTree.ProcessTxnResult;
-import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
-import org.apache.zookeeper.server.persistence.FileTxnSnapLog.PlayBackListener;
-import org.apache.zookeeper.server.persistence.TxnLog.TxnIterator;
-import org.apache.zookeeper.server.quorum.Leader;
-import org.apache.zookeeper.server.quorum.Leader.Proposal;
-import org.apache.zookeeper.server.quorum.QuorumPacket;
-import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
-import org.apache.zookeeper.server.util.SerializeUtils;
-import org.apache.zookeeper.txn.TxnHeader;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * This class maintains the in memory database of zookeeper
- * server states that includes the sessions, datatree and the
- * committed logs. It is booted up after reading the logs
- * and snapshots from the disk.
- */
-public class ZKDatabase {
-
- private static final Logger LOG = LoggerFactory.getLogger(ZKDatabase.class);
-
- /**
- * make sure on a clear you take care of
- * all these members.
- */
- protected DataTree dataTree;
- protected ConcurrentHashMap<Long, Integer> sessionsWithTimeouts;
- protected FileTxnSnapLog snapLog;
- protected long minCommittedLog, maxCommittedLog;
-
- /**
- * Default value is to use snapshot if txnlog size exceeds 1/3 the size of snapshot
- */
- public static final String SNAPSHOT_SIZE_FACTOR = "zookeeper.snapshotSizeFactor";
- public static final double DEFAULT_SNAPSHOT_SIZE_FACTOR = 0.33;
- private double snapshotSizeFactor;
-
- public static final int commitLogCount = 500;
- protected static int commitLogBuffer = 700;
- protected LinkedList<Proposal> committedLog = new LinkedList<Proposal>();
- protected ReentrantReadWriteLock logLock = new ReentrantReadWriteLock();
- volatile private boolean initialized = false;
-
- /**
- * the filetxnsnaplog that this zk database
- * maps to. There is a one to one relationship
- * between a filetxnsnaplog and zkdatabase.
- * @param snapLog the FileTxnSnapLog mapping this zkdatabase
- */
- public ZKDatabase(FileTxnSnapLog snapLog) {
- dataTree = createDataTree();
- sessionsWithTimeouts = new ConcurrentHashMap<Long, Integer>();
- this.snapLog = snapLog;
-
- try {
- snapshotSizeFactor = Double.parseDouble(
- System.getProperty(SNAPSHOT_SIZE_FACTOR,
- Double.toString(DEFAULT_SNAPSHOT_SIZE_FACTOR)));
- if (snapshotSizeFactor > 1) {
- snapshotSizeFactor = DEFAULT_SNAPSHOT_SIZE_FACTOR;
- LOG.warn("The configured {} is invalid, going to use " +
- "the default {}", SNAPSHOT_SIZE_FACTOR,
- DEFAULT_SNAPSHOT_SIZE_FACTOR);
- }
- } catch (NumberFormatException e) {
- LOG.error("Error parsing {}, using default value {}",
- SNAPSHOT_SIZE_FACTOR, DEFAULT_SNAPSHOT_SIZE_FACTOR);
- snapshotSizeFactor = DEFAULT_SNAPSHOT_SIZE_FACTOR;
- }
- LOG.info("{} = {}", SNAPSHOT_SIZE_FACTOR, snapshotSizeFactor);
- }
-
- /**
- * checks to see if the zk database has been
- * initialized or not.
- * @return true if zk database is initialized and false if not
- */
- public boolean isInitialized() {
- return initialized;
- }
-
- /**
- * clear the zkdatabase.
- * Note to developers - be careful to see that
- * the clear method does clear out all the
- * data structures in zkdatabase.
- */
- public void clear() {
- minCommittedLog = 0;
- maxCommittedLog = 0;
- /* to be safe we just create a new
- * datatree.
- */
- dataTree.shutdownWatcher();
- dataTree = createDataTree();
- sessionsWithTimeouts.clear();
- WriteLock lock = logLock.writeLock();
- try {
- lock.lock();
- committedLog.clear();
- } finally {
- lock.unlock();
- }
- initialized = false;
- }
-
- /**
- * the datatree for this zkdatabase
- * @return the datatree for this zkdatabase
- */
- public DataTree getDataTree() {
- return this.dataTree;
- }
-
- /**
- * the committed log for this zk database
- * @return the committed log for this zkdatabase
- */
- public long getmaxCommittedLog() {
- return maxCommittedLog;
- }
-
-
- /**
- * the minimum committed transaction log
- * available in memory
- * @return the minimum committed transaction
- * log available in memory
- */
- public long getminCommittedLog() {
- return minCommittedLog;
- }
- /**
- * Get the lock that controls the committedLog. If you want to get the pointer to the committedLog, you need
- * to use this lock to acquire a read lock before calling getCommittedLog()
- * @return the lock that controls the committed log
- */
- public ReentrantReadWriteLock getLogLock() {
- return logLock;
- }
-
-
- public synchronized List<Proposal> getCommittedLog() {
- ReadLock rl = logLock.readLock();
- // only make a copy if this thread isn't already holding a lock
- if(logLock.getReadHoldCount() <=0) {
- try {
- rl.lock();
- return new LinkedList<Proposal>(this.committedLog);
- } finally {
- rl.unlock();
- }
- }
- return this.committedLog;
- }
-
- /**
- * get the last processed zxid from a datatree
- * @return the last processed zxid of a datatree
- */
- public long getDataTreeLastProcessedZxid() {
- return dataTree.lastProcessedZxid;
- }
-
- /**
- * return the sessions in the datatree
- * @return the data tree sessions
- */
- public Collection<Long> getSessions() {
- return dataTree.getSessions();
- }
-
- /**
- * @return number of (global) sessions
- */
- public long getSessionCount() {
- return sessionsWithTimeouts.size();
- }
-
- /**
- * get sessions with timeouts
- * @return the hashmap of sessions with timeouts
- */
- public ConcurrentHashMap<Long, Integer> getSessionWithTimeOuts() {
- return sessionsWithTimeouts;
- }
-
- private final PlayBackListener commitProposalPlaybackListener = new PlayBackListener() {
- public void onTxnLoaded(TxnHeader hdr, Record txn){
- addCommittedProposal(hdr, txn);
- }
- };
-
- /**
- * load the database from the disk onto memory and also add
- * the transactions to the committedlog in memory.
- * @return the last valid zxid on disk
- * @throws IOException
- */
- public long loadDataBase() throws IOException {
- long startTime = Time.currentElapsedTime();
- long zxid = snapLog.restore(dataTree, sessionsWithTimeouts, commitProposalPlaybackListener);
- initialized = true;
- long loadTime = Time.currentElapsedTime() - startTime;
- ServerMetrics.DB_INIT_TIME.add(loadTime);
- LOG.info("Snapshot loaded in " + loadTime + " ms");
- return zxid;
- }
-
- /**
- * Fast forward the database adding transactions from the committed log into memory.
- * @return the last valid zxid.
- * @throws IOException
- */
- public long fastForwardDataBase() throws IOException {
- long zxid = snapLog.fastForwardFromEdits(dataTree, sessionsWithTimeouts, commitProposalPlaybackListener);
- initialized = true;
- return zxid;
- }
-
- private void addCommittedProposal(TxnHeader hdr, Record txn) {
- Request r = new Request(0, hdr.getCxid(), hdr.getType(), hdr, txn, hdr.getZxid());
- addCommittedProposal(r);
- }
-
- /**
- * maintains a list of last <i>committedLog</i>
- * or so committed requests. This is used for
- * fast follower synchronization.
- * @param request committed request
- */
- public void addCommittedProposal(Request request) {
- WriteLock wl = logLock.writeLock();
- try {
- wl.lock();
- if (committedLog.size() > commitLogCount) {
- committedLog.removeFirst();
- minCommittedLog = committedLog.getFirst().packet.getZxid();
- }
- if (committedLog.isEmpty()) {
- minCommittedLog = request.zxid;
- maxCommittedLog = request.zxid;
- }
-
- byte[] data = SerializeUtils.serializeRequest(request);
- QuorumPacket pp = new QuorumPacket(Leader.PROPOSAL, request.zxid, data, null);
- Proposal p = new Proposal();
- p.packet = pp;
- p.request = request;
- committedLog.add(p);
- maxCommittedLog = p.packet.getZxid();
- } finally {
- wl.unlock();
- }
- }
-
- public boolean isTxnLogSyncEnabled() {
- boolean enabled = snapshotSizeFactor >= 0;
- if (enabled) {
- LOG.info("On disk txn sync enabled with snapshotSizeFactor "
- + snapshotSizeFactor);
- } else {
- LOG.info("On disk txn sync disabled");
- }
- return enabled;
- }
-
- public long calculateTxnLogSizeLimit() {
- long snapSize = 0;
- try {
- File snapFile = snapLog.findMostRecentSnapshot();
- if (snapFile != null) {
- snapSize = snapFile.length();
- }
- } catch (IOException e) {
- LOG.error("Unable to get size of most recent snapshot");
- }
- return (long) (snapSize * snapshotSizeFactor);
- }
-
- /**
- * Get proposals from txnlog. Only packet part of proposal is populated.
- *
- * @param startZxid the starting zxid of the proposal
- * @param sizeLimit maximum on-disk size of txnlog to fetch
- * 0 is unlimited, negative value means disable.
- * @return list of proposal (request part of each proposal is null)
- */
- public Iterator<Proposal> getProposalsFromTxnLog(long startZxid,
- long sizeLimit) {
- if (sizeLimit < 0) {
- LOG.debug("Negative size limit - retrieving proposal via txnlog is disabled");
- return TxnLogProposalIterator.EMPTY_ITERATOR;
- }
-
- TxnIterator itr = null;
- try {
-
- itr = snapLog.readTxnLog(startZxid, false);
-
- // If we cannot guarantee that this is strictly the starting txn
- // after a given zxid, we should fail.
- if ((itr.getHeader() != null)
- && (itr.getHeader().getZxid() > startZxid)) {
- LOG.warn("Unable to find proposals from txnlog for zxid: "
- + startZxid);
- itr.close();
- return TxnLogProposalIterator.EMPTY_ITERATOR;
- }
-
- if (sizeLimit > 0) {
- long txnSize = itr.getStorageSize();
- if (txnSize > sizeLimit) {
- LOG.info("Txnlog size: " + txnSize + " exceeds sizeLimit: "
- + sizeLimit);
- itr.close();
- return TxnLogProposalIterator.EMPTY_ITERATOR;
- }
- }
- } catch (IOException e) {
- LOG.error("Unable to read txnlog from disk", e);
- try {
- if (itr != null) {
- itr.close();
- }
- } catch (IOException ioe) {
- LOG.warn("Error closing file iterator", ioe);
- }
- return TxnLogProposalIterator.EMPTY_ITERATOR;
- }
- return new TxnLogProposalIterator(itr);
- }
-
- public List<ACL> aclForNode(DataNode n) {
- return dataTree.getACL(n);
- }
- /**
- * remove a cnxn from the datatree
- * @param cnxn the cnxn to remove from the datatree
- */
- public void removeCnxn(ServerCnxn cnxn) {
- dataTree.removeCnxn(cnxn);
- }
-
- /**
- * kill a given session in the datatree
- * @param sessionId the session id to be killed
- * @param zxid the zxid of kill session transaction
- */
- public void killSession(long sessionId, long zxid) {
- dataTree.killSession(sessionId, zxid);
- }
-
- /**
- * write a text dump of all the ephemerals in the datatree
- * @param pwriter the output to write to
- */
- public void dumpEphemerals(PrintWriter pwriter) {
- dataTree.dumpEphemerals(pwriter);
- }
-
- public Map<Long, Set<String>> getEphemerals() {
- return dataTree.getEphemerals();
- }
-
- /**
- * the node count of the datatree
- * @return the node count of datatree
- */
- public int getNodeCount() {
- return dataTree.getNodeCount();
- }
-
- /**
- * the paths for ephemeral session id
- * @param sessionId the session id for which paths match to
- * @return the paths for a session id
- */
- public Set<String> getEphemerals(long sessionId) {
- return dataTree.getEphemerals(sessionId);
- }
-
- /**
- * the last processed zxid in the datatree
- * @param zxid the last processed zxid in the datatree
- */
- public void setlastProcessedZxid(long zxid) {
- dataTree.lastProcessedZxid = zxid;
- }
-
- /**
- * the process txn on the data
- * @param hdr the txnheader for the txn
- * @param txn the transaction that needs to be processed
- * @return the result of processing the transaction on this
- * datatree/zkdatabase
- */
- public ProcessTxnResult processTxn(TxnHeader hdr, Record txn) {
- return dataTree.processTxn(hdr, txn);
- }
-
- /**
- * stat the path
- * @param path the path for which stat is to be done
- * @param serverCnxn the servercnxn attached to this request
- * @return the stat of this node
- * @throws KeeperException.NoNodeException
- */
- public Stat statNode(String path, ServerCnxn serverCnxn) throws KeeperException.NoNodeException {
- return dataTree.statNode(path, serverCnxn);
- }
-
- /**
- * get the datanode for this path
- * @param path the path to lookup
- * @return the datanode for getting the path
- */
- public DataNode getNode(String path) {
- return dataTree.getNode(path);
- }
-
- /**
- * get data and stat for a path
- * @param path the path being queried
- * @param stat the stat for this path
- * @param watcher the watcher function
- * @return
- * @throws KeeperException.NoNodeException
- */
- public byte[] getData(String path, Stat stat, Watcher watcher)
- throws KeeperException.NoNodeException {
- return dataTree.getData(path, stat, watcher);
- }
-
- /**
- * set watches on the datatree
- * @param relativeZxid the relative zxid that client has seen
- * @param dataWatches the data watches the client wants to reset
- * @param existWatches the exists watches the client wants to reset
- * @param childWatches the child watches the client wants to reset
- * @param watcher the watcher function
- */
- public void setWatches(long relativeZxid, List<String> dataWatches,
- List<String> existWatches, List<String> childWatches, Watcher watcher) {
- dataTree.setWatches(relativeZxid, dataWatches, existWatches, childWatches, watcher);
- }
-
- /**
- * get acl for a path
- * @param path the path to query for acl
- * @param stat the stat for the node
- * @return the acl list for this path
- * @throws NoNodeException
- */
- public List<ACL> getACL(String path, Stat stat) throws NoNodeException {
- return dataTree.getACL(path, stat);
- }
-
- /**
- * get children list for this path
- * @param path the path of the node
- * @param stat the stat of the node
- * @param watcher the watcher function for this path
- * @return the list of children for this path
- * @throws KeeperException.NoNodeException
- */
- public List<String> getChildren(String path, Stat stat, Watcher watcher)
- throws KeeperException.NoNodeException {
- return dataTree.getChildren(path, stat, watcher);
- }
-
- /**
- * check if the path is special or not
- * @param path the input path
- * @return true if path is special and false if not
- */
- public boolean isSpecialPath(String path) {
- return dataTree.isSpecialPath(path);
- }
-
- /**
- * get the acl size of the datatree
- * @return the acl size of the datatree
- */
- public int getAclSize() {
- return dataTree.aclCacheSize();
- }
-
- /**
- * Truncate the ZKDatabase to the specified zxid
- * @param zxid the zxid to truncate zk database to
- * @return true if the truncate is successful and false if not
- * @throws IOException
- */
- public boolean truncateLog(long zxid) throws IOException {
- clear();
-
- // truncate the log
- boolean truncated = snapLog.truncateLog(zxid);
-
- if (!truncated) {
- return false;
- }
-
- loadDataBase();
- return true;
- }
-
- /**
- * deserialize a snapshot from an input archive
- * @param ia the input archive you want to deserialize from
- * @throws IOException
- */
- public void deserializeSnapshot(InputArchive ia) throws IOException {
- clear();
- SerializeUtils.deserializeSnapshot(getDataTree(),ia,getSessionWithTimeOuts());
- initialized = true;
- }
-
- /**
- * serialize the snapshot
- * @param oa the output archive to which the snapshot needs to be serialized
- * @throws IOException
- * @throws InterruptedException
- */
- public void serializeSnapshot(OutputArchive oa) throws IOException,
- InterruptedException {
- SerializeUtils.serializeSnapshot(getDataTree(), oa, getSessionWithTimeOuts());
- }
-
- /**
- * append to the underlying transaction log
- * @param si the request to append
- * @return true if the append was succesfull and false if not
- */
- public boolean append(Request si) throws IOException {
- return this.snapLog.append(si);
- }
-
- /**
- * roll the underlying log
- */
- public void rollLog() throws IOException {
- this.snapLog.rollLog();
- }
-
- /**
- * commit to the underlying transaction log
- * @throws IOException
- */
- public void commit() throws IOException {
- this.snapLog.commit();
- }
-
- /**
- * close this database. free the resources
- * @throws IOException
- */
- public void close() throws IOException {
- this.snapLog.close();
- }
-
- public synchronized void initConfigInZKDatabase(QuorumVerifier qv) {
- if (qv == null) return; // only happens during tests
- try {
- if (this.dataTree.getNode(ZooDefs.CONFIG_NODE) == null) {
- // should only happen during upgrade
- LOG.warn("configuration znode missing (should only happen during upgrade), creating the node");
- this.dataTree.addConfigNode();
- }
- this.dataTree.setData(ZooDefs.CONFIG_NODE, qv.toString().getBytes(), -1, qv.getVersion(), Time.currentWallTime());
- } catch (NoNodeException e) {
- System.out.println("configuration node missing - should not happen");
- }
- }
-
- /**
- * Use for unit testing, so we can turn this feature on/off
- * @param snapshotSizeFactor Set to minus value to turn this off.
- */
- public void setSnapshotSizeFactor(double snapshotSizeFactor) {
- this.snapshotSizeFactor = snapshotSizeFactor;
- }
-
- /**
- * Check whether the given watcher exists in datatree
- *
- * @param path
- * node to check watcher existence
- * @param type
- * type of watcher
- * @param watcher
- * watcher function
- */
- public boolean containsWatcher(String path, WatcherType type, Watcher watcher) {
- return dataTree.containsWatcher(path, type, watcher);
- }
-
- /**
- * Remove watch from the datatree
- *
- * @param path
- * node to remove watches from
- * @param type
- * type of watcher to remove
- * @param watcher
- * watcher function to remove
- */
- public boolean removeWatch(String path, WatcherType type, Watcher watcher) {
- return dataTree.removeWatch(path, type, watcher);
- }
-
- // visible for testing
- public DataTree createDataTree() {
- return new DataTree();
- }
-}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/faa7cec7/src/java/main/org/apache/zookeeper/server/ZooKeeperCriticalThread.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/ZooKeeperCriticalThread.java b/src/java/main/org/apache/zookeeper/server/ZooKeeperCriticalThread.java
deleted file mode 100644
index d52d01d..0000000
--- a/src/java/main/org/apache/zookeeper/server/ZooKeeperCriticalThread.java
+++ /dev/null
@@ -1,51 +0,0 @@
-/**
- * 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.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Represents critical thread. When there is an uncaught exception thrown by the
- * thread this will exit the system.
- */
-public class ZooKeeperCriticalThread extends ZooKeeperThread {
- private static final Logger LOG = LoggerFactory
- .getLogger(ZooKeeperCriticalThread.class);
- private final ZooKeeperServerListener listener;
-
- public ZooKeeperCriticalThread(String threadName, ZooKeeperServerListener listener) {
- super(threadName);
- this.listener = listener;
- }
-
- /**
- * This will be used by the uncaught exception handler and make the system
- * exit.
- *
- * @param threadName
- * - thread name
- * @param e
- * - exception object
- */
- @Override
- protected void handleException(String threadName, Throwable e) {
- LOG.error("Severe unrecoverable error, from thread : {}", threadName, e);
- listener.notifyStopping(threadName, ExitCode.UNEXPECTED_ERROR.getValue());
- }
-}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/faa7cec7/src/java/main/org/apache/zookeeper/server/ZooKeeperSaslServer.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/ZooKeeperSaslServer.java b/src/java/main/org/apache/zookeeper/server/ZooKeeperSaslServer.java
deleted file mode 100644
index af19e5d..0000000
--- a/src/java/main/org/apache/zookeeper/server/ZooKeeperSaslServer.java
+++ /dev/null
@@ -1,65 +0,0 @@
-/**
- * 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 javax.security.auth.Subject;
-import javax.security.sasl.SaslException;
-import javax.security.sasl.SaslServer;
-
-import org.apache.zookeeper.Login;
-import org.apache.zookeeper.util.SecurityUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class ZooKeeperSaslServer {
- public static final String LOGIN_CONTEXT_NAME_KEY = "zookeeper.sasl.serverconfig";
- public static final String DEFAULT_LOGIN_CONTEXT_NAME = "Server";
-
- private static final Logger LOG = LoggerFactory.getLogger(ZooKeeperSaslServer.class);
- private SaslServer saslServer;
-
- ZooKeeperSaslServer(final Login login) {
- saslServer = createSaslServer(login);
- }
-
- private SaslServer createSaslServer(final Login login) {
- synchronized (login) {
- Subject subject = login.getSubject();
- return SecurityUtils.createSaslServer(subject, "zookeeper",
- "zk-sasl-md5", login.callbackHandler, LOG);
- }
- }
-
- public byte[] evaluateResponse(byte[] response) throws SaslException {
- return saslServer.evaluateResponse(response);
- }
-
- public boolean isComplete() {
- return saslServer.isComplete();
- }
-
- public String getAuthorizationID() {
- return saslServer.getAuthorizationID();
- }
-
-}
-
-
-
-
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/faa7cec7/src/java/main/org/apache/zookeeper/server/ZooKeeperServer.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/ZooKeeperServer.java b/src/java/main/org/apache/zookeeper/server/ZooKeeperServer.java
deleted file mode 100644
index 02df585..0000000
--- a/src/java/main/org/apache/zookeeper/server/ZooKeeperServer.java
+++ /dev/null
@@ -1,1270 +0,0 @@
-/**
- * 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.common.Time;
-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.metrics.MetricsContext;
-import org.apache.zookeeper.metrics.impl.NullMetricsProvider;
-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.auth.ServerAuthenticationProvider;
-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
- * is more of a sanity check.
- */
- 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 Map<String, ChangeRecord> outstandingChangesForPath =
- new HashMap<String, ChangeRecord>();
-
- protected ServerCnxnFactory serverCnxnFactory;
- protected ServerCnxnFactory secureServerCnxnFactory;
-
- private final ServerStats serverStats;
- private final ZooKeeperServerListener listener;
- private MetricsContext rootMetricsContext = NullMetricsProvider.NullMetricsContext.INSTANCE;
- 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(),
- serverCnxnFactory.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
- List<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() {
- takeSnapshot(false);
- }
-
- public void takeSnapshot(boolean syncSnap){
- long start = Time.currentElapsedTime();
- try {
- txnLogFactory.save(zkDb.getDataTree(), zkDb.getSessionWithTimeOuts(), syncSnap);
- } 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(ExitCode.TXNLOG_ERROR_TAKING_SNAPSHOT.getValue());
- }
- long elapsed = Time.currentElapsedTime() - start;
- LOG.info("Snapshot taken in " + elapsed + " ms");
- ServerMetrics.SNAPSHOT_TIME.add(elapsed);
- }
-
- @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;
- }
-
- public MetricsContext getRootMetricsContext() {
- return rootMetricsContext;
- }
-
- public void setRootMetricsContext(MetricsContext rootMetricsContext) {
- this.rootMetricsContext = rootMetricsContext;
- }
-
- /**
- * 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()));
- }
- ServerMetrics.CONNECTION_REQUEST_COUNT.add(1);
- 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();
- ServerAuthenticationProvider ap = ProviderRegistry.getServerProvider(scheme);
- Code authReturn = KeeperException.Code.AUTHFAILED;
- if(ap != null) {
- try {
- authReturn = ap.handleAuthentication(new ServerAuthenticationProvider.ServerObjs(this, 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 {
- cnxn.incrOutstandingRequests(h);
- 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);
- return;
- }
- }
-
- 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);
- byte[] responseToken = null;
- try {
- ZooKeeperSaslServer saslServer = cnxn.zooKeeperSaslServer;
- try {
- // note that clientToken might be empty (clientToken.length == 0):
- // if using the DIGEST-MD5 mechanism, clientToken will be empty at the beginning of the
- // SASL negotiation process.
- responseToken = saslServer.evaluateResponse(clientToken);
- if (saslServer.isComplete()) {
- String authorizationID = saslServer.getAuthorizationID();
- LOG.info("adding SASL authorization for authorizationID: " + authorizationID);
- cnxn.addAuthInfo(new Id("sasl",authorizationID));
- if (System.getProperty("zookeeper.superUser") != null &&
- authorizationID.equals(System.getProperty("zookeeper.superUser"))) {
- cnxn.addAuthInfo(new Id("super", ""));
- }
- }
- }
- catch (SaslException e) {
- LOG.warn("Client failed to SASL authenticate: " + e, e);
- if ((System.getProperty("zookeeper.allowSaslFailedClients") != null)
- &&
- (System.getProperty("zookeeper.allowSaslFailedClients").equals("true"))) {
- LOG.warn("Maintaining client connection despite SASL authentication failure.");
- } else {
- LOG.warn("Closing client connection due to SASL authentication failure.");
- cnxn.close();
- }
- }
- }
- catch (NullPointerException e) {
- LOG.error("cnxn.saslServer is null: cnxn object did not initialize its saslServer properly.");
- }
- if (responseToken != null) {
- LOG.debug("Size of server SASL response: " + responseToken.length);
- }
- // wrap SASL response token to client inside a Response object.
- return new SetSASLResponse(responseToken);
- }
-
- // entry point for quorum/Learner.java
- public ProcessTxnResult processTxn(TxnHeader hdr, Record txn) {
- return processTxn(null, hdr, txn);
- }
-
- // entry point for FinalRequestProcessor.java
- public ProcessTxnResult processTxn(Request request) {
- return processTxn(request, request.getHdr(), request.getTxn());
- }
-
- private ProcessTxnResult processTxn(Request request, TxnHeader hdr,
- Record txn) {
- ProcessTxnResult rc;
- int opCode = request != null ? request.type : hdr.getType();
- long sessionId = request != null ? request.sessionId : hdr.getClientId();
-
- if (opCode == OpCode.createSession) {
- if (hdr != null && txn instanceof CreateSessionTxn) {
- CreateSessionTxn cst = (CreateSessionTxn) txn;
- sessionTracker.commitSession(sessionId, cst.getTimeOut());
- } else if (request == null || !request.isLocalSession()) {
- LOG.warn("*****>>>>> Got "
- + txn.getClass() + " "
- + txn.toString());
- }
- } else if (opCode == OpCode.closeSession) {
- sessionTracker.removeSession(sessionId);
- }
-
- if (hdr != null) {
- rc = getZKDatabase().processTxn(hdr, txn);
- } else {
- rc = new ProcessTxnResult();
- }
- return rc;
- }
-
- public Map<Long, Set<Long>> getSessionExpiryMap() {
- return sessionTracker.getSessionExpiryMap();
- }
-
- /**
- * This method is used to register the ZooKeeperServerShutdownHandler to get
- * server's error or shutdown state change notifications.
- * {@link ZooKeeperServerShutdownHandler#handle(State)} will be called for
- * every server state changes {@link #setState(State)}.
- *
- * @param zkShutdownHandler shutdown handler
- */
- void registerServerShutdownHandler(ZooKeeperServerShutdownHandler zkShutdownHandler) {
- this.zkShutdownHandler = zkShutdownHandler;
- }
-}