From 4a8fda7031d68236441b13bd878936b2607c5244 Mon Sep 17 00:00:00 2001 From: Enrico Olivelli - Diennea Date: Thu, 3 Jan 2019 16:32:46 +0100 Subject: [PATCH 01/14] ZOOKEEPER-3217: owasp job flagging slf4j on trunk Disable OWASP checks about slf4j. We are not using EventData, so ZooKeeper is not subject to https://nvd.nist.gov/vuln/detail/CVE-2018-8088 Author: Enrico Olivelli - Diennea Author: Enrico Olivelli Reviewers: phunt@apache.org, andor@apache.org Closes #736 from eolivelli/fix/ZOOKEEPER-3217-owasp and squashes the following commits: 7dd4473a1 [Enrico Olivelli] Add missing license header dc9bd75cd [Enrico Olivelli - Diennea] ZOOKEEPER-3217 owasp job flagging slf4j on trunk --- build.xml | 1 + owaspSuppressions.xml | 25 +++++++++++++++++++++++++ 2 files changed, 26 insertions(+) create mode 100644 owaspSuppressions.xml diff --git a/build.xml b/build.xml index f8a0546740a..50bc94ffc17 100644 --- a/build.xml +++ b/build.xml @@ -1705,6 +1705,7 @@ xmlns:cs="antlib:com.puppycrawl.tools.checkstyle.ant"> reportoutputdirectory="${owasp.out.dir}" reportformat="ALL" failBuildOnCVSS="0"> + diff --git a/owaspSuppressions.xml b/owaspSuppressions.xml new file mode 100644 index 00000000000..0165b9ada13 --- /dev/null +++ b/owaspSuppressions.xml @@ -0,0 +1,25 @@ + + + + + + + + CVE-2018-8088 + + From b752ef66876a141035a42f30aad69e3166cad746 Mon Sep 17 00:00:00 2001 From: Enrico Olivelli Date: Mon, 7 Jan 2019 14:24:07 +0100 Subject: [PATCH 02/14] ZOOKEEPER-3223: Configure Spotbugs - add spotbugs configuration (default) - make build pass spotbugs Author: Enrico Olivelli Reviewers: andor@apache.org Closes #742 from eolivelli/fix/ZOOKEEPER-3223-spotbugs and squashes the following commits: a43cecf41 [Enrico Olivelli] Fix false positive c00d296ad [Enrico Olivelli] Add Suppression for false positive 35c8a4dde [Enrico Olivelli] fix tests 1ae629bcd [Enrico Olivelli] revert file c0bb9d903 [Enrico Olivelli] Add spotbugs annotations to ant based build dabe4fafc [Enrico Olivelli] [ZOOKEEPER-3223] Configure Spotbugs - add spotbugs configuration - make build pass spotbugs --- build.xml | 1 + excludeFindBugsFilter.xml | 14 ++++++++++++ ivy.xml | 1 + pom.xml | 22 ++++++++++++++++++- zookeeper-jute/pom.xml | 8 +++++++ zookeeper-server/pom.xml | 6 +++++ .../java/org/apache/zookeeper/ClientCnxn.java | 4 ++++ .../java/org/apache/zookeeper/ZooDefs.java | 4 ++++ .../org/apache/zookeeper/server/DataNode.java | 2 ++ .../zookeeper/server/EphemeralType.java | 3 +++ .../server/persistence/FileTxnLog.java | 2 +- .../server/quorum/AuthFastLeaderElection.java | 3 +++ .../server/quorum/CommitProcessor.java | 3 +++ .../server/quorum/ObserverMaster.java | 5 +++++ .../zookeeper/server/util/BitHashSet.java | 7 ++++-- .../apache/zookeeper/server/util/BitMap.java | 3 +++ 16 files changed, 84 insertions(+), 4 deletions(-) create mode 100644 excludeFindBugsFilter.xml diff --git a/build.xml b/build.xml index 50bc94ffc17..265a2845b3b 100644 --- a/build.xml +++ b/build.xml @@ -28,6 +28,7 @@ xmlns:cs="antlib:com.puppycrawl.tools.checkstyle.ant"> + diff --git a/excludeFindBugsFilter.xml b/excludeFindBugsFilter.xml new file mode 100644 index 00000000000..c836911dbe9 --- /dev/null +++ b/excludeFindBugsFilter.xml @@ -0,0 +1,14 @@ + + + + + + + + + + + + + + diff --git a/ivy.xml b/ivy.xml index c7f79b66346..a95331780d8 100644 --- a/ivy.xml +++ b/ivy.xml @@ -46,6 +46,7 @@ + diff --git a/pom.xml b/pom.xml index 58c9e50d366..0dc71743c5f 100755 --- a/pom.xml +++ b/pom.xml @@ -276,6 +276,7 @@ 1.1.0 1.56 3.2.2 + 3.1.8 @@ -408,6 +409,13 @@ jline ${jline.version} + + com.github.spotbugs + spotbugs-annotations + ${spotbugsannotations.version} + provided + true + @@ -459,6 +467,14 @@ clover-maven-plugin 4.3.1 + + com.github.spotbugs + spotbugs-maven-plugin + 3.1.8 + + excludeFindBugsFilter.xml + + @@ -486,7 +502,11 @@ - + + com.github.spotbugs + spotbugs-maven-plugin + + diff --git a/zookeeper-jute/pom.xml b/zookeeper-jute/pom.xml index bc133b29b7a..9bb696e9a7d 100755 --- a/zookeeper-jute/pom.xml +++ b/zookeeper-jute/pom.xml @@ -145,6 +145,14 @@ + + + com.github.spotbugs + spotbugs-maven-plugin + + true + + diff --git a/zookeeper-server/pom.xml b/zookeeper-server/pom.xml index b4ee890d2a4..90624693524 100755 --- a/zookeeper-server/pom.xml +++ b/zookeeper-server/pom.xml @@ -34,6 +34,12 @@ ZooKeeper server + + com.github.spotbugs + spotbugs-annotations + provided + true + org.hamcrest hamcrest-all diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/ClientCnxn.java b/zookeeper-server/src/main/java/org/apache/zookeeper/ClientCnxn.java index ef53edf0a8d..db2b4866af0 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/ClientCnxn.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/ClientCnxn.java @@ -18,6 +18,7 @@ package org.apache.zookeeper; +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.io.BufferedReader; import java.io.ByteArrayOutputStream; import java.io.IOException; @@ -97,6 +98,7 @@ * connected to as needed. * */ +@SuppressFBWarnings({"EI_EXPOSE_REP", "EI_EXPOSE_REP2"}) public class ClientCnxn { private static final Logger LOG = LoggerFactory.getLogger(ClientCnxn.class); @@ -479,6 +481,7 @@ public void queueCallback(AsyncCallback cb, int rc, String path, waitingEvents.add(new LocalCallback(cb, rc, path, ctx)); } + @SuppressFBWarnings("JLM_JSR166_UTILCONCURRENT_MONITORENTER") public void queuePacket(Packet packet) { if (wasKilled) { synchronized (waitingEvents) { @@ -495,6 +498,7 @@ public void queueEventOfDeath() { } @Override + @SuppressFBWarnings("JLM_JSR166_UTILCONCURRENT_MONITORENTER") public void run() { try { isRunning = true; diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/ZooDefs.java b/zookeeper-server/src/main/java/org/apache/zookeeper/ZooDefs.java index f685e32e2a7..97aa28a7c7d 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/ZooDefs.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/ZooDefs.java @@ -18,6 +18,7 @@ package org.apache.zookeeper; +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.util.ArrayList; import java.util.Collections; @@ -118,18 +119,21 @@ public interface Ids { /** * This is a completely open ACL . */ + @SuppressFBWarnings(value = "MS_MUTABLE_COLLECTION", justification = "Cannot break API") public final ArrayList OPEN_ACL_UNSAFE = new ArrayList( Collections.singletonList(new ACL(Perms.ALL, ANYONE_ID_UNSAFE))); /** * This ACL gives the creators authentication id's all permissions. */ + @SuppressFBWarnings(value = "MS_MUTABLE_COLLECTION", justification = "Cannot break API") public final ArrayList CREATOR_ALL_ACL = new ArrayList( Collections.singletonList(new ACL(Perms.ALL, AUTH_IDS))); /** * This ACL gives the world the ability to read. */ + @SuppressFBWarnings(value = "MS_MUTABLE_COLLECTION", justification = "Cannot break API") public final ArrayList READ_ACL_UNSAFE = new ArrayList( Collections .singletonList(new ACL(Perms.READ, ANYONE_ID_UNSAFE))); diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/DataNode.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/DataNode.java index 5922d16584f..550b1515295 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/DataNode.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/DataNode.java @@ -18,6 +18,7 @@ package org.apache.zookeeper.server; +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.io.IOException; import java.util.HashSet; import java.util.Set; @@ -36,6 +37,7 @@ * array of ACLs, a stat object, and a set of its children's paths. * */ +@SuppressFBWarnings("EI_EXPOSE_REP2") public class DataNode implements Record { /** the data for this datanode */ byte data[]; diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/EphemeralType.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/EphemeralType.java index f5d58ae8bec..d4c6c8086ed 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/EphemeralType.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/EphemeralType.java @@ -18,6 +18,7 @@ package org.apache.zookeeper.server; +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import org.apache.zookeeper.CreateMode; import java.util.Collections; @@ -210,6 +211,8 @@ public static void validateServerId(long serverId) { * @param ttl ttl * @throws IllegalArgumentException if the ttl is not valid for the mode */ + @SuppressFBWarnings(value = "RV_RETURN_VALUE_IGNORED_NO_SIDE_EFFECT", + justification = "toEphemeralOwner may throw IllegalArgumentException") public static void validateTTL(CreateMode mode, long ttl) { if (mode.isTTL()) { TTL.toEphemeralOwner(ttl); diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/persistence/FileTxnLog.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/persistence/FileTxnLog.java index d95dac84644..c7391527992 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/persistence/FileTxnLog.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/persistence/FileTxnLog.java @@ -180,7 +180,7 @@ public static void setPreallocSize(long size) { * @param serverStats used to update fsyncThresholdExceedCount */ @Override - public void setServerStats(ServerStats serverStats) { + public synchronized void setServerStats(ServerStats serverStats) { this.serverStats = serverStats; } diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/AuthFastLeaderElection.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/AuthFastLeaderElection.java index 93471526d26..933cbfd486c 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/AuthFastLeaderElection.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/AuthFastLeaderElection.java @@ -18,6 +18,7 @@ package org.apache.zookeeper.server.quorum; +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.io.IOException; import java.net.DatagramPacket; import java.net.DatagramSocket; @@ -460,6 +461,8 @@ public void run() { } } + @SuppressFBWarnings(value = "RV_RETURN_VALUE_IGNORED", + justification = "tryAcquire result not chacked, but it is not an issue") private void process(ToSend m) { int attempts = 0; byte zeroes[]; diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/CommitProcessor.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/CommitProcessor.java index 1eb7faca563..9982f150618 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/CommitProcessor.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/CommitProcessor.java @@ -18,6 +18,7 @@ package org.apache.zookeeper.server.quorum; +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.io.IOException; import java.util.HashMap; import java.util.LinkedList; @@ -395,6 +396,7 @@ public void doWork() throws RequestProcessorException { } } + @SuppressFBWarnings("NN_NAKED_NOTIFY") synchronized private void wakeup() { notifyAll(); } @@ -416,6 +418,7 @@ public void commit(Request request) { wakeup(); } + @Override public void processRequest(Request request) { if (stopped) { return; diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/ObserverMaster.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/ObserverMaster.java index 07de57be86a..7308f6597a6 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/ObserverMaster.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/ObserverMaster.java @@ -461,9 +461,14 @@ synchronized public void start() throws IOException { } public void run() { + ServerSocket ss; + synchronized(this) { + ss = this.ss; + } while (listenerRunning) { try { Socket s = ss.accept(); + // start with the initLimit, once the ack is processed // in LearnerHandler switch to the syncLimit s.setSoTimeout(self.tickTime * self.initLimit); diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/util/BitHashSet.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/util/BitHashSet.java index b60f1d47546..a8de793c25d 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/util/BitHashSet.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/util/BitHashSet.java @@ -120,7 +120,10 @@ public synchronized int size() { */ @Override public Iterator iterator() { - if (cache.size() == elementCount) { + // sample current size at the beginning + int currentSize = size(); + + if (cache.size() == currentSize) { return cache.iterator(); } @@ -130,7 +133,7 @@ public Iterator iterator() { @Override public boolean hasNext() { - return returnedCount < elementCount; + return returnedCount < currentSize; } @Override diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/util/BitMap.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/util/BitMap.java index 1a0fb3bac5a..691c5a7293a 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/util/BitMap.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/util/BitMap.java @@ -18,6 +18,7 @@ package org.apache.zookeeper.server.util; +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.util.Map; import java.util.HashMap; import java.util.BitSet; @@ -37,6 +38,8 @@ public class BitMap { private final ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock(); + @SuppressFBWarnings(value = "DLS_DEAD_LOCAL_STORE", + justification = "SpotBugs false positive") public Integer add(T value) { /* * Optimized for code which will add the same value again and again, From bc992480ec938a3fad4b90f75a52dd186e1b968a Mon Sep 17 00:00:00 2001 From: maoling Date: Mon, 7 Jan 2019 14:39:13 +0100 Subject: [PATCH 03/14] ZOOKEEPER-2641: AvgRequestLatency metric improves to be more accurate - some original review historys were included [here ](https://github.com/apache/zookeeper/pull/629) - more details in [ZOOKEEPER-2641](https://issues.apache.org/jira/browse/ZOOKEEPER-2641) Author: maoling Reviewers: andor@apache.org Closes #748 from maoling/ZOOKEEPER-2641 and squashes the following commits: e0d4fc890 [maoling] fix the flaky test in the FourLetterWordsTest.testValidateStatOutput 1739dbf1c [maoling] fix the flaky test in the CommandsTest.testMonitor 01af4002e [maoling] ZOOKEEPER-2641:AvgRequestLatency metric improves to be more accurate --- .../src/main/resources/markdown/zookeeperAdmin.md | 2 +- .../org/apache/zookeeper/server/ServerMetrics.java | 6 +++--- .../org/apache/zookeeper/server/ServerStats.java | 2 +- .../zookeeper/server/ZooKeeperServerBean.java | 2 +- .../zookeeper/server/ZooKeeperServerMXBean.java | 2 +- .../zookeeper/server/command/MonitorCommand.java | 4 ++++ .../zookeeper/server/metric/AvgMinMaxCounter.java | 13 ++++++++----- .../org/apache/zookeeper/server/metric/Metric.java | 2 +- .../zookeeper/server/metric/SimpleCounter.java | 4 ++-- .../apache/zookeeper/server/ServerMetricsTest.java | 10 +++++----- .../apache/zookeeper/server/ServerStatsTest.java | 6 ++---- .../apache/zookeeper/server/admin/CommandsTest.java | 8 ++++++-- .../apache/zookeeper/test/FourLetterWordsTest.java | 2 +- 13 files changed, 36 insertions(+), 27 deletions(-) diff --git a/zookeeper-docs/src/main/resources/markdown/zookeeperAdmin.md b/zookeeper-docs/src/main/resources/markdown/zookeeperAdmin.md index 0e92cf9737e..2b68e38268e 100644 --- a/zookeeper-docs/src/main/resources/markdown/zookeeperAdmin.md +++ b/zookeeper-docs/src/main/resources/markdown/zookeeperAdmin.md @@ -1270,7 +1270,7 @@ Moving forward, Four Letter Words will be deprecated, please use $ echo mntr | nc localhost 2185 zk_version 3.4.0 - zk_avg_latency 0 + zk_avg_latency 0.7561 - be account to four decimal places zk_max_latency 0 zk_min_latency 0 zk_packets_received 70 diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerMetrics.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerMetrics.java index 203d0f69761..c5d82deebbb 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerMetrics.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerMetrics.java @@ -83,12 +83,12 @@ public void reset() { metric.reset(); } - Map getValues() { + Map getValues() { return metric.values(); } - static public Map getAllValues() { - LinkedHashMap m = new LinkedHashMap<>(); + static public Map getAllValues() { + LinkedHashMap m = new LinkedHashMap<>(); for (ServerMetrics metric : ServerMetrics.values()) { m.putAll(metric.getValues()); } diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerStats.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerStats.java index e52ef8e0045..bb5cbb4b757 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerStats.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerStats.java @@ -64,7 +64,7 @@ public long getMinLatency() { return requestLatency.getMin(); } - public long getAvgLatency() { + public double getAvgLatency() { return requestLatency.getAvg(); } diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServerBean.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServerBean.java index 6ac76021b31..cf84b2f9e5d 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServerBean.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServerBean.java @@ -59,7 +59,7 @@ public String getVersion() { return Version.getFullVersion(); } - public long getAvgRequestLatency() { + public double getAvgRequestLatency() { return zks.serverStats().getAvgLatency(); } diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServerMXBean.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServerMXBean.java index d7e50d3ccfc..feb6875870f 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServerMXBean.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServerMXBean.java @@ -41,7 +41,7 @@ public interface ZooKeeperServerMXBean { /** * @return average request latency in ms */ - public long getAvgRequestLatency(); + public double getAvgRequestLatency(); /** * @return max request latency in ms */ diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/command/MonitorCommand.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/command/MonitorCommand.java index b89d5574736..e3ac230cc74 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/command/MonitorCommand.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/command/MonitorCommand.java @@ -84,6 +84,10 @@ public void commandRun() { private void print(String key, long number) { print(key, "" + number); } + + private void print(String key, double number) { + print(key, "" + number); + } private void print(String key, String value) { pw.print("zk_"); diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/metric/AvgMinMaxCounter.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/metric/AvgMinMaxCounter.java index 499c9a0e2d6..3029f9485c5 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/metric/AvgMinMaxCounter.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/metric/AvgMinMaxCounter.java @@ -18,6 +18,7 @@ package org.apache.zookeeper.server.metric; +import java.math.BigDecimal; import java.util.LinkedHashMap; import java.util.Map; import java.util.concurrent.atomic.AtomicLong; @@ -36,7 +37,7 @@ public class AvgMinMaxCounter implements Metric { public AvgMinMaxCounter(String name) { this.name = name; } - + public void addDataPoint(long value) { total.addAndGet(value); count.incrementAndGet(); @@ -58,13 +59,15 @@ private void setMin(long value) { ; } - public long getAvg() { + public double getAvg() { // There is possible race-condition but we don't need the stats to be // extremely accurate. long currentCount = count.get(); long currentTotal = total.get(); if (currentCount > 0) { - return currentTotal / currentCount; + double avgLatency = currentTotal / (double)currentCount; + BigDecimal bg = new BigDecimal(avgLatency); + return bg.setScale(4, BigDecimal.ROUND_HALF_UP).doubleValue(); } return 0; } @@ -102,8 +105,8 @@ public void add(long value) { addDataPoint(value); } - public Map values() { - Map m = new LinkedHashMap(); + public Map values() { + Map m = new LinkedHashMap(); m.put("avg_" + name, this.getAvg()); m.put("min_" + name, this.getMin()); m.put("max_" + name, this.getMax()); diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/metric/Metric.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/metric/Metric.java index c475055d9ca..bc4d1667fb7 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/metric/Metric.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/metric/Metric.java @@ -23,5 +23,5 @@ public interface Metric { void add(long value); void reset(); - Map values(); + Map values(); } diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/metric/SimpleCounter.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/metric/SimpleCounter.java index 4bf80461928..7755247d790 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/metric/SimpleCounter.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/metric/SimpleCounter.java @@ -45,8 +45,8 @@ public long getCount() { } @Override - public Map values() { - Map m = new LinkedHashMap(); + public Map values() { + Map m = new LinkedHashMap(); m.put(name, this.getCount()); return m; } diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/ServerMetricsTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/ServerMetricsTest.java index 80f850b4c99..81407246237 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/server/ServerMetricsTest.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/server/ServerMetricsTest.java @@ -65,19 +65,19 @@ private void testAvgMinMaxCounter(AvgMinMaxCounter metric, int size) { long expectedMax = Arrays.stream(values).max().orElse(0); long expectedSum = Arrays.stream(values).sum(); long expectedCnt = values.length; - long expectedAvg = expectedSum / Math.max(1, expectedCnt); + double expectedAvg = expectedSum / Math.max(1, expectedCnt); - Assert.assertEquals(expectedAvg, metric.getAvg()); + Assert.assertEquals(expectedAvg, metric.getAvg(), (double)200); Assert.assertEquals(expectedMin, metric.getMin()); Assert.assertEquals(expectedMax, metric.getMax()); Assert.assertEquals(expectedCnt, metric.getCount()); Assert.assertEquals(expectedSum, metric.getTotal()); - final Map results = metric.values(); + final Map results = metric.values(); Assert.assertEquals(expectedMax, (long)results.get("max_test")); Assert.assertEquals(expectedMin, (long)results.get("min_test")); Assert.assertEquals(expectedCnt, (long)results.get("cnt_test")); - Assert.assertEquals(expectedAvg, (long)results.get("avg_test")); + Assert.assertEquals(expectedAvg, (double)results.get("avg_test"), (double)200); metric.reset(); } @@ -101,7 +101,7 @@ private void testSimpleCounter(SimpleCounter metric, int size) { long expectedCount = Arrays.stream(values).sum(); Assert.assertEquals(expectedCount, metric.getCount()); - final Map results = metric.values(); + final Map results = metric.values(); Assert.assertEquals(expectedCount, (long)results.get("test")); metric.reset(); diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/ServerStatsTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/ServerStatsTest.java index d28dc8ca2f6..357b73e18ea 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/server/ServerStatsTest.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/server/ServerStatsTest.java @@ -19,7 +19,6 @@ package org.apache.zookeeper.server; import org.apache.zookeeper.ZKTestCase; -import org.apache.zookeeper.common.Time; import org.junit.Assert; import org.junit.Before; import org.junit.Test; @@ -77,8 +76,7 @@ public void testLatencyMetrics() { lessThanOrEqualTo(serverStats.getMaxLatency())); assertThat("Min latency check", 1000L, lessThanOrEqualTo(serverStats.getMinLatency())); - assertThat("Avg latency check", 1500L, - lessThanOrEqualTo(serverStats.getAvgLatency())); + Assert.assertEquals((double)1500, serverStats.getAvgLatency(), (double)200); // When reset... serverStats.resetLatency(); @@ -138,7 +136,7 @@ private void assertAllPacketsZero(ServerStats serverStats) { private void assertAllLatencyZero(ServerStats serverStats) { Assert.assertEquals(0L, serverStats.getMaxLatency()); Assert.assertEquals(0L, serverStats.getMinLatency()); - Assert.assertEquals(0L, serverStats.getAvgLatency()); + Assert.assertEquals((double)0, serverStats.getAvgLatency(), (double)0.00001); } private void assertFsyncThresholdExceedCountZero(ServerStats serverStats) { diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/admin/CommandsTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/admin/CommandsTest.java index fedbe0fe873..9b30c555dc0 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/server/admin/CommandsTest.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/server/admin/CommandsTest.java @@ -171,7 +171,7 @@ public void testIsReadOnly() throws IOException, InterruptedException { public void testMonitor() throws IOException, InterruptedException { ArrayList fields = new ArrayList<>(Arrays.asList( new Field("version", String.class), - new Field("avg_latency", Long.class), + new Field("avg_latency", Double.class), new Field("max_latency", Long.class), new Field("min_latency", Long.class), new Field("packets_received", Long.class), @@ -193,7 +193,11 @@ public void testMonitor() throws IOException, InterruptedException { new Field("local_sessions", Long.class) )); for (String metric : ServerMetrics.getAllValues().keySet()) { - fields.add(new Field(metric, Long.class)); + if (metric.startsWith("avg_")) { + fields.add(new Field(metric, Double.class)); + } else { + fields.add(new Field(metric, Long.class)); + } } Field fieldsArray[] = fields.toArray(new Field[0]); testCommand("monitor", fieldsArray); diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/FourLetterWordsTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/FourLetterWordsTest.java index ad71eabb3bd..8330da23c68 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/test/FourLetterWordsTest.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/FourLetterWordsTest.java @@ -153,7 +153,7 @@ public void testValidateStatOutput() throws Exception { Assert.assertTrue(count >= 2); line = in.readLine(); - Assert.assertTrue(Pattern.matches("^Latency min/avg/max: \\d+/\\d+/\\d+$", line)); + Assert.assertTrue(Pattern.matches("^Latency min/avg/max: \\d+/-?[0-9]*.?[0-9]*/\\d+$", line)); line = in.readLine(); Assert.assertTrue(Pattern.matches("^Received: \\d+$", line)); line = in.readLine(); From 9e309557c02315344d6b39012ff4c9633b54c3d3 Mon Sep 17 00:00:00 2001 From: Fangmin Lyu Date: Mon, 7 Jan 2019 14:55:43 +0100 Subject: [PATCH 04/14] ZOOKEEPER-3203: Tracking the number of non voting followers in ZK Author: Fangmin Lyu Reviewers: andor@apache.org Closes #722 from lvfangmin/ZOOKEEPER-3203 --- .../zookeeper/server/admin/Commands.java | 1 + .../server/command/MonitorCommand.java | 1 + .../zookeeper/server/quorum/Leader.java | 14 +++++++++- .../zookeeper/server/quorum/LeaderBean.java | 9 +++++++ .../zookeeper/server/quorum/LeaderMXBean.java | 5 ++++ .../server/quorum/LeaderBeanTest.java | 26 ++++++++++++++++++- 6 files changed, 54 insertions(+), 2 deletions(-) diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/admin/Commands.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/admin/Commands.java index 29e18456a8b..f1e5500563c 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/admin/Commands.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/admin/Commands.java @@ -369,6 +369,7 @@ public CommandResponse run(ZooKeeperServer zkServer, Map kwargs) response.put("learners", leader.getLearners().size()); response.put("synced_followers", leader.getForwardingFollowers().size()); + response.put("synced_non_voting_followers", leader.getNonVotingFollowers().size()); response.put("synced_observers", leader.getObservingLearners().size()); response.put("pending_syncs", leader.getNumPendingSyncs()); response.put("leader_uptime", leader.getUptime()); diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/command/MonitorCommand.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/command/MonitorCommand.java index e3ac230cc74..aeca32da156 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/command/MonitorCommand.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/command/MonitorCommand.java @@ -73,6 +73,7 @@ public void commandRun() { print("learners", leader.getLearners().size()); print("synced_followers", leader.getForwardingFollowers().size()); + print("synced_non_voting_followers", leader.getNonVotingFollowers().size()); print("pending_syncs", leader.getNumPendingSyncs()); print("last_proposal_size", leader.getProposalStats().getLastBufferSize()); diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/Leader.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/Leader.java index 397ea6d4e86..c284debfc49 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/Leader.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/Leader.java @@ -152,7 +152,19 @@ public List getForwardingFollowers() { } } - private void addForwardingFollower(LearnerHandler lh) { + public List getNonVotingFollowers() { + List nonVotingFollowers = new ArrayList(); + synchronized (forwardingFollowers) { + for (LearnerHandler lh : forwardingFollowers) { + if (!isParticipant(lh.getSid())) { + nonVotingFollowers.add(lh); + } + } + } + return nonVotingFollowers; + } + + void addForwardingFollower(LearnerHandler lh) { synchronized (forwardingFollowers) { forwardingFollowers.add(lh); } diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/LeaderBean.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/LeaderBean.java index 0c3be4a2528..1c178f6bd9d 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/LeaderBean.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/LeaderBean.java @@ -51,6 +51,15 @@ public String followerInfo() { return sb.toString(); } + @Override + public String nonVotingFollowerInfo() { + StringBuilder sb = new StringBuilder(); + for (LearnerHandler handler : leader.getNonVotingFollowers()) { + sb.append(handler.toString()).append("\n"); + } + return sb.toString(); + } + @Override public long getElectionTimeTaken() { return leader.self.getElectionTimeTaken(); diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/LeaderMXBean.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/LeaderMXBean.java index 7a1a439fa02..4aed18608d5 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/LeaderMXBean.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/LeaderMXBean.java @@ -34,6 +34,11 @@ public interface LeaderMXBean extends ZooKeeperServerMXBean { */ public String followerInfo(); + /** + * @return information about current non-voting followers + */ + public String nonVotingFollowerInfo(); + /** * @return time taken for leader election in milliseconds. */ diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/LeaderBeanTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/LeaderBeanTest.java index 69dac1ff186..38539b3f880 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/LeaderBeanTest.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/LeaderBeanTest.java @@ -45,6 +45,7 @@ import java.util.Map; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; import static org.junit.Assert.assertNotEquals; import static org.mockito.Matchers.any; import static org.mockito.Matchers.anyString; @@ -58,6 +59,7 @@ public class LeaderBeanTest { private FileTxnSnapLog fileTxnSnapLog; private LeaderZooKeeperServer zks; private QuorumPeer qp; + private QuorumVerifier quorumVerifierMock; @Before public void setUp() throws IOException, X509Exception { @@ -73,7 +75,7 @@ public void setUp() throws IOException, X509Exception { new InetSocketAddress(clientIP, PortAssignment.unique()), new InetSocketAddress(clientIP, clientPort), LearnerType.PARTICIPANT)); - QuorumVerifier quorumVerifierMock = mock(QuorumVerifier.class); + quorumVerifierMock = mock(QuorumVerifier.class); when(quorumVerifierMock.getAllMembers()).thenReturn(peersView); qp.setQuorumVerifier(quorumVerifierMock, false); @@ -173,12 +175,21 @@ public Object answer(InvocationOnMock invocation) throws Throwable { @Test public void testFollowerInfo() throws IOException { + Map votingMembers = new HashMap(); + votingMembers.put(1L, null); + votingMembers.put(2L, null); + votingMembers.put(3L, null); + when(quorumVerifierMock.getVotingMembers()).thenReturn(votingMembers); + LearnerHandler follower = mock(LearnerHandler.class); when(follower.getLearnerType()).thenReturn(LearnerType.PARTICIPANT); when(follower.toString()).thenReturn("1"); + when(follower.getSid()).thenReturn(1L); leader.addLearnerHandler(follower); + leader.addForwardingFollower(follower); assertEquals("1\n", leaderBean.followerInfo()); + assertEquals("", leaderBean.nonVotingFollowerInfo()); LearnerHandler observer = mock(LearnerHandler.class); when(observer.getLearnerType()).thenReturn(LearnerType.OBSERVER); @@ -186,5 +197,18 @@ public void testFollowerInfo() throws IOException { leader.addLearnerHandler(observer); assertEquals("1\n", leaderBean.followerInfo()); + assertEquals("", leaderBean.nonVotingFollowerInfo()); + + LearnerHandler nonVotingFollower = mock(LearnerHandler.class); + when(nonVotingFollower.getLearnerType()).thenReturn(LearnerType.PARTICIPANT); + when(nonVotingFollower.toString()).thenReturn("5"); + when(nonVotingFollower.getSid()).thenReturn(5L); + leader.addLearnerHandler(nonVotingFollower); + leader.addForwardingFollower(nonVotingFollower); + + String followerInfo = leaderBean.followerInfo(); + assertTrue(followerInfo.contains("1")); + assertTrue(followerInfo.contains("5")); + assertEquals("5\n", leaderBean.nonVotingFollowerInfo()); } } From 36b8711ebfd7ea749d547a52ddcd7d8853931728 Mon Sep 17 00:00:00 2001 From: Colm O hEigeartaigh Date: Mon, 7 Jan 2019 15:34:02 +0100 Subject: [PATCH 05/14] ZOOKEEPER-3197: Improve documentation in ZooKeeperServer.superSecret Author: Colm O hEigeartaigh Reviewers: andor@apache.org Closes #752 from coheigea/ZOOKEEPER-3197 --- .../java/org/apache/zookeeper/server/ZooKeeperServer.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServer.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServer.java index b417a8f095d..20ab023ec5a 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServer.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServer.java @@ -116,8 +116,10 @@ protected enum State { } /** - * This is the secret that we use to generate passwords, for the moment it - * is more of a sanity check. + * This is the secret that we use to generate passwords. For the moment, + * it's more of a checksum that's used in reconnection, which carries no + * security weight, and is treated internally as if it carries no + * security weight. */ static final private long superSecret = 0XB3415C00L; From c358dce653874ae4b97ce26629e3ddba00c8b669 Mon Sep 17 00:00:00 2001 From: Stanislav Knot Date: Tue, 8 Jan 2019 17:21:00 +0100 Subject: [PATCH 06/14] ZOOKEEPER-3210: typo in zookeeperInternals doc Author: Stanislav Knot Reviewers: fangmin@apache.org, andor@apache.org Closes #732 from stanlyDoge/patch-1 --- .../src/main/resources/markdown/zookeeperInternals.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/zookeeper-docs/src/main/resources/markdown/zookeeperInternals.md b/zookeeper-docs/src/main/resources/markdown/zookeeperInternals.md index 84793f127c6..13cde102409 100644 --- a/zookeeper-docs/src/main/resources/markdown/zookeeperInternals.md +++ b/zookeeper-docs/src/main/resources/markdown/zookeeperInternals.md @@ -185,7 +185,7 @@ to the leader, the leader will tell the follower to discard U. A new leader establishes a zxid to start using for new proposals by getting the epoch, e, of the highest zxid it has seen and setting the next zxid to use to be -(e+1, 0), fter the leader syncs with a follower, it will propose a NEW_LEADER +(e+1, 0), after the leader syncs with a follower, it will propose a NEW_LEADER proposal. Once the NEW_LEADER proposal has been committed, the leader will activate and start receiving and issuing proposals. From a5b3114d70d03f70b068b209fe393388f3c77991 Mon Sep 17 00:00:00 2001 From: Colm O hEigeartaigh Date: Wed, 9 Jan 2019 15:09:14 +0100 Subject: [PATCH 07/14] ZOOKEEPER-3235: Enable secure processing and disallow DTDs in the SAXParserFactory It's good security practice to set the secure processing feature on SAXParserFactory and to disallow Doctypes if they aren't needed. Author: Colm O hEigeartaigh Reviewers: andor@apache.org Closes #716 from coheigea/sax_secureproc --- .../src/main/java/org/apache/jute/XmlInputArchive.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/zookeeper-jute/src/main/java/org/apache/jute/XmlInputArchive.java b/zookeeper-jute/src/main/java/org/apache/jute/XmlInputArchive.java index 99e11d10eaf..a4ae9381c3a 100644 --- a/zookeeper-jute/src/main/java/org/apache/jute/XmlInputArchive.java +++ b/zookeeper-jute/src/main/java/org/apache/jute/XmlInputArchive.java @@ -143,6 +143,8 @@ public XmlInputArchive(InputStream in) valList = new ArrayList(); DefaultHandler handler = new XMLParser(valList); SAXParserFactory factory = SAXParserFactory.newInstance(); + factory.setFeature(javax.xml.XMLConstants.FEATURE_SECURE_PROCESSING, Boolean.TRUE); + factory.setFeature("http://apache.org/xml/features/disallow-doctype-decl", true); SAXParser parser = factory.newSAXParser(); parser.parse(in, handler); vLen = valList.size(); From 14eefcb7bbd1cc93d4738998ca7343cd9cae8277 Mon Sep 17 00:00:00 2001 From: Michael Edwards Date: Mon, 14 Jan 2019 07:36:15 -0700 Subject: [PATCH 08/14] ZOOKEEPER-3202: Add timing margin to improve reliability of testClientServerSSL() Allowing just 5 seconds for 3 quorum peers to start and elect a leader is a bit tight, at least when running 4 test processes in parallel inside a (Linux) Docker container on a (non-Linux) laptop. Add up to 10 seconds of extra margin. Author: Michael Edwards Reviewers: andor@apache.org Closes #723 from mkedwards/ZOOKEEPER-3202 --- .../test/java/org/apache/zookeeper/test/ClientSSLTest.java | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/ClientSSLTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/ClientSSLTest.java index a09fc897575..15048341dfd 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/test/ClientSSLTest.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/ClientSSLTest.java @@ -95,6 +95,10 @@ public void testClientServerSSL() throws Exception { mt[i].start(); } + // Add some timing margin for the quorum to elect a leader + // (without this margin, timeouts have been observed in parallel test runs) + ClientBase.waitForServerUp("127.0.0.1:" + clientPorts[0], 2 * TIMEOUT); + // Servers have been set up. Now go test if secure connection is successful. for (int i = 0; i < SERVER_COUNT; i++) { Assert.assertTrue("waiting for server " + i + " being up", From 9828685d20bc99272ec0689a590d27b0f8f7ebb6 Mon Sep 17 00:00:00 2001 From: Brian Nixon Date: Mon, 14 Jan 2019 11:38:53 -0700 Subject: [PATCH 09/14] =?UTF-8?q?ZOOKEEPER-3180:=20Add=20response=20cache?= =?UTF-8?q?=20to=20improve=20the=20throughput=20of=20read=20=E2=80=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit …heavy traffic Introduces a ResponseCache that interacts with ServerCnxn to cache the serialized response for getData requests. Author: Brian Nixon Reviewers: hanm@apache.org, fangmin@apache.org, andor@apache.org Closes #684 from enixon/response-cache and squashes the following commits: c264b88f8 [Brian Nixon] fix the types in ResponseCacheTest::checkCacheStatus to match the recent change 5e36b396a [Brian Nixon] add documentation e413de1a6 [Brian Nixon] remove JMX use from ResponseCacheTest c7a3f8eb0 [Brian Nixon] ZOOKEEPER-3180: Add response cache to improve the throughput of read heavy traffic --- .../main/resources/markdown/zookeeperAdmin.md | 11 ++ .../apache/zookeeper/server/DumbWatcher.java | 7 +- .../server/FinalRequestProcessor.java | 51 +++++--- .../zookeeper/server/NIOServerCnxn.java | 52 ++++++--- .../zookeeper/server/NettyServerCnxn.java | 15 ++- .../zookeeper/server/ResponseCache.java | 84 ++++++++++++++ .../apache/zookeeper/server/ServerCnxn.java | 73 +++++++++--- .../zookeeper/server/ServerMetrics.java | 5 +- .../zookeeper/server/ZooKeeperServer.java | 39 +++++++ .../zookeeper/server/ZooKeeperServerBean.java | 10 ++ .../server/ZooKeeperServerMXBean.java | 3 + .../zookeeper/server/MockServerCnxn.java | 5 +- .../org/apache/zookeeper/test/JMXEnv.java | 45 ++++++++ .../zookeeper/test/ResponseCacheTest.java | 109 ++++++++++++++++++ 14 files changed, 451 insertions(+), 58 deletions(-) create mode 100644 zookeeper-server/src/main/java/org/apache/zookeeper/server/ResponseCache.java create mode 100644 zookeeper-server/src/test/java/org/apache/zookeeper/test/ResponseCacheTest.java diff --git a/zookeeper-docs/src/main/resources/markdown/zookeeperAdmin.md b/zookeeper-docs/src/main/resources/markdown/zookeeperAdmin.md index 2b68e38268e..d808b612a99 100644 --- a/zookeeper-docs/src/main/resources/markdown/zookeeperAdmin.md +++ b/zookeeper-docs/src/main/resources/markdown/zookeeperAdmin.md @@ -685,6 +685,17 @@ property, when available, is noted below. defaults to 1000. This value can only be set as a system property. +* *maxResponseCacheSize* : + (Java system property: **zookeeper.maxResponseCacheSize**) + When set to a positive integer, it determines the size + of the cache that stores the serialized form of recently + read records. Helps save the serialization cost on + popular znodes. The metrics **response_packet_cache_hits** + and **response_packet_cache_misses** can be used to tune + this value to a given workload. The feature is turned on + by default with a value of 400, set to 0 or a negative + integer to turn the feature off. + * *autopurge.snapRetainCount* : (No Java system property) **New in 3.4.0:** diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/DumbWatcher.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/DumbWatcher.java index f384d7c5c76..1f64dd09dcc 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/DumbWatcher.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/DumbWatcher.java @@ -26,8 +26,7 @@ import org.apache.jute.Record; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.proto.ReplyHeader; -import org.apache.zookeeper.server.ServerCnxn; -import org.apache.zookeeper.server.ServerStats; +import org.apache.zookeeper.data.Stat; /** * A empty watcher implementation used in bench and unit test. @@ -58,7 +57,7 @@ public void process(WatchedEvent event) { } void close() { } @Override - public void sendResponse(ReplyHeader h, Record r, String tag) throws IOException { } + public void sendResponse(ReplyHeader h, Record r, String tag, String cacheKey, Stat stat) throws IOException { } @Override public void sendCloseSession() { } @@ -70,7 +69,7 @@ public void sendCloseSession() { } void setSessionId(long sessionId) { } @Override - void sendBuffer(ByteBuffer closeConn) { } + void sendBuffer(ByteBuffer... closeConn) { } @Override void enableRecv() { } diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java index b9427e8f400..d022193e739 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java @@ -168,6 +168,7 @@ public void processRequest(Request request) { zks.decInProcess(); Code err = Code.OK; Record rsp = null; + String path = null; try { if (request.getHdr() != null && request.getHdr().getType() == OpCode.error) { /* @@ -316,7 +317,7 @@ public void processRequest(Request request) { ExistsRequest existsRequest = new ExistsRequest(); ByteBufferInputStream.byteBuffer2Record(request.request, existsRequest); - String path = existsRequest.getPath(); + path = existsRequest.getPath(); if (path.indexOf('\0') != -1) { throw new KeeperException.BadArgumentsException(); } @@ -330,15 +331,16 @@ public void processRequest(Request request) { GetDataRequest getDataRequest = new GetDataRequest(); ByteBufferInputStream.byteBuffer2Record(request.request, getDataRequest); - DataNode n = zks.getZKDatabase().getNode(getDataRequest.getPath()); + path = getDataRequest.getPath(); + DataNode n = zks.getZKDatabase().getNode(path); if (n == null) { throw new KeeperException.NoNodeException(); } PrepRequestProcessor.checkACL(zks, request.cnxn, zks.getZKDatabase().aclForNode(n), ZooDefs.Perms.READ, - request.authInfo, getDataRequest.getPath(), null); + request.authInfo, path, null); Stat stat = new Stat(); - byte b[] = zks.getZKDatabase().getData(getDataRequest.getPath(), stat, + byte b[] = zks.getZKDatabase().getData(path, stat, getDataRequest.getWatch() ? cnxn : null); rsp = new GetDataResponse(b, stat); break; @@ -362,8 +364,9 @@ public void processRequest(Request request) { ByteBufferInputStream.byteBuffer2Record(request.request, getACLRequest); Stat stat = new Stat(); + path = getACLRequest.getPath(); List acl = - zks.getZKDatabase().getACL(getACLRequest.getPath(), stat); + zks.getZKDatabase().getACL(path, stat); rsp = new GetACLResponse(acl, stat); break; } @@ -372,15 +375,16 @@ public void processRequest(Request request) { GetChildrenRequest getChildrenRequest = new GetChildrenRequest(); ByteBufferInputStream.byteBuffer2Record(request.request, getChildrenRequest); - DataNode n = zks.getZKDatabase().getNode(getChildrenRequest.getPath()); + path = getChildrenRequest.getPath(); + DataNode n = zks.getZKDatabase().getNode(path); if (n == null) { throw new KeeperException.NoNodeException(); } PrepRequestProcessor.checkACL(zks, request.cnxn, zks.getZKDatabase().aclForNode(n), ZooDefs.Perms.READ, - request.authInfo, getChildrenRequest.getPath(), null); + request.authInfo, path, null); List children = zks.getZKDatabase().getChildren( - getChildrenRequest.getPath(), null, getChildrenRequest + path, null, getChildrenRequest .getWatch() ? cnxn : null); rsp = new GetChildrenResponse(children); break; @@ -391,15 +395,16 @@ public void processRequest(Request request) { ByteBufferInputStream.byteBuffer2Record(request.request, getChildren2Request); Stat stat = new Stat(); - DataNode n = zks.getZKDatabase().getNode(getChildren2Request.getPath()); + path = getChildren2Request.getPath(); + DataNode n = zks.getZKDatabase().getNode(path); if (n == null) { throw new KeeperException.NoNodeException(); } PrepRequestProcessor.checkACL(zks, request.cnxn, zks.getZKDatabase().aclForNode(n), ZooDefs.Perms.READ, - request.authInfo, getChildren2Request.getPath(), null); + request.authInfo, path, null); List children = zks.getZKDatabase().getChildren( - getChildren2Request.getPath(), stat, getChildren2Request + path, stat, getChildren2Request .getWatch() ? cnxn : null); rsp = new GetChildren2Response(children, stat); break; @@ -410,11 +415,12 @@ public void processRequest(Request request) { ByteBufferInputStream.byteBuffer2Record(request.request, checkWatches); WatcherType type = WatcherType.fromInt(checkWatches.getType()); + path = checkWatches.getPath(); boolean containsWatcher = zks.getZKDatabase().containsWatcher( - checkWatches.getPath(), type, cnxn); + path, type, cnxn); if (!containsWatcher) { String msg = String.format(Locale.ENGLISH, "%s (type: %s)", - checkWatches.getPath(), type); + path, type); throw new KeeperException.NoWatcherException(msg); } break; @@ -425,11 +431,12 @@ public void processRequest(Request request) { ByteBufferInputStream.byteBuffer2Record(request.request, removeWatches); WatcherType type = WatcherType.fromInt(removeWatches.getType()); + path = removeWatches.getPath(); boolean removed = zks.getZKDatabase().removeWatch( - removeWatches.getPath(), type, cnxn); + path, type, cnxn); if (!removed) { String msg = String.format(Locale.ENGLISH, "%s (type: %s)", - removeWatches.getPath(), type); + path, type); throw new KeeperException.NoWatcherException(msg); } break; @@ -468,7 +475,19 @@ public void processRequest(Request request) { updateStats(request, lastOp, lastZxid); try { - cnxn.sendResponse(hdr, rsp, "response"); + if (request.type == OpCode.getData && path != null && rsp != null) { + // Serialized read responses could be cached by the connection object. + // Cache entries are identified by their path and last modified zxid, + // so these values are passed along with the response. + GetDataResponse getDataResponse = (GetDataResponse)rsp; + Stat stat = null; + if (getDataResponse != null && getDataResponse.getStat() != null) { + stat = getDataResponse.getStat(); + } + cnxn.sendResponse(hdr, rsp, "response", path, stat); + } else { + cnxn.sendResponse(hdr, rsp, "response"); + } if (request.type == OpCode.closeSession) { cnxn.sendCloseSession(); } diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/NIOServerCnxn.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/NIOServerCnxn.java index b48eb3dc3bf..c2ab78487a8 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/NIOServerCnxn.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/NIOServerCnxn.java @@ -19,7 +19,6 @@ package org.apache.zookeeper.server; import java.io.BufferedWriter; -import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.PrintWriter; import java.io.Writer; @@ -36,10 +35,10 @@ import java.util.concurrent.atomic.AtomicInteger; import org.apache.jute.BinaryInputArchive; -import org.apache.jute.BinaryOutputArchive; import org.apache.jute.Record; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.data.Id; +import org.apache.zookeeper.data.Stat; import org.apache.zookeeper.proto.ReplyHeader; import org.apache.zookeeper.proto.RequestHeader; import org.apache.zookeeper.proto.WatcherEvent; @@ -137,12 +136,17 @@ void sendBufferSync(ByteBuffer bb) { * sendBuffer pushes a byte buffer onto the outgoing buffer queue for * asynchronous writes. */ - public void sendBuffer(ByteBuffer bb) { + public void sendBuffer(ByteBuffer... buffers) { if (LOG.isTraceEnabled()) { LOG.trace("Add a buffer to outgoingBuffers, sk " + sk + " is valid: " + sk.isValid()); } - outgoingBuffers.add(bb); + synchronized (outgoingBuffers) { + for (ByteBuffer buffer : buffers) { + outgoingBuffers.add(buffer); + } + outgoingBuffers.add(packetSentinel); + } requestInterestOpsUpdate(); } @@ -221,10 +225,12 @@ void handleWrite(SelectionKey k) throws IOException, CloseRequestException { if (bb == ServerCnxnFactory.closeConn) { throw new CloseRequestException("close requested"); } + if (bb == packetSentinel) { + packetSent(); + } if (bb.remaining() > 0) { break; } - packetSent(); outgoingBuffers.remove(); } } else { @@ -269,6 +275,9 @@ void handleWrite(SelectionKey k) throws IOException, CloseRequestException { if (bb == ServerCnxnFactory.closeConn) { throw new CloseRequestException("close requested"); } + if (bb == packetSentinel) { + packetSent(); + } if (sent < bb.remaining()) { /* * We only partially sent this buffer, so we update @@ -277,7 +286,6 @@ void handleWrite(SelectionKey k) throws IOException, CloseRequestException { bb.position(bb.position() + sent); break; } - packetSent(); /* We've sent the whole buffer, so drop the buffer */ sent -= bb.remaining(); outgoingBuffers.remove(); @@ -648,16 +656,34 @@ public static void closeSock(SocketChannel sock) { } } - /* - * (non-Javadoc) + private final static ByteBuffer packetSentinel = ByteBuffer.allocate(0); + + /** + * Serializes a ZooKeeper response and enqueues it for sending. + * + * Serializes client response parts and enqueues them into outgoing queue. + * + * If both cache key and last modified zxid are provided, the serialized + * response is caсhed under the provided key, the last modified zxid is + * stored along with the value. A cache entry is invalidated if the + * provided last modified zxid is more recent than the stored one. + * + * Attention: this function is not thread safe, due to caching not being + * thread safe. * - * @see org.apache.zookeeper.server.ServerCnxnIface#sendResponse(org.apache.zookeeper.proto.ReplyHeader, - * org.apache.jute.Record, java.lang.String) + * @param h reply header + * @param r reply payload, can be null + * @param tag Jute serialization tag, can be null + * @param cacheKey key for caching the serialized payload. a null value + * prvents caching + * @param stat stat information for the the reply payload, used + * for cache invalidation. a value of 0 prevents caching. */ @Override - public void sendResponse(ReplyHeader h, Record r, String tag) { + public void sendResponse(ReplyHeader h, Record r, String tag, + String cacheKey, Stat stat) { try { - super.sendResponse(h, r, tag); + sendBuffer(serialize(h, r, tag, cacheKey, stat)); decrOutstandingAndCheckThrottle(h); } catch(Exception e) { LOG.warn("Unexpected exception. Destruction averted.", e); @@ -682,7 +708,7 @@ public void process(WatchedEvent event) { // Convert WatchedEvent to a type that can be sent over the wire WatcherEvent e = event.getWrapper(); - sendResponse(h, e, "notification"); + sendResponse(h, e, "notification", null, null); } /* diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/NettyServerCnxn.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/NettyServerCnxn.java index 311d3c1d204..b6bb343f49d 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/NettyServerCnxn.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/NettyServerCnxn.java @@ -39,6 +39,7 @@ import io.netty.util.ReferenceCountUtil; import org.apache.jute.BinaryInputArchive; import org.apache.jute.Record; +import org.apache.zookeeper.data.Stat; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.proto.ReplyHeader; import org.apache.zookeeper.proto.WatcherEvent; @@ -160,12 +161,14 @@ public void process(WatchedEvent event) { } @Override - public void sendResponse(ReplyHeader h, Record r, String tag) - throws IOException { + public void sendResponse(ReplyHeader h, Record r, String tag, + String cacheKey, Stat stat) throws IOException { + // cacheKey and stat are used in caching, which is not + // implemented here. Implementation example can be found in NIOServerCnxn. if (closingChannel || !channel.isOpen()) { return; } - super.sendResponse(h, r, tag); + sendBuffer(serialize(h, r, tag, cacheKey, stat)); decrOutstandingAndCheckThrottle(h); } @@ -176,12 +179,12 @@ public void setSessionId(long sessionId) { } @Override - public void sendBuffer(ByteBuffer sendBuffer) { - if (sendBuffer == ServerCnxnFactory.closeConn) { + public void sendBuffer(ByteBuffer... buffers) { + if (buffers.length == 1 && buffers[0] == ServerCnxnFactory.closeConn) { close(); return; } - channel.writeAndFlush(Unpooled.wrappedBuffer(sendBuffer)).addListener(f -> { + channel.writeAndFlush(Unpooled.wrappedBuffer(buffers)).addListener(f -> { if (f.isSuccess()) { packetSent(); } diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ResponseCache.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ResponseCache.java new file mode 100644 index 00000000000..73db7d58026 --- /dev/null +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ResponseCache.java @@ -0,0 +1,84 @@ +/** + * 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.Collections; +import java.util.LinkedHashMap; +import java.util.Map; + +import org.apache.zookeeper.data.Stat; + +@SuppressWarnings("serial") +public class ResponseCache { + // Magic number chosen to be "big enough but not too big" + private static final int DEFAULT_RESPONSE_CACHE_SIZE = 400; + + private static class Entry { + public Stat stat; + public byte[] data; + } + + private Map cache = Collections.synchronizedMap( + new LRUCache(getResponseCacheSize())); + + public ResponseCache() { + } + + public void put(String path, byte[] data, Stat stat) { + Entry entry = new Entry(); + entry.data = data; + entry.stat = stat; + cache.put(path, entry); + } + + public byte[] get(String key, Stat stat) { + Entry entry = cache.get(key); + if (entry == null) { + return null; + } + if (!stat.equals(entry.stat)) { + // The node has been modified, invalidate cache. + cache.remove(key); + return null; + } else { + return entry.data; + } + } + + private static int getResponseCacheSize() { + return Integer.getInteger("zookeeper.maxResponseCacheSize", DEFAULT_RESPONSE_CACHE_SIZE); + } + + public static boolean isEnabled() { + return getResponseCacheSize() > 0; + } + + private static class LRUCache extends LinkedHashMap { + private int cacheSize; + + LRUCache(int cacheSize) { + super(cacheSize/4); + this.cacheSize = cacheSize; + } + + protected boolean removeEldestEntry(Map.Entry eldest) { + return size() >= cacheSize; + } + } +} diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerCnxn.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerCnxn.java index 8e145cbeb13..b0088d1fba7 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerCnxn.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerCnxn.java @@ -37,9 +37,11 @@ import org.apache.jute.BinaryOutputArchive; import org.apache.jute.Record; +import org.apache.zookeeper.Quotas; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.data.Id; +import org.apache.zookeeper.data.Stat; import org.apache.zookeeper.proto.ReplyHeader; import org.apache.zookeeper.proto.RequestHeader; import org.slf4j.Logger; @@ -103,25 +105,64 @@ public void decrOutstandingAndCheckThrottle(ReplyHeader h) { abstract void close(); + public abstract void sendResponse(ReplyHeader h, Record r, + String tag, String cacheKey, Stat stat) throws IOException; + public void sendResponse(ReplyHeader h, Record r, String tag) throws IOException { - ByteArrayOutputStream baos = new ByteArrayOutputStream(); - // Make space for length + sendResponse(h, r, tag, null, null); + } + + protected byte[] serializeRecord(Record record) throws IOException { + ByteArrayOutputStream baos = new ByteArrayOutputStream( + ZooKeeperServer.intBufferStartingSizeBytes); BinaryOutputArchive bos = BinaryOutputArchive.getArchive(baos); - try { - baos.write(fourBytes); - bos.writeRecord(h, "header"); - if (r != null) { - bos.writeRecord(r, tag); + bos.writeRecord(record, null); + return baos.toByteArray(); + } + + protected ByteBuffer[] serialize(ReplyHeader h, Record r, String tag, + String cacheKey, Stat stat) throws IOException { + byte[] header = serializeRecord(h); + byte[] data = null; + if (r != null) { + ResponseCache cache = zkServer.getReadResponseCache(); + if (cache != null && stat != null && cacheKey != null && + !cacheKey.endsWith(Quotas.statNode)) { + // Use cache to get serialized data. + // + // NB: Tag is ignored both during cache lookup and serialization, + // since is is not used in read responses, which are being cached. + data = cache.get(cacheKey, stat); + if (data == null) { + // Cache miss, serialize the response and put it in cache. + data = serializeRecord(r); + cache.put(cacheKey, data, stat); + ServerMetrics.RESPONSE_PACKET_CACHE_MISSING.add(1); + } else { + ServerMetrics.RESPONSE_PACKET_CACHE_HITS.add(1); + } + } else { + data = serializeRecord(r); } - baos.close(); - } catch (IOException e) { - LOG.error("Error serializing response"); } - byte b[] = baos.toByteArray(); - serverStats().updateClientResponseSize(b.length - 4); - ByteBuffer bb = ByteBuffer.wrap(b); - bb.putInt(b.length - 4).rewind(); - sendBuffer(bb); + int dataLength = data == null ? 0 : data.length; + int packetLength = header.length + dataLength; + ServerStats serverStats = serverStats(); + if (serverStats != null) { + serverStats.updateClientResponseSize(packetLength); + } + ByteBuffer lengthBuffer = ByteBuffer.allocate(4).putInt(packetLength); + lengthBuffer.rewind(); + + int bufferLen = data != null ? 3 : 2; + ByteBuffer[] buffers = new ByteBuffer[bufferLen]; + + buffers[0] = lengthBuffer; + buffers[1] = ByteBuffer.wrap(header); + if (data != null) { + buffers[2] = ByteBuffer.wrap(data); + } + return buffers; } /* notify the client the session is closing and close/cleanup socket */ @@ -146,7 +187,7 @@ public boolean removeAuthInfo(Id id) { return authInfo.remove(id); } - abstract void sendBuffer(ByteBuffer closeConn); + abstract void sendBuffer(ByteBuffer... buffers); abstract void enableRecv(); diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerMetrics.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerMetrics.java index c5d82deebbb..3420b88e8b8 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerMetrics.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerMetrics.java @@ -67,7 +67,10 @@ public enum ServerMetrics { SNAP_COUNT(new SimpleCounter("snap_count")), COMMIT_COUNT(new SimpleCounter("commit_count")), CONNECTION_REQUEST_COUNT(new SimpleCounter("connection_request_count")), - BYTES_RECEIVED_COUNT(new SimpleCounter("bytes_received_count")); + BYTES_RECEIVED_COUNT(new SimpleCounter("bytes_received_count")), + + RESPONSE_PACKET_CACHE_HITS(new SimpleCounter("response_packet_cache_hits")), + RESPONSE_PACKET_CACHE_MISSING(new SimpleCounter("response_packet_cache_misses")); private final Metric metric; diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServer.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServer.java index 20ab023ec5a..833c79bab0d 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServer.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServer.java @@ -106,10 +106,12 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider { protected SessionTracker sessionTracker; private FileTxnSnapLog txnLogFactory = null; private ZKDatabase zkDb; + private ResponseCache readResponseCache; 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; + private boolean isResponseCachingEnabled = true; protected enum State { INITIAL, RUNNING, SHUTDOWN, ERROR @@ -138,6 +140,30 @@ protected enum State { private ZooKeeperServerShutdownHandler zkShutdownHandler; private volatile int createSessionTrackerServerId = 1; + /** + * Starting size of read and write ByteArroyOuputBuffers. Default is 32 bytes. + * Flag not used for small transfers like connectResponses. + */ + public static final String INT_BUFFER_STARTING_SIZE_BYTES = + "zookeeper.intBufferStartingSizeBytes"; + public static final int DEFAULT_STARTING_BUFFER_SIZE = 1024; + public static final int intBufferStartingSizeBytes; + + static { + intBufferStartingSizeBytes = Integer.getInteger( + INT_BUFFER_STARTING_SIZE_BYTES, + DEFAULT_STARTING_BUFFER_SIZE); + + if (intBufferStartingSizeBytes < 32) { + String msg = "Buffer starting size must be greater than 0." + + "Configure with \"-Dzookeeper.intBufferStartingSizeBytes=\" "; + LOG.error(msg); + throw new IllegalArgumentException(msg); + } + + LOG.info(INT_BUFFER_STARTING_SIZE_BYTES + " = " + intBufferStartingSizeBytes); + } + void removeCnxn(ServerCnxn cnxn) { zkDb.removeCnxn(cnxn); } @@ -170,6 +196,7 @@ public ZooKeeperServer(FileTxnSnapLog txnLogFactory, int tickTime, setMinSessionTimeout(minSessionTimeout); setMaxSessionTimeout(maxSessionTimeout); listener = new ZooKeeperServerListenerImpl(this); + readResponseCache = new ResponseCache(); LOG.info("Created server with tickTime " + tickTime + " minSessionTimeout " + getMinSessionTimeout() + " maxSessionTimeout " + getMaxSessionTimeout() @@ -1282,4 +1309,16 @@ public Map> getSessionExpiryMap() { void registerServerShutdownHandler(ZooKeeperServerShutdownHandler zkShutdownHandler) { this.zkShutdownHandler = zkShutdownHandler; } + + public boolean isResponseCachingEnabled() { + return isResponseCachingEnabled; + } + + public void setResponseCachingEnabled(boolean isEnabled) { + isResponseCachingEnabled = isEnabled; + } + + public ResponseCache getReadResponseCache() { + return isResponseCachingEnabled ? readResponseCache : null; + } } diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServerBean.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServerBean.java index cf84b2f9e5d..deae98d9b40 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServerBean.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServerBean.java @@ -197,4 +197,14 @@ public int getMinClientResponseSize() { public int getMaxClientResponseSize() { return zks.serverStats().getClientResponseStats().getMaxBufferSize(); } + + @Override + public boolean getResponseCachingEnabled() { + return zks.isResponseCachingEnabled(); + } + + @Override + public void setResponseCachingEnabled(boolean isEnabled) { + zks.setResponseCachingEnabled(isEnabled); + } } diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServerMXBean.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServerMXBean.java index feb6875870f..bd4d3498d2e 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServerMXBean.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServerMXBean.java @@ -95,6 +95,9 @@ public interface ZooKeeperServerMXBean { */ public void setMaxSessionTimeout(int max); + public boolean getResponseCachingEnabled(); + public void setResponseCachingEnabled(boolean isEnabled); + /** * Reset packet and latency statistics */ diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/MockServerCnxn.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/MockServerCnxn.java index 20cf36dc88c..a8fdeaf7ddf 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/server/MockServerCnxn.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/server/MockServerCnxn.java @@ -24,6 +24,7 @@ import org.apache.jute.Record; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.proto.ReplyHeader; +import org.apache.zookeeper.data.Stat; public class MockServerCnxn extends ServerCnxn { public Certificate[] clientChain; @@ -43,7 +44,7 @@ void close() { } @Override - public void sendResponse(ReplyHeader h, Record r, String tag) + public void sendResponse(ReplyHeader h, Record r, String tag, String cacheKey, Stat stat) throws IOException { } @@ -80,7 +81,7 @@ public void setClientCertificateChain(Certificate[] chain) { } @Override - void sendBuffer(ByteBuffer closeConn) { + void sendBuffer(ByteBuffer... closeConn) { } @Override diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/JMXEnv.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/JMXEnv.java index 4edcc0eb123..d8a923a865a 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/test/JMXEnv.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/JMXEnv.java @@ -19,9 +19,12 @@ package org.apache.zookeeper.test; import java.io.IOException; +import java.util.ArrayList; import java.util.Arrays; import java.util.HashSet; +import java.util.List; import java.util.Set; +import java.util.regex.Pattern; import javax.management.MBeanServer; import javax.management.MBeanServerConnection; @@ -319,4 +322,46 @@ private static boolean compare(String bean, String name) { } return false; } + + static Pattern standaloneRegEx = Pattern.compile( + "^org.apache.ZooKeeperService:name0=StandaloneServer_port-?\\d+$" + ); + static Pattern instanceRegEx = Pattern.compile( + "^org.apache.ZooKeeperService:name0=ReplicatedServer_id(\\d+)" + + ",name1=replica.(\\d+),name2=(Follower|Leader)$" + ); + static Pattern observerRegEx = Pattern.compile( + "^org.apache.ZooKeeperService:name0=ReplicatedServer_id(-?\\d+)" + + ",name1=replica.(-?\\d+),name2=(StandaloneServer_port-?\\d+)$" + ); + static List beanPatterns = Arrays.asList(standaloneRegEx, instanceRegEx, observerRegEx); + + public static List getServerBeans() throws IOException { + ArrayList serverBeans = new ArrayList<>(); + Set beans; + try { + beans = conn().queryNames( + new ObjectName(CommonNames.DOMAIN + ":*"), null); + } catch (MalformedObjectNameException e) { + throw new RuntimeException(e); + } + for (ObjectName bean : beans) { + String name = bean.toString(); + LOG.info("bean:" + name); + for (Pattern pattern : beanPatterns) { + if (pattern.matcher(name).find()) { + serverBeans.add(bean); + } + } + } + return serverBeans; + } + + public static ObjectName getServerBean() throws Exception { + List serverBeans = getServerBeans(); + if (serverBeans.size() != 1) { + throw new RuntimeException("Unable to find one and only one server bean"); + } + return serverBeans.get(0); + } } diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/ResponseCacheTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/ResponseCacheTest.java new file mode 100644 index 00000000000..e220c61828d --- /dev/null +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/ResponseCacheTest.java @@ -0,0 +1,109 @@ +/** + * 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.test; + +import java.util.Map; + +import org.apache.zookeeper.CreateMode; +import org.apache.zookeeper.ZooDefs; +import org.apache.zookeeper.ZooKeeper; +import org.apache.zookeeper.data.Stat; +import org.apache.zookeeper.server.ServerMetrics; +import org.junit.Assert; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class ResponseCacheTest extends ClientBase { + protected static final Logger LOG = + LoggerFactory.getLogger(ResponseCacheTest.class); + + @Test + public void testResponseCache() throws Exception { + ZooKeeper zk = createClient(); + + try { + performCacheTest(zk, "/cache", true); + performCacheTest(zk, "/nocache", false); + } + finally { + zk.close(); + } + } + + private void checkCacheStatus(long expectedHits, long expectedMisses) { + Map metrics = ServerMetrics.getAllValues(); + Assert.assertEquals(expectedHits, metrics.get("response_packet_cache_hits")); + Assert.assertEquals(expectedMisses, metrics.get("response_packet_cache_misses")); + } + + public void performCacheTest(ZooKeeper zk, String path, boolean useCache) throws Exception { + ServerMetrics.resetAll(); + Stat writeStat = new Stat(); + Stat readStat = new Stat(); + byte[] readData = null; + int reads = 10; + long expectedHits = 0; + long expectedMisses = 0; + + getServer(serverFactory).setResponseCachingEnabled(useCache); + LOG.info("caching: {}", useCache); + + byte[] writeData = "test1".getBytes(); + zk.create(path, writeData, ZooDefs.Ids.OPEN_ACL_UNSAFE, + CreateMode.PERSISTENT, writeStat); + for (int i = 0; i < reads; ++i) { + readData = zk.getData(path, false, readStat); + Assert.assertArrayEquals(writeData, readData); + Assert.assertEquals(writeStat, readStat); + } + if (useCache) { + expectedMisses += 1; + expectedHits += reads - 1; + } + checkCacheStatus(expectedHits, expectedMisses); + + writeData = "test2".getBytes(); + writeStat = zk.setData(path, writeData, -1); + for (int i = 0; i < 10; ++i) { + readData = zk.getData(path, false, readStat); + Assert.assertArrayEquals(writeData, readData); + Assert.assertEquals(writeStat, readStat); + } + if (useCache) { + expectedMisses += 1; + expectedHits += reads - 1; + } + checkCacheStatus(expectedHits, expectedMisses); + + // Create a child beneath the tested node. This won't change the data of + // the tested node, but will change it's pzxid. The next read of the tested + // node should miss in the cache. The data should still match what was written + // before, but the stat information should not. + zk.create(path + "/child", "child".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, + CreateMode.PERSISTENT, null); + readData = zk.getData(path, false, readStat); + if (useCache) { + expectedMisses++; + } + Assert.assertArrayEquals(writeData, readData); + Assert.assertNotSame(writeStat, readStat); + checkCacheStatus(expectedHits, expectedMisses); + } +} From 38fab85ab476ffea3dd9668eac6fd6e09a190a09 Mon Sep 17 00:00:00 2001 From: Ilya Maykov Date: Mon, 14 Jan 2019 11:40:19 -0700 Subject: [PATCH 10/14] ZOOKEEPER-3195: TLS - disable client-initiated renegotiation Summary: client-initiated renegotiation is insecure and is vulnerable to MITM attacks. Unfortunately, the feature is enabled in Java by default. This disables it. See https://bugs.openjdk.java.net/browse/JDK-7188658 and https://www.oracle.com/technetwork/java/javase/documentation/tlsreadme-141115.html Test Plan: manually tested by running a secure ZK server and probing the listening port with python's sslyze tool (using `sslyze --reneg ...`). Tested on Java 8, 9, 10, and 11. Author: Ilya Maykov Reviewers: andor@apache.org Closes #710 from ivmaykov/ZOOKEEPER-3195 --- .../org/apache/zookeeper/common/X509Util.java | 15 +++ .../apache/zookeeper/common/X509UtilTest.java | 95 ++++++++++++++++++- 2 files changed, 109 insertions(+), 1 deletion(-) diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/common/X509Util.java b/zookeeper-server/src/main/java/org/apache/zookeeper/common/X509Util.java index 4ea105b3e13..310f3e66fa3 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/common/X509Util.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/common/X509Util.java @@ -66,6 +66,21 @@ public abstract class X509Util implements Closeable, AutoCloseable { private static final Logger LOG = LoggerFactory.getLogger(X509Util.class); + private static final String REJECT_CLIENT_RENEGOTIATION_PROPERTY = + "jdk.tls.rejectClientInitiatedRenegotiation"; + static { + // Client-initiated renegotiation in TLS is unsafe and + // allows MITM attacks, so we should disable it unless + // it was explicitly enabled by the user. + // A brief summary of the issue can be found at + // https://www.ietf.org/proceedings/76/slides/tls-7.pdf + if (System.getProperty(REJECT_CLIENT_RENEGOTIATION_PROPERTY) == null) { + LOG.info("Setting -D {}=true to disable client-initiated TLS renegotiation", + REJECT_CLIENT_RENEGOTIATION_PROPERTY); + System.setProperty(REJECT_CLIENT_RENEGOTIATION_PROPERTY, Boolean.TRUE.toString()); + } + } + static final String DEFAULT_PROTOCOL = "TLSv1.2"; private static final String[] DEFAULT_CIPHERS_JAVA8 = { "TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA256", diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/common/X509UtilTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/common/X509UtilTest.java index 1058010febb..9c4a9b0f0b9 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/common/X509UtilTest.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/common/X509UtilTest.java @@ -17,10 +17,24 @@ */ package org.apache.zookeeper.common; +import java.io.IOException; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.Socket; import java.security.Security; import java.util.Collection; - +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicInteger; + +import javax.net.ssl.HandshakeCompletedEvent; +import javax.net.ssl.HandshakeCompletedListener; import javax.net.ssl.SSLContext; +import javax.net.ssl.SSLHandshakeException; import javax.net.ssl.SSLServerSocket; import javax.net.ssl.SSLSocket; import javax.net.ssl.X509KeyManager; @@ -389,6 +403,85 @@ public void testGetSslHandshakeDetectionTimeoutMillisProperty() { } } + private static void forceClose(Socket s) { + if (s == null || s.isClosed()) { + return; + } + try { + s.close(); + } catch (IOException e) { + } + } + + private static void forceClose(ServerSocket s) { + if (s == null || s.isClosed()) { + return; + } + try { + s.close(); + } catch (IOException e) { + } + } + + // This test makes sure that client-initiated TLS renegotiation does not + // succeed. We explicitly disable it at the top of X509Util.java. + @Test(expected = SSLHandshakeException.class) + public void testClientRenegotiationFails() throws Throwable { + int port = PortAssignment.unique(); + ExecutorService workerPool = Executors.newCachedThreadPool(); + final SSLServerSocket listeningSocket = x509Util.createSSLServerSocket(); + SSLSocket clientSocket = null; + SSLSocket serverSocket = null; + final AtomicInteger handshakesCompleted = new AtomicInteger(0); + try { + InetSocketAddress localServerAddress = new InetSocketAddress( + InetAddress.getLoopbackAddress(), port); + listeningSocket.bind(localServerAddress); + Future acceptFuture; + acceptFuture = workerPool.submit(new Callable() { + @Override + public SSLSocket call() throws Exception { + SSLSocket sslSocket = (SSLSocket) listeningSocket.accept(); + sslSocket.addHandshakeCompletedListener(new HandshakeCompletedListener() { + @Override + public void handshakeCompleted(HandshakeCompletedEvent handshakeCompletedEvent) { + handshakesCompleted.getAndIncrement(); + } + }); + Assert.assertEquals(1, sslSocket.getInputStream().read()); + try { + // 2nd read is after the renegotiation attempt and will fail + sslSocket.getInputStream().read(); + return sslSocket; + } catch (Exception e) { + forceClose(sslSocket); + throw e; + } + } + }); + clientSocket = x509Util.createSSLSocket(); + clientSocket.connect(localServerAddress); + clientSocket.getOutputStream().write(1); + // Attempt to renegotiate after establishing the connection + clientSocket.startHandshake(); + clientSocket.getOutputStream().write(1); + // The exception is thrown on the server side, we need to unwrap it + try { + serverSocket = acceptFuture.get(); + } catch (ExecutionException e) { + throw e.getCause(); + } + } finally { + forceClose(serverSocket); + forceClose(clientSocket); + forceClose(listeningSocket); + workerPool.shutdown(); + // Make sure the first handshake completed and only the second + // one failed. + Assert.assertEquals(1, handshakesCompleted.get()); + } + } + // Warning: this will reset the x509Util private void setCustomCipherSuites() { System.setProperty(x509Util.getCipherSuitesProperty(), customCipherSuites[0] + "," + customCipherSuites[1]); From d149b134049dc8527ba81a9546836942fcabc5b6 Mon Sep 17 00:00:00 2001 From: maoling Date: Mon, 14 Jan 2019 11:43:00 -0700 Subject: [PATCH 11/14] ZOOKEEPER-2284: LogFormatter and SnapshotFormatter does not handle FileNotFoundException gracefully - More details in [ZOOKEEPER-2284](https://issues.apache.org/jira/browse/ZOOKEEPER-2284) .Thanks [arshad.mohammad](https://issues.apache.org/jira/secure/ViewProfile.jspa?name=arshad.mohammad) for original work! Author: maoling Reviewers: fangmin@apache.org, andor@apache.org Closes #630 from maoling/ZOOKEEPER-2284 and squashes the following commits: 393cccdc5 [maoling] git rebase the code again 92c855f30 [maoling] rebase the code 5042a65bc [maoling] ZOOKEEPER-2284:LogFormatter and SnapshotFormatter does not handle FileNotFoundException gracefully --- .../java/org/apache/zookeeper/ZKUtil.java | 19 ++++ .../apache/zookeeper/server/LogFormatter.java | 8 ++ .../zookeeper/server/SnapshotFormatter.java | 9 +- .../java/org/apache/zookeeper/ZKUtilTest.java | 88 +++++++++++++++++++ 4 files changed, 123 insertions(+), 1 deletion(-) create mode 100644 zookeeper-server/src/test/java/org/apache/zookeeper/ZKUtilTest.java diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/ZKUtil.java b/zookeeper-server/src/main/java/org/apache/zookeeper/ZKUtil.java index a6abf2f42e1..f9cfe4bfc28 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/ZKUtil.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/ZKUtil.java @@ -17,6 +17,7 @@ */ package org.apache.zookeeper; +import java.io.File; import java.util.ArrayList; import java.util.Collections; import java.util.Deque; @@ -88,6 +89,24 @@ public static void deleteRecursive(ZooKeeper zk, final String pathRoot, VoidCall zk.delete(tree.get(i), -1, cb, ctx); //Delete all versions of the node with -1. } } + + /** + * @param filePath the file path to be validated + * @return Returns null if valid otherwise error message + */ + public static String validateFileInput(String filePath) { + File file = new File(filePath); + if (!file.exists()) { + return "File '" + file.getAbsolutePath() + "' does not exist."; + } + if (!file.canRead()) { + return "Read permission is denied on the file '" + file.getAbsolutePath() + "'"; + } + if (file.isDirectory()) { + return "'" + file.getAbsolutePath() + "' is a direcory. it must be a file."; + } + return null; + } /** * BFS Traversal of the system under pathRoot, with the entries in the list, in the diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/LogFormatter.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/LogFormatter.java index 0e72a6739b4..92ac5037555 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/LogFormatter.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/LogFormatter.java @@ -31,6 +31,7 @@ import org.apache.yetus.audience.InterfaceAudience; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.zookeeper.ZKUtil; import org.apache.zookeeper.server.persistence.FileHeader; import org.apache.zookeeper.server.persistence.FileTxnLog; import org.apache.zookeeper.server.util.SerializeUtils; @@ -48,6 +49,13 @@ public static void main(String[] args) throws Exception { System.err.println("USAGE: LogFormatter log_file"); System.exit(ExitCode.INVALID_INVOCATION.getValue()); } + + String error = ZKUtil.validateFileInput(args[0]); + if (null != error) { + System.err.println(error); + System.exit(ExitCode.INVALID_INVOCATION.getValue()); + } + FileInputStream fis = new FileInputStream(args[0]); BinaryInputArchive logStream = BinaryInputArchive.getArchive(fis); FileHeader fhdr = new FileHeader(); diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/SnapshotFormatter.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/SnapshotFormatter.java index 2a80d897bdc..50230eda9d9 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/SnapshotFormatter.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/SnapshotFormatter.java @@ -34,6 +34,7 @@ import org.apache.jute.BinaryInputArchive; import org.apache.jute.InputArchive; import org.apache.yetus.audience.InterfaceAudience; +import org.apache.zookeeper.ZKUtil; import org.apache.zookeeper.data.StatPersisted; import org.apache.zookeeper.server.persistence.FileSnap; import org.apache.zookeeper.server.persistence.Util; @@ -78,7 +79,13 @@ public static void main(String[] args) throws Exception { System.err.println(" -json dump znode info in json format"); System.exit(ExitCode.INVALID_INVOCATION.getValue()); } - + + String error = ZKUtil.validateFileInput(snapshotFile); + if (null != error) { + System.err.println(error); + System.exit(ExitCode.INVALID_INVOCATION.getValue()); + } + if (dumpData && dumpJson) { System.err.println("Cannot specify both data dump (-d) and json mode (-json) in same call"); System.exit(ExitCode.INVALID_INVOCATION.getValue()); diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/ZKUtilTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/ZKUtilTest.java new file mode 100644 index 00000000000..cd8e59b00ce --- /dev/null +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/ZKUtilTest.java @@ -0,0 +1,88 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.zookeeper; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assume.assumeTrue; + +import java.io.File; +import java.io.IOException; +import java.util.UUID; + +import org.junit.BeforeClass; +import org.junit.Test; + +public class ZKUtilTest { + private static final File testData = new File(System.getProperty("test.data.dir", "build/test/data")); + + @BeforeClass + public static void init() { + testData.mkdirs(); + } + + @Test + public void testValidateFileInput() throws IOException { + File file = File.createTempFile("test", ".junit", testData); + file.deleteOnExit(); + String absolutePath = file.getAbsolutePath(); + String error = ZKUtil.validateFileInput(absolutePath); + assertNull(error); + } + + @Test + public void testValidateFileInputNotExist() { + String fileName = UUID.randomUUID().toString(); + File file = new File(testData, fileName); + String absolutePath = file.getAbsolutePath(); + String error = ZKUtil.validateFileInput(absolutePath); + assertNotNull(error); + String expectedMessage = "File '" + absolutePath + "' does not exist."; + assertEquals(expectedMessage, error); + } + + @Test + public void testValidateFileInputDirectory() throws Exception { + File file = File.createTempFile("test", ".junit", testData); + file.deleteOnExit(); + // delete file, as we need directory not file + file.delete(); + file.mkdir(); + String absolutePath = file.getAbsolutePath(); + String error = ZKUtil.validateFileInput(absolutePath); + assertNotNull(error); + String expectedMessage = "'" + absolutePath + "' is a direcory. it must be a file."; + assertEquals(expectedMessage, error); + } + + @Test + public void testUnreadableFileInput() throws Exception { + //skip this test on Windows, coverage on Linux + assumeTrue(!org.apache.zookeeper.Shell.WINDOWS); + File file = File.createTempFile("test", ".junit", testData); + file.setReadable(false, false); + file.deleteOnExit(); + String absolutePath = file.getAbsolutePath(); + String error = ZKUtil.validateFileInput(absolutePath); + assertNotNull(error); + String expectedMessage = "Read permission is denied on the file '" + absolutePath + "'"; + assertEquals(expectedMessage, error); + } + +} \ No newline at end of file From b66802ef2f24181a3967152782d1cc8bb7886f35 Mon Sep 17 00:00:00 2001 From: Norbert Kalmar Date: Mon, 14 Jan 2019 16:42:39 -0700 Subject: [PATCH 12/14] ZOOKEEPER-3122: MAVEN MIGRATION - - remove contrib build from default build profile Author: Norbert Kalmar Reviewers: andor@apache.org Closes #774 from nkalmar/ZK-3122 --- pom.xml | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/pom.xml b/pom.xml index 0dc71743c5f..22faf7ae8bc 100755 --- a/pom.xml +++ b/pom.xml @@ -62,7 +62,6 @@ zookeeper-server zookeeper-client zookeeper-recipes - zookeeper-contrib @@ -243,6 +242,9 @@ full-build + + zookeeper-contrib + java-build From 2eb8dd0baf91a1a7c09d76f22ff658009898ca0b Mon Sep 17 00:00:00 2001 From: Dinesh Appavoo Date: Wed, 16 Jan 2019 06:21:28 -0700 Subject: [PATCH 13/14] ZOOKEEPER-3209: New `getEphemerals` api to get all the ephemeral nodes created by the session See https://issues.apache.org/jira/browse/ZOOKEEPER-3209 for details about the API. New API `getEphemerals()` to get all the ephemeral nodes created by the session by providing the prefix path. * Get the prefix path as a input parameter and return a list of string (ephemeral nodes) * If the prefix path is `/` or empty return all the ephemeral nodes created by the session * Provide synchronous and asynchronous API's with same functionality Author: Dinesh Appavoo Reviewers: fangmin@apache.org, andor@apache.org Closes #735 from dineshappavoo/ZOOKEEPER-3209 --- ivy.xml | 4 +- .../markdown/zookeeperProgrammers.md | 14 +- .../src/main/resources/zookeeper.jute | 8 + .../org/apache/zookeeper/AsyncCallback.java | 14 ++ .../java/org/apache/zookeeper/ClientCnxn.java | 16 +- .../java/org/apache/zookeeper/ZooDefs.java | 2 + .../java/org/apache/zookeeper/ZooKeeper.java | 65 +++++ .../org/apache/zookeeper/ZooKeeperMain.java | 2 + .../zookeeper/cli/GetEphemeralsCommand.java | 73 ++++++ .../server/FinalRequestProcessor.java | 24 ++ .../server/PrepRequestProcessor.java | 1 + .../org/apache/zookeeper/server/Request.java | 4 + .../zookeeper/server/TraceFormatter.java | 2 + .../apache/zookeeper/GetEphemeralsTest.java | 235 ++++++++++++++++++ 14 files changed, 460 insertions(+), 4 deletions(-) create mode 100644 zookeeper-server/src/main/java/org/apache/zookeeper/cli/GetEphemeralsCommand.java create mode 100644 zookeeper-server/src/test/java/org/apache/zookeeper/GetEphemeralsTest.java diff --git a/ivy.xml b/ivy.xml index a95331780d8..b69df51e07a 100644 --- a/ivy.xml +++ b/ivy.xml @@ -91,8 +91,8 @@ rev="${apache-rat-tasks.version}" conf="releaseaudit->default"> - + diff --git a/zookeeper-jute/src/main/resources/zookeeper.jute b/zookeeper-jute/src/main/resources/zookeeper.jute index 2533ddf8d11..303d7cea5fc 100644 --- a/zookeeper-jute/src/main/resources/zookeeper.jute +++ b/zookeeper-jute/src/main/resources/zookeeper.jute @@ -222,6 +222,14 @@ module org.apache.zookeeper.proto { ustring path; int type; } + + class GetEphemeralsRequest { + ustring prefixPath; + } + + class GetEphemeralsResponse { + vector ephemerals; + } } module org.apache.zookeeper.server.quorum { diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/AsyncCallback.java b/zookeeper-server/src/main/java/org/apache/zookeeper/AsyncCallback.java index c5529d7ccbd..cca7be48bd0 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/AsyncCallback.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/AsyncCallback.java @@ -328,4 +328,18 @@ interface MultiCallback extends AsyncCallback { public void processResult(int rc, String path, Object ctx, List opResults); } + + /** + * This callback is used to process the getEphemerals results from + * a single getEphemerals call. + */ + interface EphemeralsCallback extends AsyncCallback { + /** + * + * @param rc The return code or the result of the call. + * @param ctx Whatever context object that we passed to asynchronous calls. + * @param paths The path that we passed to asynchronous calls. + */ + public void processResult(int rc, Object ctx, List paths); + } } diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/ClientCnxn.java b/zookeeper-server/src/main/java/org/apache/zookeeper/ClientCnxn.java index db2b4866af0..5c4171ae984 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/ClientCnxn.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/ClientCnxn.java @@ -53,6 +53,7 @@ import org.apache.zookeeper.AsyncCallback.ChildrenCallback; import org.apache.zookeeper.AsyncCallback.Create2Callback; import org.apache.zookeeper.AsyncCallback.DataCallback; +import org.apache.zookeeper.AsyncCallback.EphemeralsCallback; import org.apache.zookeeper.AsyncCallback.MultiCallback; import org.apache.zookeeper.AsyncCallback.StatCallback; import org.apache.zookeeper.AsyncCallback.StringCallback; @@ -78,6 +79,7 @@ import org.apache.zookeeper.proto.GetChildren2Response; import org.apache.zookeeper.proto.GetChildrenResponse; import org.apache.zookeeper.proto.GetDataResponse; +import org.apache.zookeeper.proto.GetEphemeralsResponse; import org.apache.zookeeper.proto.GetSASLRequest; import org.apache.zookeeper.proto.ReplyHeader; import org.apache.zookeeper.proto.RequestHeader; @@ -557,6 +559,9 @@ private void processEvent(Object event) { } else if (lcb.cb instanceof StringCallback) { ((StringCallback) lcb.cb).processResult(lcb.rc, lcb.path, lcb.ctx, null); + } else if (lcb.cb instanceof AsyncCallback.EphemeralsCallback) { + ((AsyncCallback.EphemeralsCallback) lcb.cb).processResult(lcb.rc, + lcb.ctx, null); } else { ((VoidCallback) lcb.cb).processResult(lcb.rc, lcb.path, lcb.ctx); @@ -670,7 +675,16 @@ private void processEvent(Object event) { } else { cb.processResult(rc, clientPath, p.ctx, null); } - } else if (p.cb instanceof VoidCallback) { + } else if (p.response instanceof GetEphemeralsResponse) { + EphemeralsCallback cb = (EphemeralsCallback) p.cb; + GetEphemeralsResponse rsp = (GetEphemeralsResponse) p.response; + if (rc == 0) { + cb.processResult(rc, p.ctx, rsp.getEphemerals()); + } else { + cb.processResult(rc, p.ctx, null); + } + } + else if (p.cb instanceof VoidCallback) { VoidCallback cb = (VoidCallback) p.cb; cb.processResult(rc, clientPath, p.ctx); } diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/ZooDefs.java b/zookeeper-server/src/main/java/org/apache/zookeeper/ZooDefs.java index 97aa28a7c7d..a3b959eb753 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/ZooDefs.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/ZooDefs.java @@ -81,6 +81,8 @@ public interface OpCode { public final int sasl = 102; + public final int getEphemerals = 103; + public final int createSession = -10; public final int closeSession = -11; diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeper.java b/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeper.java index 6cac98e5e71..14752207442 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeper.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeper.java @@ -57,6 +57,9 @@ import org.apache.zookeeper.proto.GetChildrenResponse; import org.apache.zookeeper.proto.GetDataRequest; import org.apache.zookeeper.proto.GetDataResponse; +import org.apache.zookeeper.proto.GetEphemeralsRequest; +import org.apache.zookeeper.proto.GetEphemeralsResponse; +import org.apache.zookeeper.proto.ReconfigRequest; import org.apache.zookeeper.proto.RemoveWatchesRequest; import org.apache.zookeeper.proto.ReplyHeader; import org.apache.zookeeper.proto.RequestHeader; @@ -2665,6 +2668,68 @@ public void getChildren(String path, boolean watch, Children2Callback cb, getChildren(path, watch ? watchManager.defaultWatcher : null, cb, ctx); } + /** + * Synchronously gets all the ephemeral nodes created by this session. + * + * @since 3.6.0 + * + */ + public List getEphemerals() + throws KeeperException, InterruptedException { + return getEphemerals("/"); + } + + /** + * Synchronously gets all the ephemeral nodes matching prefixPath + * created by this session. If prefixPath is "/" then it returns all + * ephemerals + * + * @since 3.6.0 + * + */ + public List getEphemerals(String prefixPath) + throws KeeperException, InterruptedException { + PathUtils.validatePath(prefixPath); + RequestHeader h = new RequestHeader(); + h.setType(ZooDefs.OpCode.getEphemerals); + GetEphemeralsRequest request = new GetEphemeralsRequest(prefixPath); + GetEphemeralsResponse response = new GetEphemeralsResponse(); + ReplyHeader r = cnxn.submitRequest(h, request, response, null); + if (r.getErr() != 0) { + throw KeeperException.create(KeeperException.Code.get(r.getErr())); + } + return response.getEphemerals(); + } + + /** + * Asynchronously gets all the ephemeral nodes matching prefixPath + * created by this session. If prefixPath is "/" then it returns all + * ephemerals + * + * @since 3.6.0 + * + */ + public void getEphemerals(String prefixPath, AsyncCallback.EphemeralsCallback cb, Object ctx) { + PathUtils.validatePath(prefixPath); + RequestHeader h = new RequestHeader(); + h.setType(ZooDefs.OpCode.getEphemerals); + GetEphemeralsRequest request = new GetEphemeralsRequest(prefixPath); + GetEphemeralsResponse response = new GetEphemeralsResponse(); + cnxn.queuePacket(h, new ReplyHeader(), request, response, cb, + null, null, ctx, null); + } + + /** + * Asynchronously gets all the ephemeral nodes created by this session. + * ephemerals + * + * @since 3.6.0 + * + */ + public void getEphemerals(AsyncCallback.EphemeralsCallback cb, Object ctx) { + getEphemerals("/", cb, ctx); + } + /** * Asynchronous sync. Flushes channel between process and leader. * @param path diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeperMain.java b/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeperMain.java index 300143acdf8..7dbdf2f830b 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeperMain.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/ZooKeeperMain.java @@ -37,6 +37,7 @@ import org.apache.yetus.audience.InterfaceAudience; import org.apache.zookeeper.cli.CliException; import org.apache.zookeeper.cli.CommandNotFoundException; +import org.apache.zookeeper.cli.GetEphemeralsCommand; import org.apache.zookeeper.cli.MalformedCommandException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -122,6 +123,7 @@ public boolean getPrintWatches( ) { new ReconfigCommand().addToMap(commandMapCli); new GetConfigCommand().addToMap(commandMapCli); new RemoveWatchesCommand().addToMap(commandMapCli); + new GetEphemeralsCommand().addToMap(commandMapCli); // add all to commandMap for (Entry entry : commandMapCli.entrySet()) { diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/cli/GetEphemeralsCommand.java b/zookeeper-server/src/main/java/org/apache/zookeeper/cli/GetEphemeralsCommand.java new file mode 100644 index 00000000000..e045112b30c --- /dev/null +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/cli/GetEphemeralsCommand.java @@ -0,0 +1,73 @@ +/** + * 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.cli; + +import java.util.List; + +import org.apache.commons.cli.CommandLine; +import org.apache.commons.cli.Options; +import org.apache.commons.cli.ParseException; +import org.apache.commons.cli.Parser; +import org.apache.commons.cli.PosixParser; +import org.apache.zookeeper.KeeperException; + +/** + * getEphemerals command for CLI + */ +public class GetEphemeralsCommand extends CliCommand { + private static Options options = new Options(); + private String[] args; + + public GetEphemeralsCommand() { + super("getEphemerals", "path"); + } + + @Override + public CliCommand parse(String[] cmdArgs) throws CliParseException { + Parser parser = new PosixParser(); + CommandLine cl; + try { + cl = parser.parse(options, cmdArgs); + } catch (ParseException ex) { + throw new CliParseException(ex); + } + args = cl.getArgs(); + + return this; + } + + @Override + public boolean exec() throws CliException { + String path; + List ephemerals; + try { + if (args.length < 2) { + // gets all the ephemeral nodes for the session + ephemerals = zk.getEphemerals(); + } else { + path = args[1]; + ephemerals = zk.getEphemerals(path); + } + } catch (IllegalArgumentException ex) { + throw new MalformedPathException(ex.getMessage()); + } catch (KeeperException|InterruptedException ex) { + throw new CliWrapperException(ex); + } + out.println(ephemerals); + return false; + } +} diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java index d022193e739..2c8c5b2fc24 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java @@ -18,6 +18,7 @@ package org.apache.zookeeper.server; +import org.apache.commons.lang.StringUtils; import org.apache.jute.Record; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.KeeperException.Code; @@ -48,6 +49,8 @@ import org.apache.zookeeper.proto.GetChildrenResponse; import org.apache.zookeeper.proto.GetDataRequest; import org.apache.zookeeper.proto.GetDataResponse; +import org.apache.zookeeper.proto.GetEphemeralsRequest; +import org.apache.zookeeper.proto.GetEphemeralsResponse; import org.apache.zookeeper.proto.RemoveWatchesRequest; import org.apache.zookeeper.proto.ReplyHeader; import org.apache.zookeeper.proto.SetACLResponse; @@ -65,8 +68,10 @@ import java.io.IOException; import java.nio.ByteBuffer; +import java.util.ArrayList; import java.util.List; import java.util.Locale; +import java.util.Set; /** * This Request processor actually applies any transaction associated with a @@ -441,6 +446,25 @@ public void processRequest(Request request) { } break; } + case OpCode.getEphemerals: { + lastOp = "GETE"; + GetEphemeralsRequest getEphemerals = new GetEphemeralsRequest(); + ByteBufferInputStream.byteBuffer2Record(request.request, getEphemerals); + String prefixPath = getEphemerals.getPrefixPath(); + Set allEphems = zks.getZKDatabase().getDataTree().getEphemerals(request.sessionId); + List ephemerals = new ArrayList<>(); + if (StringUtils.isBlank(prefixPath) || "/".equals(prefixPath.trim())) { + ephemerals.addAll(allEphems); + } else { + for (String path: allEphems) { + if(path.startsWith(prefixPath)) { + ephemerals.add(path); + } + } + } + rsp = new GetEphemeralsResponse(ephemerals); + break; + } } } catch (SessionMovedException e) { // session moved is a connection level error, we need to tear diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/PrepRequestProcessor.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/PrepRequestProcessor.java index 57eaf1e31bc..be87530aa1d 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/PrepRequestProcessor.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/PrepRequestProcessor.java @@ -859,6 +859,7 @@ protected void pRequest(Request request) throws RequestProcessorException { case OpCode.setWatches: case OpCode.checkWatches: case OpCode.removeWatches: + case OpCode.getEphemerals: zks.sessionTracker.checkSession(request.sessionId, request.getOwner()); break; diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/Request.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/Request.java index ede9280441a..20877c927a7 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/Request.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/Request.java @@ -147,6 +147,7 @@ static boolean isValid(int type) { case OpCode.getChildren: case OpCode.getChildren2: case OpCode.getData: + case OpCode.getEphemerals: case OpCode.multi: case OpCode.ping: case OpCode.reconfig: @@ -169,6 +170,7 @@ public boolean isQuorum() { case OpCode.getChildren: case OpCode.getChildren2: case OpCode.getData: + case OpCode.getEphemerals: return false; case OpCode.create: case OpCode.create2: @@ -229,6 +231,8 @@ static String op2String(int op) { return "getChildren"; case OpCode.getChildren2: return "getChildren2"; + case OpCode.getEphemerals: + return "getEphemerals"; case OpCode.ping: return "ping"; case OpCode.createSession: diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/TraceFormatter.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/TraceFormatter.java index f4c138363a6..dff6dfdfd48 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/TraceFormatter.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/TraceFormatter.java @@ -61,6 +61,8 @@ public static String op2String(int op) { return "getChildren"; case OpCode.getChildren2: return "getChildren2"; + case OpCode.getEphemerals: + return "getEphemerals"; case OpCode.ping: return "ping"; case OpCode.createSession: diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/GetEphemeralsTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/GetEphemeralsTest.java new file mode 100644 index 00000000000..355223fc60e --- /dev/null +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/GetEphemeralsTest.java @@ -0,0 +1,235 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.zookeeper; + +import org.apache.zookeeper.AsyncCallback; +import org.apache.zookeeper.CreateMode; +import org.apache.zookeeper.KeeperException; +import org.apache.zookeeper.ZooDefs.Ids; +import org.apache.zookeeper.ZooKeeper; +import org.apache.zookeeper.test.ClientBase; +import org.junit.Assert; +import org.junit.Test; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +public class GetEphemeralsTest extends ClientBase { + private static final String BASE = "/base"; + private static final int PERSISTENT_CNT = 2; + private static final int EPHEMERAL_CNT = 2; + private static final String NEWLINE = System.getProperty("line.separator"); + private String[] expected; + private ZooKeeper zk; + + @Override + public void setUp() throws Exception { + super.setUp(); + + zk = createClient(); + expected = generatePaths(PERSISTENT_CNT, EPHEMERAL_CNT); + } + + @Override + public void tearDown() throws Exception { + super.tearDown(); + + zk.close(); + } + + @Test + public void testGetEphemeralsSync() throws KeeperException, InterruptedException { + List actual = zk.getEphemerals(); + Assert.assertEquals( "Expected ephemeral count for allPaths", + actual.size() , expected.length); + for (int i = 0; i < expected.length; i++) { + String path = expected[i]; + Assert.assertTrue( + String.format("Path=%s exists in get All Ephemerals list ", path), + actual.contains(path)); + } + } + + @Test + public void testGetEphemeralsSyncByPath() throws KeeperException, InterruptedException { + final String prefixPath = BASE + 0; + List actual = zk.getEphemerals(prefixPath); + Assert.assertEquals( "Expected ephemeral count for allPaths", + actual.size() , EPHEMERAL_CNT); + for (int i = 0; i < EPHEMERAL_CNT; i++) { + String path = expected[i]; + Assert.assertTrue(String.format("Path=%s exists in getEphemerals(%s) list ", + path, prefixPath), actual.contains(path)); + } + } + + @Test + public void testGetEphemerals() + throws IOException, KeeperException, InterruptedException { + + final CountDownLatch doneProcessing = new CountDownLatch(1); + final List unexpectedBehavior = new ArrayList(); + zk.getEphemerals(new AsyncCallback.EphemeralsCallback() { + @Override + public void processResult(int rc, Object ctx, List paths) { + if (paths == null ) { + unexpectedBehavior.add(String.format("Expected ephemeral count for" + + " allPaths to be %d but was null", expected.length)); + } else if (paths.size() != expected.length) { + unexpectedBehavior.add( + String.format("Expected ephemeral count for allPaths to be %d but was %d", + expected.length, paths.size())); + } + for (int i = 0; i < expected.length; i++) { + String path = expected[i]; + if (!paths.contains(path)) { + unexpectedBehavior.add( + String.format("Path=%s exists in getEphemerals list ", path)); + } + } + doneProcessing.countDown(); + } + }, null); + long waitForCallbackSecs = 2l; + if (!doneProcessing.await(waitForCallbackSecs, TimeUnit.SECONDS)) { + Assert.fail(String.format("getEphemerals didn't callback within %d seconds", + waitForCallbackSecs)); + } + checkForUnexpectedBehavior(unexpectedBehavior); + + } + + @Test + public void testGetEphemeralsByPath() + throws IOException, KeeperException, InterruptedException { + + final CountDownLatch doneProcessing = new CountDownLatch(1); + final String checkPath = BASE + "0"; + final List unexpectedBehavior = new ArrayList(); + zk.getEphemerals(checkPath, new AsyncCallback.EphemeralsCallback() { + @Override + public void processResult(int rc, Object ctx, List paths) { + if (paths == null ) { + unexpectedBehavior.add( + String.format("Expected ephemeral count for %s to be %d but was null", + checkPath, expected.length)); + } else if (paths.size() != EPHEMERAL_CNT) { + unexpectedBehavior.add( + String.format("Expected ephemeral count for %s to be %d but was %d", + checkPath, EPHEMERAL_CNT, paths.size())); + } + for (int i = 0; i < EPHEMERAL_CNT; i++) { + String path = expected[i]; + if(! paths.contains(path)) { + unexpectedBehavior.add(String.format("Expected path=%s didn't exist " + + "in getEphemerals list.", path)); + } + } + doneProcessing.countDown(); + } + }, null); + long waitForCallbackSecs = 2l; + if (!doneProcessing.await(waitForCallbackSecs, TimeUnit.SECONDS)) { + Assert.fail(String.format("getEphemerals(%s) didn't callback within %d seconds", + checkPath, waitForCallbackSecs)); + } + checkForUnexpectedBehavior(unexpectedBehavior); + } + + @Test + public void testGetEphemeralsEmpty() + throws IOException, KeeperException, InterruptedException { + + final CountDownLatch doneProcessing = new CountDownLatch(1); + final String checkPath = "/unknownPath"; + final int expectedSize = 0; + final List unexpectedBehavior = new ArrayList(); + zk.getEphemerals(checkPath, new AsyncCallback.EphemeralsCallback() { + @Override + public void processResult(int rc, Object ctx, List paths) { + if (paths == null ) { + unexpectedBehavior.add( + String.format("Expected ephemeral count for %s to be %d but was null", + checkPath, expectedSize)); + } else if (paths.size() != expectedSize) { + unexpectedBehavior.add( + String.format("Expected ephemeral count for %s to be %d but was %d", + checkPath, expectedSize, paths.size())); + } + doneProcessing.countDown(); + } + }, null); + long waitForCallbackSecs = 2l; + if (!doneProcessing.await(waitForCallbackSecs, TimeUnit.SECONDS)) { + Assert.fail(String.format("getEphemerals(%s) didn't callback within %d seconds", + checkPath, waitForCallbackSecs)); + } + checkForUnexpectedBehavior(unexpectedBehavior); + } + + @Test + public void testGetEphemeralsErrors() throws KeeperException { + try { + zk.getEphemerals(null, null, null); + Assert.fail("Should have thrown a IllegalArgumentException for a null prefixPath"); + } catch (IllegalArgumentException e) { + //pass + } + + try { + zk.getEphemerals("no leading slash", null, null); + Assert.fail("Should have thrown a IllegalArgumentException " + + "for a prefix with no leading slash"); + } catch (IllegalArgumentException e) { + //pass + } + } + + private String[] generatePaths(int persistantCnt, int ephemeralCnt) + throws KeeperException, InterruptedException { + + final String[] expected = new String[persistantCnt * ephemeralCnt]; + for (int p = 0; p < persistantCnt; p++) { + String base = BASE + p; + zk.create(base, base.getBytes(), Ids.OPEN_ACL_UNSAFE, + CreateMode.PERSISTENT); + for (int e = 0; e < ephemeralCnt; e++) { + String ephem = base + "/ephem" + e; + zk.create(ephem, ephem.getBytes(), Ids.OPEN_ACL_UNSAFE, + CreateMode.EPHEMERAL); + expected[p * ephemeralCnt + e] = ephem; + } + } + return expected; + } + + private void checkForUnexpectedBehavior(List unexpectedBehavior) { + if (unexpectedBehavior.size() > 0) { + StringBuilder b = new StringBuilder("The test failed for the following reasons:"); + b.append(NEWLINE); + for (String error : unexpectedBehavior) { + b.append("ERROR: ").append(error).append(NEWLINE); + } + Assert.fail(b.toString()); + } + } +} From ed4fad3c1af9b15f2c54ccd91f7c09a913c23c40 Mon Sep 17 00:00:00 2001 From: Dinesh Appavoo Date: Wed, 16 Jan 2019 15:23:04 -0700 Subject: [PATCH 14/14] ZOOKEEPER-3209: Fix compilation error Author: Dinesh Appavoo Reviewers: andor@apache.org Closes #778 from anmolnar/ZOOKEEPER-3209_buildfix --- .../org/apache/zookeeper/server/FinalRequestProcessor.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java index 2c8c5b2fc24..38ebd3fe8a2 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java @@ -456,9 +456,9 @@ public void processRequest(Request request) { if (StringUtils.isBlank(prefixPath) || "/".equals(prefixPath.trim())) { ephemerals.addAll(allEphems); } else { - for (String path: allEphems) { - if(path.startsWith(prefixPath)) { - ephemerals.add(path); + for (String p: allEphems) { + if(p.startsWith(prefixPath)) { + ephemerals.add(p); } } }