You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zookeeper.apache.org by br...@apache.org on 2009/06/03 07:14:17 UTC
svn commit: r781264 - in /hadoop/zookeeper/trunk: ./
src/java/main/org/apache/zookeeper/
src/java/main/org/apache/zookeeper/server/
src/java/main/org/apache/zookeeper/server/persistence/
src/java/main/org/apache/zookeeper/server/quorum/ src/java/test/c...
Author: breed
Date: Wed Jun 3 05:14:16 2009
New Revision: 781264
URL: http://svn.apache.org/viewvc?rev=781264&view=rev
Log:
ZOOKEEPER-406. address all findbugs warnings in persistence classes
Modified:
hadoop/zookeeper/trunk/CHANGES.txt
hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/ClientCnxn.java
hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/ZooKeeperMain.java
hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/DataNode.java
hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/DataTree.java
hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/FinalRequestProcessor.java
hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/PrepRequestProcessor.java
hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/SyncRequestProcessor.java
hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/persistence/FileTxnLog.java
hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/persistence/FileTxnSnapLog.java
hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/persistence/Util.java
hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/quorum/FollowerHandler.java
hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/quorum/LeaderZooKeeperServer.java
hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/quorum/ProposalRequestProcessor.java
hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/quorum/QuorumCnxManager.java
hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/quorum/QuorumPeer.java
hadoop/zookeeper/trunk/src/java/test/config/findbugsExcludeFile.xml
hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/ACLTest.java
hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/AsyncOpsTest.java
hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/ClientBase.java
hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/FLENewEpochTest.java
hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/FLETest.java
hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/HierarchicalQuorumTest.java
hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/LETest.java
hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/QuorumBase.java
Modified: hadoop/zookeeper/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/CHANGES.txt?rev=781264&r1=781263&r2=781264&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/CHANGES.txt (original)
+++ hadoop/zookeeper/trunk/CHANGES.txt Wed Jun 3 05:14:16 2009
@@ -111,6 +111,8 @@
ZOOKEEPER-421. zkpython run_tests.sh is missing #!
(henry robinson via phunt)
+ ZOOKEEPER-406. address all findbugs warnings in persistence classes. (phunt et al via breed)
+
IMPROVEMENTS:
ZOOKEEPER-308. improve the atomic broadcast performance 3x.
(breed via mahadev)
Modified: hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/ClientCnxn.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/ClientCnxn.java?rev=781264&r1=781263&r2=781264&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/ClientCnxn.java (original)
+++ hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/ClientCnxn.java Wed Jun 3 05:14:16 2009
@@ -80,9 +80,9 @@
* Clients automatically reset watches during session reconnect, this
* option allows the client to turn off this behavior by setting
* the environment variable "zookeeper.disableAutoWatchReset" to "true" */
- public static boolean disableAutoWatchReset;
+ public static final boolean disableAutoWatchReset;
- public static int packetLen;
+ public static final int packetLen;
static {
// this var should not be public, but otw there is no easy way
// to test
Modified: hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/ZooKeeperMain.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/ZooKeeperMain.java?rev=781264&r1=781263&r2=781264&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/ZooKeeperMain.java (original)
+++ hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/ZooKeeperMain.java Wed Jun 3 05:14:16 2009
@@ -46,7 +46,7 @@
*/
public class ZooKeeperMain {
private static final Logger LOG = Logger.getLogger(ZooKeeperMain.class);
- protected static Map<String,String> commandMap = new HashMap<String,String>( );
+ protected static final Map<String,String> commandMap = new HashMap<String,String>( );
protected MyCommandOptions cl = new MyCommandOptions();
protected HashMap<Integer,String> history = new HashMap<Integer,String>( );
@@ -522,8 +522,11 @@
// now create the direct children
// and the stat and quota nodes
String[] splits = path.split("/");
+ StringBuffer sb = new StringBuffer();
+ sb.append(quotaPath);
for (int i=1; i<splits.length; i++) {
- quotaPath = quotaPath + "/" + splits[i];
+ sb.append("/" + splits[i]);
+ quotaPath = sb.toString();
try {
zk.create(quotaPath, null, Ids.OPEN_ACL_UNSAFE ,
CreateMode.PERSISTENT);
Modified: hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/DataNode.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/DataNode.java?rev=781264&r1=781263&r2=781264&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/DataNode.java (original)
+++ hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/DataNode.java Wed Jun 3 05:14:16 2009
@@ -53,7 +53,9 @@
/** the list of children for this node. note
* that the list of children string does not
* contain the parent path -- just the last
- * part of the path.
+ * part of the path. This should be
+ * synchronized on except deserializing
+ * (for speed up issues).
*/
Set<String> children = new HashSet<String>();
@@ -83,7 +85,7 @@
* fully
* @param children
*/
- public void setChildren(HashSet<String> children) {
+ public synchronized void setChildren(HashSet<String> children) {
this.children = children;
}
@@ -91,12 +93,12 @@
* convenience methods to get the children
* @return the children of this datanode
*/
- public Set<String> getChildren() {
+ public synchronized Set<String> getChildren() {
return this.children;
}
- public void copyStat(Stat to) {
+ synchronized public void copyStat(Stat to) {
to.setAversion(stat.getAversion());
to.setCtime(stat.getCtime());
to.setCversion(stat.getCversion());
@@ -110,7 +112,7 @@
to.setNumChildren(children.size());
}
- public void deserialize(InputArchive archive, String tag)
+ synchronized public void deserialize(InputArchive archive, String tag)
throws IOException {
archive.startRecord("node");
data = archive.readBuffer("data");
Modified: hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/DataTree.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/DataTree.java?rev=781264&r1=781263&r2=781264&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/DataTree.java (original)
+++ hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/DataTree.java Wed Jun 3 05:14:16 2009
@@ -244,8 +244,10 @@
public long approximateDataSize() {
long result = 0;
for (Map.Entry<String, DataNode> entry : nodes.entrySet()) {
- result += entry.getKey().length();
- result += entry.getValue().data.length;
+ synchronized(entry.getValue()) {
+ result += entry.getKey().length();
+ result += (entry.getValue().data == null? 0 : entry.getValue().data.length);
+ }
}
return result;
}
@@ -335,6 +337,11 @@
String statNode = Quotas.statPath(lastPrefix);
DataNode node = nodes.get(statNode);
StatsTrack updatedStat = null;
+ if (node == null) {
+ //should not happen
+ LOG.error("Missing count node for stat " + statNode);
+ return;
+ }
synchronized(node) {
updatedStat = new StatsTrack(new String(node.data));
updatedStat.setCount(updatedStat.getCount() + diff);
@@ -344,6 +351,11 @@
String quotaNode = Quotas.quotaPath(lastPrefix);
node = nodes.get(quotaNode);
StatsTrack thisStats = null;
+ if (node == null) {
+ //should not happen
+ LOG.error("Missing count node for quota " + quotaNode);
+ return;
+ }
synchronized(node) {
thisStats = new StatsTrack(new String(node.data));
}
@@ -358,10 +370,17 @@
* update the count of bytes of this stat datanode
* @param lastPrefix the path of the node that is quotaed
* @param diff the diff to added to number of bytes
+ * @throws IOException if path is not found
*/
- public void updateBytes(String lastPrefix, long diff) {
+ public void updateBytes(String lastPrefix, long diff) {
String statNode = Quotas.statPath(lastPrefix);
DataNode node = nodes.get(statNode);
+ if (node == null) {
+ //should never be null but just to make
+ // findbugs happy
+ LOG.error("Missing stat node for bytes " + statNode);
+ return;
+ }
StatsTrack updatedStat = null;
synchronized(node) {
updatedStat = new StatsTrack(new String(node.data));
@@ -371,6 +390,12 @@
// now check if the bytes match the quota
String quotaNode = Quotas.quotaPath(lastPrefix);
node = nodes.get(quotaNode);
+ if (node == null) {
+ //should never be null but just to make
+ // findbugs happy
+ LOG.error("Missing quota node for bytes " + quotaNode);
+ return;
+ }
StatsTrack thisStats = null;
synchronized(node) {
thisStats = new StatsTrack(new String(node.data));
@@ -510,7 +535,11 @@
if (!rootZookeeper.equals(lastPrefix) && !("".equals(lastPrefix))) {
// ok we have some match and need to update
updateCount(lastPrefix, -1);
- updateBytes(lastPrefix, node.data == null? 0:-(node.data.length));
+ int bytes = 0;
+ synchronized (node) {
+ bytes = (node.data == null? 0:-(node.data.length));
+ }
+ updateBytes(lastPrefix, bytes);
}
ZooTrace.logTraceMessage(LOG,
ZooTrace.EVENT_DELIVERY_TRACE_MASK,
@@ -531,8 +560,9 @@
if (n == null) {
throw new KeeperException.NoNodeException();
}
- byte lastdata[] = n.data;
+ byte lastdata[] = null;
synchronized (n) {
+ lastdata = n.data;
n.data = data;
n.stat.setMtime(time);
n.stat.setMzxid(zxid);
@@ -775,13 +805,15 @@
return;
}
String[] children = null;
+ int len = 0;
synchronized (node) {
children = node.children.toArray(new
String[node.children.size()]);
+ len = (node.data == null? 0: node.data.length);
}
// add itself
counts.count += 1;
- counts.bytes += (long)node.data.length;
+ counts.bytes += len;
if (children.length == 0) {
return;
}
@@ -959,6 +991,10 @@
} else {
String parentPath = path.substring(0, lastSlash);
node.parent = nodes.get(parentPath);
+ if (node.parent == null) {
+ throw new IOException("Invalid Datatree, unable to find " +
+ "parent " + parentPath + " of path " + path);
+ }
node.parent.children.add(path.substring(lastSlash + 1));
long eowner = node.stat.getEphemeralOwner();
if (eowner != 0) {
Modified: hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/FinalRequestProcessor.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/FinalRequestProcessor.java?rev=781264&r1=781263&r2=781264&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/FinalRequestProcessor.java (original)
+++ hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/FinalRequestProcessor.java Wed Jun 3 05:14:16 2009
@@ -192,7 +192,11 @@
if (n == null) {
throw new KeeperException.NoNodeException();
}
- PrepRequestProcessor.checkACL(zks, zks.dataTree.convertLong(n.acl),
+ Long aclL;
+ synchronized(n) {
+ aclL = n.acl;
+ }
+ PrepRequestProcessor.checkACL(zks, zks.dataTree.convertLong(aclL),
ZooDefs.Perms.READ,
request.authInfo);
stat = new Stat();
@@ -229,7 +233,11 @@
if (n == null) {
throw new KeeperException.NoNodeException();
}
- PrepRequestProcessor.checkACL(zks, zks.dataTree.convertLong(n.acl),
+ Long aclG;
+ synchronized(n) {
+ aclG = n.acl;
+ }
+ PrepRequestProcessor.checkACL(zks, zks.dataTree.convertLong(aclG),
ZooDefs.Perms.READ,
request.authInfo);
List<String> children = zks.dataTree.getChildren(
Modified: hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/PrepRequestProcessor.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/PrepRequestProcessor.java?rev=781264&r1=781263&r2=781264&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/PrepRequestProcessor.java (original)
+++ hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/PrepRequestProcessor.java Wed Jun 3 05:14:16 2009
@@ -23,6 +23,7 @@
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
+import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;
import org.apache.jute.Record;
@@ -73,7 +74,7 @@
* this is only for testing purposes.
* should never be useed otherwise
*/
- public static boolean failCreate = false;
+ public static final boolean failCreate = false;
LinkedBlockingQueue<Request> submittedRequests = new LinkedBlockingQueue<Request>();
@@ -124,8 +125,14 @@
if (lastChange == null) {
DataNode n = zks.dataTree.getNode(path);
if (n != null) {
- lastChange = new ChangeRecord(-1, path, n.stat, n.children
- .size(), zks.dataTree.convertLong(n.acl));
+ Long acl;
+ Set<String> children;
+ synchronized(n) {
+ acl = n.acl;
+ children = n.children;
+ }
+ lastChange = new ChangeRecord(-1, path, n.stat, children
+ .size(), zks.dataTree.convertLong(acl));
}
}
}
Modified: hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/SyncRequestProcessor.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/SyncRequestProcessor.java?rev=781264&r1=781263&r2=781264&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/SyncRequestProcessor.java (original)
+++ hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/SyncRequestProcessor.java Wed Jun 3 05:14:16 2009
@@ -51,7 +51,7 @@
/**
* The number of log entries to log before starting a snapshot
*/
- public static int snapCount = ZooKeeperServer.getSnapCount();
+ public static final int snapCount = ZooKeeperServer.getSnapCount();
private Request requestOfDeath = Request.requestOfDeath;
Modified: hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/persistence/FileTxnLog.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/persistence/FileTxnLog.java?rev=781264&r1=781263&r2=781264&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/persistence/FileTxnLog.java (original)
+++ hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/persistence/FileTxnLog.java Wed Jun 3 05:14:16 2009
@@ -49,25 +49,44 @@
*
*/
public class FileTxnLog implements TxnLog {
+ private static final Logger LOG;
+
+ static long preAllocSize = 65536 * 1024;
+
+ public final static int TXNLOG_MAGIC =
+ ByteBuffer.wrap("ZKLG".getBytes()).getInt();
+
+ public final static int VERSION = 2;
+
+ static {
+ LOG = Logger.getLogger(FileTxnLog.class);
+
+ forceSync =
+ !System.getProperty("zookeeper.forceSync", "yes").equals("no");
+
+ String size = System.getProperty("zookeeper.preAllocSize");
+ if (size != null) {
+ try {
+ preAllocSize = Long.parseLong(size) * 1024;
+ } catch (NumberFormatException e) {
+ LOG.warn(size + " is not a valid value for preAllocSize");
+ }
+ }
+ }
+
long lastZxidSeen;
volatile BufferedOutputStream logStream = null;
volatile OutputArchive oa;
volatile FileOutputStream fos = null;
-
-
+
File logDir;
- public final static int TXNLOG_MAGIC =
- ByteBuffer.wrap("ZKLG".getBytes()).getInt();
- public final static int VERSION = 2;
- private boolean forceSync = true;
+ private static boolean forceSync = true;
long dbId;
- private LinkedList<FileOutputStream> streamsToFlush = new LinkedList<FileOutputStream>();
- static long preAllocSize = 65536 * 1024;
+ private LinkedList<FileOutputStream> streamsToFlush =
+ new LinkedList<FileOutputStream>();
long currentSize;
File logFileWrite = null;
-
- private static final Logger LOG = Logger.getLogger(FileTxnLog.class);
-
+
/**
* constructor for FileTxnLog. Take the directory
* where the txnlogs are stored
@@ -75,27 +94,17 @@
*/
public FileTxnLog(File logDir) {
this.logDir = logDir;
- forceSync = !System.getProperty("zookeeper.forceSync", "yes").equals(
- "no");
- String size = System.getProperty("zookeeper.preAllocSize");
- if (size != null) {
- try {
- preAllocSize = Long.parseLong(size) * 1024;
- } catch (NumberFormatException e) {
- LOG.warn(size + " is not a valid value for preAllocSize");
- }
- }
}
-
+
/**
* method to allow setting preallocate size
* of log file to pad the file.
- * @param size the size to set to
+ * @param size the size to set to in bytes
*/
public static void setPreallocSize(long size) {
preAllocSize = size;
}
-
+
/**
* creates a checksum alogrithm to be used
* @return the checksum used for this txnlog
@@ -107,7 +116,7 @@
/**
* rollover the current log file to a new one.
- * @throws IOException
+ * @throws IOException
*/
public void rollLog() throws IOException {
if (logStream != null) {
@@ -122,7 +131,7 @@
* @param hdr the header of the transaction
* @param txn the transaction part of the entry
*/
- public synchronized void append(TxnHeader hdr, Record txn)
+ public synchronized void append(TxnHeader hdr, Record txn)
throws IOException {
if (hdr != null) {
if (hdr.getZxid() <= lastZxidSeen) {
@@ -131,7 +140,7 @@
+ hdr.getType());
}
if (logStream==null) {
- logFileWrite = new File(logDir, ("log." +
+ logFileWrite = new File(logDir, ("log." +
Long.toHexString(hdr.getZxid())));
fos = new FileOutputStream(logFileWrite);
logStream=new BufferedOutputStream(fos);
@@ -145,15 +154,15 @@
byte[] buf = Util.marshallTxnEntry(hdr, txn);
if (buf == null || buf.length == 0) {
throw new IOException("Faulty serialization for header " +
- "and txn");
+ "and txn");
}
Checksum crc = makeChecksumAlgorithm();
crc.update(buf, 0, buf.length);
oa.writeLong(crc.getValue(), "txnEntryCRC");
Util.writeTxnBytes(oa, buf);
- }
+ }
}
-
+
/**
* pad the current file to increase its size
* @param out the outputstream to be padded
@@ -162,7 +171,7 @@
private void padFile(FileOutputStream out) throws IOException {
currentSize = Util.padLogFile(out, currentSize, preAllocSize);
}
-
+
/**
* 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,
@@ -181,7 +190,7 @@
if (fzxid > snapshotZxid) {
continue;
}
- // the files
+ // the files
// are sorted with zxid's
if (fzxid > logZxid) {
logZxid = fzxid;
@@ -196,9 +205,9 @@
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
@@ -207,11 +216,10 @@
File[] files = getLogFiles(logDir.listFiles(), 0);
long maxLog=files.length>0?
Util.getZxidFromName(files[files.length-1].getName(),"log"):-1;
-
- // if a log file is more recent we must scan it to find
+
+ // if a log file is more recent we must scan it to find
// the highest zxid
long zxid = maxLog;
- FileOutputStream logStream = null;
try {
FileTxnLog txn = new FileTxnLog(logDir);
TxnIterator itr = txn.read(maxLog);
@@ -223,17 +231,12 @@
}
} catch (IOException e) {
LOG.warn("Unexpected exception", e);
- } finally {
- if (logStream != null)
- try {
- logStream.close();
- } catch(IOException io){}
}
return zxid;
}
-
+
/**
- * commit the logs. make sure that evertyhing hits the
+ * commit the logs. make sure that evertyhing hits the
* disk
*/
public synchronized void commit() throws IOException {
@@ -250,7 +253,7 @@
streamsToFlush.removeFirst().close();
}
}
-
+
/**
* start reading all the transactions from the given zxid
* @param zxid the zxid to start reading transactions from
@@ -259,8 +262,8 @@
*/
public TxnIterator read(long zxid) throws IOException {
return new FileTxnIterator(logDir, zxid);
- }
-
+ }
+
/**
* truncate the current transaction logs
* @param zxid the zxid to truncate the logs to
@@ -275,11 +278,13 @@
raf.setLength(pos);
raf.close();
while(itr.goToNextLog()) {
- itr.logFile.delete();
+ if (!itr.logFile.delete()) {
+ LOG.warn("Unable to truncate " + itr.logFile);
+ }
}
return true;
}
-
+
/**
* read the header of the transaction file
* @param file the transaction file to read
@@ -288,20 +293,21 @@
*/
private static FileHeader readHeader(File file) throws IOException {
InputStream is =null;
- try{
+ 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){
+ } 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
@@ -314,10 +320,10 @@
throw new IOException("Unsupported Format.");
return fh.getDbid();
}
-
+
/**
- * this class implements the txnlog iterator interface
- * which is used for reading the transaction logs
+ * this class implements the txnlog iterator interface
+ * which is used for reading the transaction logs
*/
public static class FileTxnIterator implements TxnLog.TxnIterator {
File logDir;
@@ -328,10 +334,10 @@
InputArchive ia;
static final String CRC_ERROR="CRC check failed";
FileInputStream inputStream=null;
- //stored files is the list of files greater than
+ //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
@@ -343,10 +349,10 @@
this.zxid = zxid;
init();
}
-
+
/**
* initialize to the zxid specified
- * this is inclusive of the zxid
+ * this is inclusive of the zxid
* @throws IOException
*/
void init() throws IOException {
@@ -370,10 +376,10 @@
return;
}
}
-
+
/**
- * go to the next logfile
- * @return true if there is one and false if there is no
+ * go to the next logfile
+ * @return true if there is one and false if there is no
* new file to be read
* @throws IOException
*/
@@ -385,23 +391,23 @@
}
return false;
}
-
+
/**
* read the header fomr the inputarchive
* @param ia the inputarchive to be read from
- * @param is the inputstream
+ * @param is the inputstream
* @throws IOException
*/
- protected void inStreamCreated(InputArchive ia, FileInputStream is)
+ protected void inStreamCreated(InputArchive ia, FileInputStream is)
throws IOException{
FileHeader header= new FileHeader();
header.deserialize(ia, "fileheader");
if (header.getMagic() != FileTxnLog.TXNLOG_MAGIC) {
- throw new IOException("Invalid magic number " + header.getMagic()
+ throw new IOException("Invalid magic number " + header.getMagic()
+ " != " + FileTxnLog.TXNLOG_MAGIC);
- }
+ }
}
-
+
/**
* Invoked to indicate that the input stream has been created.
* @param ia input archive
@@ -418,15 +424,15 @@
}
return ia;
}
-
+
/**
- * create a checksum algorithm
+ * create a checksum algorithm
* @return the checksum algorithm
*/
protected Checksum makeChecksumAlgorithm(){
return new Adler32();
}
-
+
/**
* the iterator that moves to the next transaction
* @return true if there is more transactions to be read
@@ -441,12 +447,12 @@
byte[] bytes = Util.readTxnBytes(ia);
// Since we preallocate, we define EOF to be an
if (bytes == null || bytes.length==0)
- throw new EOFException("Failed to read");
+ throw new EOFException("Failed to read");
// EOF or corrupted record
// validate CRC
Checksum crc = makeChecksumAlgorithm();
crc.update(bytes, 0, bytes.length);
- if (crcValue != crc.getValue())
+ if (crcValue != crc.getValue())
throw new IOException(CRC_ERROR);
if (bytes == null || bytes.length == 0)
return false;
@@ -459,7 +465,7 @@
inputStream.close();
inputStream = null;
ia = null;
- // thsi means that the file has ended
+ // thsi means that the file has ended
// we shoud go to the next file
if (!goToNextLog()) {
return false;
@@ -467,10 +473,10 @@
}
return true;
}
-
+
/**
- * reutrn the current header
- * @return the current header that
+ * reutrn the current header
+ * @return the current header that
* is read
*/
public TxnHeader getHeader() {
@@ -485,9 +491,9 @@
public Record getTxn() {
return record;
}
-
+
/**
- * close the iterator
+ * close the iterator
* and release the resources.
*/
public void close() throws IOException {
Modified: hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/persistence/FileTxnSnapLog.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/persistence/FileTxnSnapLog.java?rev=781264&r1=781263&r2=781264&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/persistence/FileTxnSnapLog.java (original)
+++ hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/persistence/FileTxnSnapLog.java Wed Jun 3 05:14:16 2009
@@ -71,14 +71,20 @@
* @param dataDir the trasaction directory
* @param snapDir the snapshot directory
*/
- public FileTxnSnapLog(File dataDir, File snapDir) {
+ public FileTxnSnapLog(File dataDir, File snapDir) throws IOException {
this.dataDir = new File(dataDir, version + VERSION);
this.snapDir = new File(snapDir, version + VERSION);
if (!this.dataDir.exists()) {
- this.dataDir.mkdirs();
+ if (!this.dataDir.mkdirs()) {
+ throw new IOException("Unable to create data directory "
+ + this.dataDir);
+ }
}
if (!this.snapDir.exists()) {
- this.snapDir.mkdirs();
+ if (!this.snapDir.mkdirs()) {
+ throw new IOException("Unable to create snap directory "
+ + this.snapDir);
+ }
}
txnLog = new FileTxnLog(this.dataDir);
snapLog = new FileSnap(this.snapDir);
Modified: hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/persistence/Util.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/persistence/Util.java?rev=781264&r1=781263&r2=781264&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/persistence/Util.java (original)
+++ hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/persistence/Util.java Wed Jun 3 05:14:16 2009
@@ -24,6 +24,7 @@
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.RandomAccessFile;
+import java.io.Serializable;
import java.net.URI;
import java.nio.ByteBuffer;
import java.util.ArrayList;
@@ -170,7 +171,17 @@
try {
raf.seek(raf.length() - 5);
byte bytes[] = new byte[5];
- raf.read(bytes);
+ int readlen = 0;
+ int l;
+ while(readlen < 5 &&
+ (l = raf.read(bytes, readlen, bytes.length - readlen)) >= 0) {
+ readlen += l;
+ }
+ if (readlen != bytes.length) {
+ LOG.info("Invalid snapshot " + f
+ + " too short, len = " + readlen);
+ return false;
+ }
ByteBuffer bb = ByteBuffer.wrap(bytes);
int len = bb.getInt();
byte b = bb.get();
@@ -271,7 +282,11 @@
* Compare file file names of form "prefix.version". Sort order result
* returned in order of version.
*/
- private static class DataDirFileComparator implements Comparator<File> {
+ private static class DataDirFileComparator
+ implements Comparator<File>, Serializable
+ {
+ private static final long serialVersionUID = -2648639884525140318L;
+
private String prefix;
private boolean ascending;
public DataDirFileComparator(String prefix, boolean ascending) {
Modified: hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/quorum/FollowerHandler.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/quorum/FollowerHandler.java?rev=781264&r1=781263&r2=781264&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/quorum/FollowerHandler.java (original)
+++ hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/quorum/FollowerHandler.java Wed Jun 3 05:14:16 2009
@@ -412,7 +412,11 @@
* ping calls from the leader to the followers
*/
public void ping() {
- QuorumPacket ping = new QuorumPacket(Leader.PING, leader.lastProposed,
+ long id;
+ synchronized(leader) {
+ id = leader.lastProposed;
+ }
+ QuorumPacket ping = new QuorumPacket(Leader.PING, id,
null, null);
queuePacket(ping);
}
Modified: hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/quorum/LeaderZooKeeperServer.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/quorum/LeaderZooKeeperServer.java?rev=781264&r1=781263&r2=781264&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/quorum/LeaderZooKeeperServer.java (original)
+++ hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/quorum/LeaderZooKeeperServer.java Wed Jun 3 05:14:16 2009
@@ -64,8 +64,9 @@
commitProcessor = new CommitProcessor(toBeAppliedProcessor,
Long.toString(getServerId()), false);
commitProcessor.start();
- RequestProcessor proposalProcessor = new ProposalRequestProcessor(this,
+ ProposalRequestProcessor proposalProcessor = new ProposalRequestProcessor(this,
commitProcessor);
+ proposalProcessor.initialize();
firstProcessor = new PrepRequestProcessor(this, proposalProcessor);
((PrepRequestProcessor)firstProcessor).start();
}
Modified: hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/quorum/ProposalRequestProcessor.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/quorum/ProposalRequestProcessor.java?rev=781264&r1=781263&r2=781264&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/quorum/ProposalRequestProcessor.java (original)
+++ hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/quorum/ProposalRequestProcessor.java Wed Jun 3 05:14:16 2009
@@ -39,9 +39,15 @@
this.nextProcessor = nextProcessor;
AckRequestProcessor ackProcessor = new AckRequestProcessor(zks.getLeader());
syncProcessor = new SyncRequestProcessor(zks, ackProcessor);
+ }
+
+ /**
+ * initialize this processor
+ */
+ public void initialize() {
syncProcessor.start();
}
-
+
public void processRequest(Request request) {
// LOG.warn("Ack>>> cxid = " + request.cxid + " type = " +
// request.type + " id = " + request.sessionId);
Modified: hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/quorum/QuorumCnxManager.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/quorum/QuorumCnxManager.java?rev=781264&r1=781263&r2=781264&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/quorum/QuorumCnxManager.java (original)
+++ hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/quorum/QuorumCnxManager.java Wed Jun 3 05:14:16 2009
@@ -491,8 +491,13 @@
senderWorkerMap.remove(sid);
ArrayBlockingQueue<ByteBuffer> bq = queueSendMap.get(sid);
if(bq != null){
- if (bq.size() == 0)
- bq.offer(b);
+ if (bq.size() == 0) {
+ boolean ret = bq.offer(b);
+ if (!ret) {
+ // to appease findbugs
+ LOG.error("Not able to add to a quue of size 0");
+ }
+ }
} else LOG.error("No queue for server " + sid);
}
}
Modified: hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/quorum/QuorumPeer.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/quorum/QuorumPeer.java?rev=781264&r1=781263&r2=781264&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/quorum/QuorumPeer.java (original)
+++ hadoop/zookeeper/trunk/src/java/main/org/apache/zookeeper/server/quorum/QuorumPeer.java Wed Jun 3 05:14:16 2009
@@ -207,8 +207,12 @@
break;
case LEADING:
responseBuffer.putLong(myid);
+ long proposed;
try {
- responseBuffer.putLong(leader.lastProposed);
+ synchronized(leader) {
+ proposed = leader.lastProposed;
+ }
+ responseBuffer.putLong(proposed);
} catch (NullPointerException npe) {
// This can happen in state transitions,
// just ignore the request
Modified: hadoop/zookeeper/trunk/src/java/test/config/findbugsExcludeFile.xml
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/java/test/config/findbugsExcludeFile.xml?rev=781264&r1=781263&r2=781264&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/java/test/config/findbugsExcludeFile.xml (original)
+++ hadoop/zookeeper/trunk/src/java/test/config/findbugsExcludeFile.xml Wed Jun 3 05:14:16 2009
@@ -65,4 +65,43 @@
<Bug code="EI, EI2" />
</Match>
+ <Match>
+ <Class name="org.apache.zookeeper.server.DataNode" />
+ <Bug code="EI2"/>
+ </Match>
+
+ <Match>
+ <Class name="org.apache.zookeeper.server.quorum.QuorumPacket" />
+ <Bug code="EI2, EI" />
+ </Match>
+
+ <Match>
+ <Class name="org.apache.zookeeper.ClientCnxn"/>
+ <Bug code="EI, EI2" />
+ </Match>
+
+ <Match>
+ <Class name="org.apache.zookeeper.server.DataNode"/>
+ <Field name="children"/>
+ <Bug code="IS"/>
+ </Match>
+
+ <Match>
+ <Class name="org.apache.zookeeper.server.quorum.FollowerSessionTracker"/>
+ <Bug code="UrF"/>
+ </Match>
+
+ <!-- these are old classes just for upgrading and should go away -->
+ <Match>
+ <Class name="org.apache.zookeeper.server.upgrade.DataNodeV1"/>
+ </Match>
+
+ <Match>
+ <Class name="org.apache.zookeeper.server.upgrade.DataTreeV1"/>
+ </Match>
+
+ <Match>
+ <Class name="org.apache.zookeeper.server.quorum.AuthFastLeaderElection"/>
+ <Bug code="SIC"/>
+ </Match>
</FindBugsFilter>
Modified: hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/ACLTest.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/ACLTest.java?rev=781264&r1=781263&r2=781264&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/ACLTest.java (original)
+++ hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/ACLTest.java Wed Jun 3 05:14:16 2009
@@ -45,7 +45,7 @@
private static String HOSTPORT = "127.0.0.1:2355";
ZooKeeperServer zks;
private CountDownLatch startSignal;
-
+
@Override
protected void setUp() throws Exception {
LOG.info("STARTING " + getName());
@@ -56,8 +56,8 @@
}
/**
- * Verify that acl optimization of storing just
- * a few acls and there references in the data
+ * Verify that acl optimization of storing just
+ * a few acls and there references in the data
* node is actually working.
*/
public void testAcls() throws Exception {
@@ -69,7 +69,7 @@
NIOServerCnxn.Factory f = new NIOServerCnxn.Factory(PORT);
f.startup(zks);
LOG.info("starting up the zookeeper server .. waiting");
- assertTrue("waiting for server being up",
+ assertTrue("waiting for server being up",
ClientBase.waitForServerUp(HOSTPORT,CONNECTION_TIMEOUT));
ZooKeeper zk = new ZooKeeper(HOSTPORT, CONNECTION_TIMEOUT, this);
String path;
@@ -102,17 +102,17 @@
zks = new ZooKeeperServer(tmpDir, tmpDir, 3000);
f = new NIOServerCnxn.Factory(PORT);
-
+
f.startup(zks);
assertTrue("waiting for server up",
ClientBase.waitForServerUp(HOSTPORT,
CONNECTION_TIMEOUT));
-
+
startSignal.await(CONNECTION_TIMEOUT,
TimeUnit.MILLISECONDS);
assertTrue("count == 0", startSignal.getCount() == 0);
-
+
assertTrue("acl map ", (101 == zks.dataTree.longKeyMap.size()));
for (int j =200; j < 205; j++) {
path = "/" + j;
@@ -133,20 +133,20 @@
assertTrue("waiting for server down",
ClientBase.waitForServerDown(HOSTPORT,
ClientBase.CONNECTION_TIMEOUT));
-
+
}
-
- /*
- * (non-Javadoc)
- *
+
+ /*
+ * (non-Javadoc)
+ *
* @see org.apache.zookeeper.Watcher#process(org.apache.zookeeper.WatcherEvent)
- */
+ */
public void process(WatchedEvent event) {
LOG.info("Event:" + event.getState() + " " + event.getType() + " " + event.getPath());
if (event.getState() == KeeperState.SyncConnected
&& startSignal != null && startSignal.getCount() > 0)
- {
- startSignal.countDown();
+ {
+ startSignal.countDown();
}
}
}
\ No newline at end of file
Modified: hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/AsyncOpsTest.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/AsyncOpsTest.java?rev=781264&r1=781263&r2=781264&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/AsyncOpsTest.java (original)
+++ hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/AsyncOpsTest.java Wed Jun 3 05:14:16 2009
@@ -35,19 +35,19 @@
/**
* Functional testing of asynchronous operations, both positive and negative
* testing.
- *
+ *
* This just scratches the surface, but exercises the basic async functionality.
*/
public class AsyncOpsTest extends ClientBase {
private static final Logger LOG = Logger.getLogger(AsyncOpsTest.class);
private ZooKeeper zk;
-
+
@Before
@Override
protected void setUp() throws Exception {
super.setUp();
-
+
LOG.info("STARTING " + getName());
zk = createClient();
@@ -58,9 +58,9 @@
@Override
protected void tearDown() throws Exception {
zk.close();
-
+
super.tearDown();
-
+
LOG.info("Test clients shutting down");
LOG.info("FINISHED " + getName());
@@ -74,23 +74,23 @@
@Test
public void testAsyncCreateThree() {
CountDownLatch latch = new CountDownLatch(3);
-
+
StringCB op1 = new StringCB(zk, latch);
op1.setPath("/op1");
StringCB op2 = new StringCB(zk, latch);
op2.setPath("/op2");
StringCB op3 = new StringCB(zk, latch);
op3.setPath("/op3");
-
+
op1.create();
op2.create();
op3.create();
-
+
op1.verify();
op2.verify();
op3.verify();
}
-
+
@Test
public void testAsyncCreateFailure_NodeExists() {
new StringCB(zk).verifyCreateFailure_NodeExists();
Modified: hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/ClientBase.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/ClientBase.java?rev=781264&r1=781263&r2=781264&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/ClientBase.java (original)
+++ hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/ClientBase.java Wed Jun 3 05:14:16 2009
@@ -76,7 +76,7 @@
// XXX this doesn't need to be volatile! (Should probably be final)
volatile CountDownLatch clientConnected;
volatile boolean connected;
-
+
public CountdownWatcher() {
reset();
}
@@ -106,7 +106,7 @@
}
if (!connected) {
throw new TimeoutException("Did not connect");
-
+
}
}
synchronized void waitForDisconnected(long timeout) throws InterruptedException, TimeoutException {
@@ -118,11 +118,11 @@
}
if (connected) {
throw new TimeoutException("Did not disconnect");
-
+
}
}
}
-
+
protected ZooKeeper createClient()
throws IOException, InterruptedException
{
@@ -225,7 +225,7 @@
}
return false;
}
-
+
static void verifyThreadTerminated(Thread thread, long millis)
throws InterruptedException
{
@@ -233,7 +233,7 @@
if (thread.isAlive()) {
LOG.error("Thread " + thread.getName() + " : "
+ Arrays.toString(thread.getStackTrace()));
- assertFalse("thread " + thread.getName()
+ assertFalse("thread " + thread.getName()
+ " still alive after join", true);
}
}
@@ -246,17 +246,17 @@
File tmpFile = File.createTempFile("test", ".junit", parentDir);
// don't delete tmpFile - this ensures we don't attempt to create
// a tmpDir with a duplicate name
-
+
File tmpDir = new File(tmpFile + ".dir");
assertFalse(tmpDir.exists()); // never true if tmpfile does it's job
assertTrue(tmpDir.mkdirs());
-
+
return tmpDir;
}
-
+
static NIOServerCnxn.Factory createNewServerInstance(File dataDir,
NIOServerCnxn.Factory factory, String hostPort)
- throws IOException, InterruptedException
+ throws IOException, InterruptedException
{
ZooKeeperServer zks = new ZooKeeperServer(dataDir, dataDir, 3000);
final int PORT = Integer.parseInt(hostPort.split(":")[1]);
@@ -271,7 +271,7 @@
return factory;
}
-
+
static void shutdownServerInstance(NIOServerCnxn.Factory factory,
String hostPort)
{
@@ -284,7 +284,7 @@
CONNECTION_TIMEOUT));
}
}
-
+
/**
* Test specific setup
*/
@@ -294,21 +294,20 @@
// resulting in test failure (client timeout on first session).
// set env and directly in order to handle static init/gc issues
System.setProperty("zookeeper.preAllocSize", "100");
- FileTxnLog.setPreallocSize(100);
+ FileTxnLog.setPreallocSize(100 * 1024);
}
-
+
@Override
protected void setUp() throws Exception {
LOG.info("STARTING " + getName());
+ setupTestEnv();
JMXEnv.setUp();
-
+
tmpDir = createTmpDir(BASETEST);
-
- setupTestEnv();
startServer();
-
+
LOG.info("Client test setup finished");
}
@@ -318,7 +317,7 @@
// ensure that only server and data bean are registered
JMXEnv.ensureOnly("InMemoryDataTree", "StandaloneServer_port");
}
-
+
protected void stopServer() throws Exception {
LOG.info("STOPPING server");
shutdownServerInstance(serverFactory, hostPort);
@@ -326,19 +325,19 @@
// ensure no beans are leftover
JMXEnv.ensureOnly();
}
-
+
@Override
protected void tearDown() throws Exception {
LOG.info("tearDown starting");
stopServer();
-
+
if (tmpDir != null) {
//assertTrue("delete " + tmpDir.toString(), recursiveDelete(tmpDir));
// FIXME see ZOOKEEPER-121 replace following line with previous
recursiveDelete(tmpDir);
}
-
+
JMXEnv.tearDown();
LOG.info("FINISHED " + getName());
@@ -347,7 +346,7 @@
public static MBeanServerConnection jmxConn() throws IOException {
return JMXEnv.conn();
}
-
+
private static boolean recursiveDelete(File d) {
if (d.isDirectory()) {
File children[] = d.listFiles();
Modified: hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/FLENewEpochTest.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/FLENewEpochTest.java?rev=781264&r1=781263&r2=781264&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/FLENewEpochTest.java (original)
+++ hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/FLENewEpochTest.java Wed Jun 3 05:14:16 2009
@@ -45,10 +45,10 @@
File tmpdir[];
int port[];
volatile int [] round;
-
+
Semaphore start0;
Semaphore finish3, finish0;
-
+
@Override
public void setUp() throws Exception {
count = 3;
@@ -64,11 +64,11 @@
round[0] = 0;
round[1] = 0;
round[2] = 0;
-
+
start0 = new Semaphore(0);
finish0 = new Semaphore(0);
finish3 = new Semaphore(0);
-
+
LOG.info("SetUp " + getName());
}
@@ -82,7 +82,6 @@
class LEThread extends Thread {
- FastLeaderElection le;
int i;
QuorumPeer peer;
@@ -90,67 +89,67 @@
this.i = i;
this.peer = peer;
LOG.info("Constructor: " + getName());
-
+
}
public void run(){
- boolean flag = true;
+ boolean flag = true;
try{
- while(flag){
- Vote v = null;
- peer.setPeerState(ServerState.LOOKING);
- LOG.info("Going to call leader election again: " + i);
- v = peer.getElectionAlg().lookForLeader();
-
- if(v == null){
- fail("Thread " + i + " got a null vote");
- }
-
- /*
- * A real zookeeper would take care of setting the current vote. Here
- * we do it manually.
- */
- peer.setCurrentVote(v);
-
- LOG.info("Finished election: " + i + ", " + v.id);
- //votes[i] = v;
-
- switch(i){
- case 0:
- LOG.info("First peer, do nothing, just join");
- if(finish0.tryAcquire(1000, java.util.concurrent.TimeUnit.MILLISECONDS)){
- //if(threads.get(0).peer.getPeerState() == ServerState.LEADING ){
- LOG.info("Setting flag to false");
- flag = false;
- }
- break;
- case 1:
- LOG.info("Second entering case");
- if(round[1] != 0){
- finish0.release();
- flag = false;
- }
- else{
- finish3.acquire();
- start0.release();
- }
- LOG.info("Second is going to start second round");
- round[1]++;
- break;
- case 2:
- LOG.info("Third peer, shutting it down");
- ((FastLeaderElection) peer.getElectionAlg()).shutdown();
- peer.shutdown();
- flag = false;
- round[2] = 1;
- finish3.release();
- LOG.info("Third leaving");
- break;
- }
- }
+ while(flag){
+ Vote v = null;
+ peer.setPeerState(ServerState.LOOKING);
+ LOG.info("Going to call leader election again: " + i);
+ v = peer.getElectionAlg().lookForLeader();
+
+ if(v == null){
+ fail("Thread " + i + " got a null vote");
+ }
+
+ /*
+ * A real zookeeper would take care of setting the current vote. Here
+ * we do it manually.
+ */
+ peer.setCurrentVote(v);
+
+ LOG.info("Finished election: " + i + ", " + v.id);
+ //votes[i] = v;
+
+ switch(i){
+ case 0:
+ LOG.info("First peer, do nothing, just join");
+ if(finish0.tryAcquire(1000, java.util.concurrent.TimeUnit.MILLISECONDS)){
+ //if(threads.get(0).peer.getPeerState() == ServerState.LEADING ){
+ LOG.info("Setting flag to false");
+ flag = false;
+ }
+ break;
+ case 1:
+ LOG.info("Second entering case");
+ if(round[1] != 0){
+ finish0.release();
+ flag = false;
+ }
+ else{
+ finish3.acquire();
+ start0.release();
+ }
+ LOG.info("Second is going to start second round");
+ round[1]++;
+ break;
+ case 2:
+ LOG.info("Third peer, shutting it down");
+ ((FastLeaderElection) peer.getElectionAlg()).shutdown();
+ peer.shutdown();
+ flag = false;
+ round[2] = 1;
+ finish3.release();
+ LOG.info("Third leaving");
+ break;
+ }
+ }
} catch (Exception e){
- e.printStackTrace();
- }
+ e.printStackTrace();
+ }
}
}
@@ -164,7 +163,7 @@
for(int i = 0; i < count; i++) {
peers.put(Long.valueOf(i), new QuorumServer(i, new InetSocketAddress(baseport+100+i),
new InetSocketAddress(baseLEport+100+i)));
- tmpdir[i] = File.createTempFile("letest", "test");
+ tmpdir[i] = ClientBase.createTmpDir();
port[i] = baseport+i;
}
@@ -183,7 +182,7 @@
LEThread thread = new LEThread(peer, 0);
thread.start();
threads.add(thread);
-
+
LOG.info("Started threads " + getName());
for(int i = 0; i < threads.size(); i++) {
Modified: hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/FLETest.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/FLETest.java?rev=781264&r1=781263&r2=781264&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/FLETest.java (original)
+++ hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/FLETest.java Wed Jun 3 05:14:16 2009
@@ -38,48 +38,46 @@
public class FLETest extends TestCase {
protected static final Logger LOG = Logger.getLogger(FLETest.class);
- class TestVote{
- TestVote(int id, long leader){
- this.leader = leader;
- this.id = id;
- }
+ static class TestVote {
+ TestVote(int id, long leader) {
+ this.leader = leader;
+ }
- long leader;
- int id;
+ long leader;
}
-
- int countVotes(HashSet<TestVote> hs, long id){
- int counter = 0;
- for(TestVote v : hs){
- if(v.leader == id) counter++;
+
+ int countVotes(HashSet<TestVote> hs, long id) {
+ int counter = 0;
+ for(TestVote v : hs){
+ if(v.leader == id) counter++;
}
- return counter;
+ return counter;
}
int count;
int baseport;
int baseLEport;
- HashMap<Long,QuorumServer> peers;
+ HashMap<Long,QuorumServer> peers;
ArrayList<LEThread> threads;
HashMap<Integer, HashSet<TestVote> > voteMap;
File tmpdir[];
int port[];
int successCount;
Object finalObj;
-
+
volatile Vote votes[];
volatile boolean leaderDies;
volatile long leader = -1;
- //volatile int round = 1;
+ //volatile int round = 1;
Random rand = new Random();
-
+
@Override
public void setUp() throws Exception {
count = 7;
baseport= 33003;
baseLEport = 43003;
-
+
peers = new HashMap<Long,QuorumServer>(count);
threads = new ArrayList<LEThread>(count);
voteMap = new HashMap<Integer, HashSet<TestVote> >();
@@ -88,7 +86,7 @@
port = new int[count];
successCount = 0;
finalObj = new Object();
-
+
LOG.info("SetUp " + getName());
}
@@ -99,9 +97,8 @@
}
LOG.info("FINISHED " + getName());
}
-
+
class LEThread extends Thread {
- FastLeaderElection le;
int i;
QuorumPeer peer;
//int peerRound = 1;
@@ -118,28 +115,28 @@
peer.setPeerState(ServerState.LOOKING);
LOG.info("Going to call leader election again.");
v = peer.getElectionAlg().lookForLeader();
- if(v == null){
+ if(v == null){
LOG.info("Thread " + i + " got a null vote");
break;
}
-
+
/*
* A real zookeeper would take care of setting the current vote. Here
* we do it manually.
*/
peer.setCurrentVote(v);
-
+
LOG.info("Finished election: " + i + ", " + v.id);
votes[i] = v;
-
+
/*
* Get the current value of the logical clock for this peer.
*/
int lc = (int) ((FastLeaderElection) peer.getElectionAlg()).getLogicalClock();
-
+
if (v.id == ((long) i)) {
/*
- * A leader executes this part of the code. If it is the first leader to be
+ * A leader executes this part of the code. If it is the first leader to be
* elected, then it fails right after. Otherwise, it waits until it has enough
* followers supporting it.
*/
@@ -151,22 +148,22 @@
((FastLeaderElection) peer.getElectionAlg()).shutdown();
leader = -1;
LOG.info("Leader " + i + " dead");
-
- //round++;
+
+ //round++;
FLETest.this.notifyAll();
-
+
break;
-
+
} else {
synchronized(voteMap){
if(voteMap.get(lc) == null)
voteMap.put(lc, new HashSet<TestVote>());
HashSet<TestVote> hs = voteMap.get(lc);
hs.add(new TestVote(i, v.id));
-
+
if(countVotes(hs, v.id) > (count/2)){
leader = i;
- LOG.info("Got majority: " + i);
+ LOG.info("Got majority: " + i);
} else {
voteMap.wait(3000);
LOG.info("Notified or expired: " + i);
@@ -175,7 +172,7 @@
leader = i;
LOG.info("Got majority: " + i);
} else {
- //round++;
+ //round++;
}
}
}
@@ -186,45 +183,45 @@
successCount++;
if(successCount > (count/2)) finalObj.notify();
}
-
+
break;
}
}
}
} else {
/*
- * Followers execute this part. They first add their vote to voteMap, and then
+ * Followers execute this part. They first add their vote to voteMap, and then
* they wait for bounded amount of time. A leader notifies followers through the
* FLETest.this object.
- *
+ *
* Note that I can get FLETest.this, and then voteMap before adding the vote of
* a follower, otherwise a follower would be blocked out until the leader notifies
* or leaves the synchronized block on FLEtest.this.
*/
-
-
+
+
LOG.info("Logical clock " + ((FastLeaderElection) peer.getElectionAlg()).getLogicalClock());
synchronized(voteMap){
LOG.info("Voting on " + votes[i].id + ", round " + ((FastLeaderElection) peer.getElectionAlg()).getLogicalClock());
if(voteMap.get(lc) == null)
voteMap.put(lc, new HashSet<TestVote>());
- HashSet<TestVote> hs = voteMap.get(lc);
- hs.add(new TestVote(i, votes[i].id));
+ HashSet<TestVote> hs = voteMap.get(lc);
+ hs.add(new TestVote(i, votes[i].id));
if(countVotes(hs, votes[i].id) > (count/2)){
LOG.info("Logical clock: " + lc + ", " + votes[i].id);
voteMap.notify();
- }
+ }
}
-
+
/*
* In this part a follower waits until the leader notifies it, and remove its
* vote if the leader takes too long to respond.
*/
synchronized(FLETest.this){
if (leader != votes[i].id) FLETest.this.wait(3000);
-
+
LOG.info("The leader: " + leader + " and my vote " + votes[i].id);
- synchronized(voteMap){
+ synchronized(voteMap){
if (leader == votes[i].id) {
synchronized(finalObj){
successCount++;
@@ -257,22 +254,22 @@
}
}
}
-
+
@Test
public void testLE() throws Exception {
-
+
FastLeaderElection le[] = new FastLeaderElection[count];
leaderDies = true;
boolean allowOneBadLeader = leaderDies;
-
+
LOG.info("TestLE: " + getName()+ ", " + count);
for(int i = 0; i < count; i++) {
- peers.put(Long.valueOf(i), new QuorumServer(i, new InetSocketAddress(baseport+100+i),
+ peers.put(Long.valueOf(i), new QuorumServer(i, new InetSocketAddress(baseport+100+i),
new InetSocketAddress(baseLEport+100+i)));
- tmpdir[i] = File.createTempFile("letest", "test");
- port[i] = baseport+i;
+ tmpdir[i] = ClientBase.createTmpDir();
+ port[i] = baseport+i;
}
-
+
for(int i = 0; i < le.length; i++) {
QuorumPeer peer = new QuorumPeer(peers, tmpdir[i], tmpdir[i], port[i], 3, i, 2, 2, 2);
peer.startLeaderElection();
@@ -281,8 +278,8 @@
threads.add(thread);
}
LOG.info("Started threads " + getName());
-
-
+
+
int waitCounter = 0;
synchronized(finalObj){
while((successCount <= count/2) && (waitCounter < 50)){
@@ -290,7 +287,7 @@
waitCounter++;
}
}
-
+
/*
* Lists what threads haven-t joined. A thread doesn't join if it hasn't decided
* upon a leader yet. It can happen that a peer is slow or disconnected, and it can
@@ -301,14 +298,14 @@
LOG.info("Threads didn't join: " + i);
}
}
-
+
/*
* If we have a majority, then we are good to go.
*/
if(successCount <= count/2){
fail("Fewer than a a majority has joined");
}
-
+
if(threads.get((int) leader).isAlive()){
fail("Leader hasn't joined: " + leader);
}
Modified: hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/HierarchicalQuorumTest.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/HierarchicalQuorumTest.java?rev=781264&r1=781263&r2=781264&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/HierarchicalQuorumTest.java (original)
+++ hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/HierarchicalQuorumTest.java Wed Jun 3 05:14:16 2009
@@ -17,77 +17,64 @@
*/
package org.apache.zookeeper.test;
-import java.io.IOException;
-import java.util.Properties;
-
-import java.io.File;
import java.io.ByteArrayInputStream;
+import java.io.File;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
-import java.util.HashSet;
+import java.util.Properties;
import java.util.Random;
-import org.apache.zookeeper.KeeperException;
+import junit.framework.TestCase;
+
+import org.apache.log4j.Logger;
import org.apache.zookeeper.server.quorum.FastLeaderElection;
import org.apache.zookeeper.server.quorum.QuorumPeer;
import org.apache.zookeeper.server.quorum.Vote;
import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
import org.apache.zookeeper.server.quorum.QuorumPeer.ServerState;
import org.apache.zookeeper.server.quorum.flexible.QuorumHierarchical;
-
-import org.apache.log4j.Logger;
-import junit.framework.TestCase;
import org.junit.Before;
import org.junit.Test;
public class HierarchicalQuorumTest extends TestCase {
private static final Logger LOG = Logger.getLogger(HierarchicalQuorumTest.class);
-
- class TestVote{
- TestVote(int id, long leader){
- this.leader = leader;
- this.id = id;
- }
- long leader;
- int id;
- }
-
+
Properties qp;
-
+
int count;
int baseport;
int baseLEport;
- HashMap<Long,QuorumServer> peers;
+ HashMap<Long,QuorumServer> peers;
ArrayList<LEThread> threads;
File tmpdir[];
int port[];
Object finalObj;
-
+
volatile Vote votes[];
volatile boolean leaderDies;
volatile long leader = -1;
Random rand = new Random();
-
-
+
+
@Before
@Override
protected void setUp() throws Exception {
count = 9;
baseport= 33003;
baseLEport = 43003;
-
+
peers = new HashMap<Long,QuorumServer>(count);
threads = new ArrayList<LEThread>(count);
votes = new Vote[count];
tmpdir = new File[count];
port = new int[count];
finalObj = new Object();
-
+
String config = "group.1=0:1:2\n" +
"group.2=3:4:5\n" +
- "group.3=6:7:8\n\n" +
- "weight.0=1\n" +
+ "group.3=6:7:8\n\n" +
+ "weight.0=1\n" +
"weight.1=1\n" +
"weight.2=1\n" +
"weight.3=1\n" +
@@ -96,23 +83,22 @@
"weight.6=1\n" +
"weight.7=1\n" +
"weight.8=1";
-
+
ByteArrayInputStream is = new ByteArrayInputStream(config.getBytes());
- this.qp = new Properties();
+ this.qp = new Properties();
qp.load(is);
-
+
LOG.info("SetUp " + getName());
}
-
+
protected void tearDown() throws Exception {
for(int i = 0; i < threads.size(); i++) {
((FastLeaderElection) threads.get(i).peer.getElectionAlg()).shutdown();
}
LOG.info("FINISHED " + getName());
}
-
+
class LEThread extends Thread {
- FastLeaderElection le;
int i;
QuorumPeer peer;
//int peerRound = 1;
@@ -127,25 +113,25 @@
try {
Vote v = null;
while(true){
-
+
//while(true) {
peer.setPeerState(ServerState.LOOKING);
LOG.info("Going to call leader election.");
v = peer.getElectionAlg().lookForLeader();
- if(v == null){
+ if(v == null){
LOG.info("Thread " + i + " got a null vote");
return;
}
-
+
/*
* A real zookeeper would take care of setting the current vote. Here
* we do it manually.
*/
peer.setCurrentVote(v);
-
+
LOG.info("Finished election: " + i + ", " + v.id);
votes[i] = v;
-
+
if((peer.getPeerState() == ServerState.FOLLOWING) ||
(peer.getPeerState() == ServerState.LEADING)) break;
}
@@ -155,19 +141,19 @@
}
}
}
-
+
@Test
public void testHierarchicalQuorum() throws Exception {
FastLeaderElection le[] = new FastLeaderElection[count];
-
+
LOG.info("TestHierarchicalQuorum: " + getName()+ ", " + count);
for(int i = 0; i < count; i++) {
- peers.put(Long.valueOf(i), new QuorumServer(i, new InetSocketAddress(baseport+100+i),
+ peers.put(Long.valueOf(i), new QuorumServer(i, new InetSocketAddress(baseport+100+i),
new InetSocketAddress(baseLEport+100+i)));
- tmpdir[i] = File.createTempFile("letest", "test");
- port[i] = baseport+i;
+ tmpdir[i] = ClientBase.createTmpDir();
+ port[i] = baseport+i;
}
-
+
for(int i = 0; i < le.length; i++) {
QuorumHierarchical hq = new QuorumHierarchical(qp);
QuorumPeer peer = new QuorumPeer(peers, tmpdir[i], tmpdir[i], port[i], 3, i, 2, 2, 2, hq);
@@ -177,7 +163,7 @@
threads.add(thread);
}
LOG.info("Started threads " + getName());
-
+
for(int i = 0; i < threads.size(); i++) {
threads.get(i).join(15000);
if (threads.get(i).isAlive()) {
Modified: hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/LETest.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/LETest.java?rev=781264&r1=781263&r2=781264&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/LETest.java (original)
+++ hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/LETest.java Wed Jun 3 05:14:16 2009
@@ -34,7 +34,7 @@
public class LETest extends TestCase {
volatile Vote votes[];
volatile boolean leaderDies;
- volatile long leader = -1;
+ volatile long leader = -1;
Random rand = new Random();
class LEThread extends Thread {
LeaderElection le;
@@ -59,7 +59,7 @@
System.out.println("Leader " + i + " dying");
leader = -2;
} else {
- leader = i;
+ leader = i;
}
LETest.this.notifyAll();
}
@@ -92,8 +92,8 @@
votes = new Vote[count];
for(int i = 0; i < count; i++) {
peers.put(Long.valueOf(i), new QuorumServer(i, new InetSocketAddress("127.0.0.1", baseport+100+i)));
- tmpdir[i] = File.createTempFile("letest", "test");
- port[i] = baseport+i;
+ tmpdir[i] = ClientBase.createTmpDir();
+ port[i] = baseport+i;
}
LeaderElection le[] = new LeaderElection[count];
leaderDies = true;
Modified: hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/QuorumBase.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/QuorumBase.java?rev=781264&r1=781263&r2=781264&view=diff
==============================================================================
--- hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/QuorumBase.java (original)
+++ hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/QuorumBase.java Wed Jun 3 05:14:16 2009
@@ -34,18 +34,15 @@
public class QuorumBase extends ClientBase {
private static final Logger LOG = Logger.getLogger(QuorumBase.class);
-
-
File s1dir, s2dir, s3dir, s4dir, s5dir;
QuorumPeer s1, s2, s3, s4, s5;
protected void setUp() throws Exception {
LOG.info("STARTING " + getName());
+ setupTestEnv();
JMXEnv.setUp();
- setupTestEnv();
-
hostPort = "127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183,127.0.0.1:2184,127.0.0.1:2185";
s1dir = ClientBase.createTmpDir();
@@ -108,11 +105,11 @@
// make sure we have these 5 servers listed
Set<String> ensureNames = new LinkedHashSet<String>();
for (int i = 1; i <= 5; i++) {
- ensureNames.add("InMemoryDataTree");
+ ensureNames.add("InMemoryDataTree");
}
for (int i = 1; i <= 5; i++) {
ensureNames.add("name0=ReplicatedServer_id" + i
- + ",name1=replica." + i + ",name2=");
+ + ",name1=replica." + i + ",name2=");
}
for (int i = 1; i <= 5; i++) {
for (int j = 1; j <= 5; j++) {
@@ -121,7 +118,7 @@
}
}
for (int i = 1; i <= 5; i++) {
- ensureNames.add("name0=ReplicatedServer_id" + i);
+ ensureNames.add("name0=ReplicatedServer_id" + i);
}
JMXEnv.ensureAll(ensureNames.toArray(new String[ensureNames.size()]));
}
@@ -144,7 +141,7 @@
}
JMXEnv.tearDown();
-
+
LOG.info("FINISHED " + getName());
}