queuedPackets = new LinkedBlockingQueue<>();
+ private final QuorumPacket proposalOfDeath = new QuorumPacket();
+
+ Learner learner;
+
+ public LearnerSender(Learner learner) {
+ super("LearnerSender:" + learner.zk.getServerId(), learner.zk.getZooKeeperServerListener());
+ this.learner = learner;
+ }
+
+ @Override
+ public void run() {
+ while (true) {
+ try {
+ QuorumPacket p = queuedPackets.poll();
+ if (p == null) {
+ learner.bufferedOutput.flush();
+ p = queuedPackets.take();
+ }
+
+ if (p == proposalOfDeath) {
+ // Packet of death!
+ break;
+ }
+
+ learner.messageTracker.trackSent(p.getType());
+ learner.leaderOs.writeRecord(p, "packet");
+ } catch (IOException e) {
+ handleException(this.getName(), e);
+ break;
+ } catch (InterruptedException e) {
+ handleException(this.getName(), e);
+ break;
+ }
+ }
+
+ LOG.info("LearnerSender exited");
+ }
+
+ public void queuePacket(QuorumPacket pp) throws IOException {
+ if (pp == null) {
+ learner.bufferedOutput.flush();
+ } else {
+ queuedPackets.add(pp);
+ }
+ }
+
+ public void shutdown() {
+ LOG.info("Shutting down LearnerSender");
+ queuedPackets.clear();
+ queuedPackets.add(proposalOfDeath);
+ }
+}
diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/util/ConfigUtils.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/util/ConfigUtils.java
index 508dc112565..d6f75729798 100644
--- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/util/ConfigUtils.java
+++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/util/ConfigUtils.java
@@ -95,4 +95,29 @@ public static String[] getHostAndPort(String s) throws ConfigException {
}
}
+ /**
+ * Some old configuration properties are not configurable in zookeeper configuration file
+ * zoo.cfg. To make these properties configurable in zoo.cfg old properties are prepended
+ * with zookeeper. For example prop.x.y.z changed to zookeeper.prop.x.y.z. But for backward
+ * compatibility both prop.x.y.z and zookeeper.prop.x.y.z should be supported.
+ * This method first gets value from new property, if first property is not configured
+ * then gets value from old property
+ *
+ * @param newPropertyKey new property key which starts with zookeeper.
+ * @return either new or old system property value. Null if none of the properties are set.
+ */
+ public static String getPropertyBackwardCompatibleWay(String newPropertyKey) {
+ String newKeyValue = System.getProperty(newPropertyKey);
+ if (newKeyValue != null) {
+ return newKeyValue.trim();
+ }
+ String oldPropertyKey = newPropertyKey.replace("zookeeper.", "");
+ String oldKeyValue = System.getProperty(oldPropertyKey);
+
+ if (oldKeyValue != null) {
+ return oldKeyValue.trim();
+ }
+ return null;
+ }
+
}
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/LearnerMetricsTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/LearnerMetricsTest.java
index c6de02f80ed..659ba31f8a2 100644
--- a/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/LearnerMetricsTest.java
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/LearnerMetricsTest.java
@@ -20,6 +20,8 @@
import static org.hamcrest.core.Is.is;
import static org.hamcrest.number.OrderingComparison.greaterThanOrEqualTo;
+import java.util.Arrays;
+import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import org.apache.zookeeper.CreateMode;
@@ -31,14 +33,46 @@
import org.apache.zookeeper.test.ClientBase;
import org.hamcrest.Matcher;
import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+@RunWith(Parameterized.class)
public class LearnerMetricsTest extends QuorumPeerTestBase {
private static final int TIMEOUT_SECONDS = 30;
private static final int SERVER_COUNT = 4; // 1 observer, 3 participants
private final QuorumPeerTestBase.MainThread[] mt = new QuorumPeerTestBase.MainThread[SERVER_COUNT];
private ZooKeeper zk_client;
+ private boolean asyncSending;
+ private static boolean bakAsyncSending;
+
+ public LearnerMetricsTest(boolean asyncSending) {
+ this.asyncSending = asyncSending;
+ }
+
+ @Parameterized.Parameters
+ public static Collection sendingModes() {
+ return Arrays.asList(new Object[][]{{true}, {false}});
+ }
+
+ @Before
+ public void setAsyncSendingFlag() {
+ Learner.setAsyncSending(asyncSending);
+ }
+
+ @BeforeClass
+ public static void saveAsyncSendingFlag() {
+ bakAsyncSending = Learner.getAsyncSending();
+ }
+
+ @AfterClass
+ public static void resetAsyncSendingFlag() {
+ Learner.setAsyncSending(bakAsyncSending);
+ }
@Test
public void testLearnerMetricsTest() throws Exception {
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/ReconfigDuringLeaderSyncTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/ReconfigDuringLeaderSyncTest.java
index bff6cbf0e45..ee51baf955e 100644
--- a/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/ReconfigDuringLeaderSyncTest.java
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/ReconfigDuringLeaderSyncTest.java
@@ -25,6 +25,8 @@
import java.io.File;
import java.io.IOException;
import java.net.InetSocketAddress;
+import java.util.Arrays;
+import java.util.Collection;
import java.util.Map;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.PortAssignment;
@@ -38,23 +40,51 @@
import org.apache.zookeeper.test.ClientBase;
import org.apache.zookeeper.test.ClientBase.CountdownWatcher;
import org.junit.After;
+import org.junit.AfterClass;
import org.junit.Before;
+import org.junit.BeforeClass;
import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+@RunWith(Parameterized.class)
public class ReconfigDuringLeaderSyncTest extends QuorumPeerTestBase {
- protected static final Logger LOG = LoggerFactory.getLogger(ReconfigDuringLeaderSyncTest.class);
+ private static final Logger LOG = LoggerFactory.getLogger(ReconfigDuringLeaderSyncTest.class);
private static int SERVER_COUNT = 3;
private MainThread[] mt;
+ private static boolean bakAsyncSending;
+
+ private boolean asyncSending;
+
+ public ReconfigDuringLeaderSyncTest(boolean asyncSending) {
+ this.asyncSending = asyncSending;
+ }
+
+ @Parameterized.Parameters
+ public static Collection sendingModes() {
+ return Arrays.asList(new Object[][]{{true}, {false}});
+ }
@Before
public void setup() {
System.setProperty("zookeeper.DigestAuthenticationProvider.superDigest", "super:D/InIHSb7yEEbrWz8b9l71RjZJU="/* password is 'test'*/);
+ Learner.setAsyncSending(asyncSending);
QuorumPeerConfig.setReconfigEnabled(true);
}
+ @BeforeClass
+ public static void saveAsyncSendingFlag() {
+ bakAsyncSending = Learner.getAsyncSending();
+ }
+
+ @AfterClass
+ public static void resetAsyncSendingFlag() {
+ Learner.setAsyncSending(bakAsyncSending);
+ }
+
/**
*
* Test case for https://issues.apache.org/jira/browse/ZOOKEEPER-2172.
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/util/ConfigUtilsTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/util/ConfigUtilsTest.java
index ba68b229ec0..d259a5d7cbf 100644
--- a/zookeeper-server/src/test/java/org/apache/zookeeper/server/util/ConfigUtilsTest.java
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/server/util/ConfigUtilsTest.java
@@ -19,6 +19,7 @@
package org.apache.zookeeper.server.util;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
import org.apache.zookeeper.server.quorum.QuorumPeerConfig.ConfigException;
import org.junit.Test;
@@ -69,4 +70,45 @@ public void testGetHostAndPortWithoutPort() throws ConfigException {
assertEquals(nsa.length, 1);
}
+ @Test
+ public void testGetPropertyBackwardCompatibleWay() throws ConfigException {
+ String newProp = "zookeeper.prop.x.y.z";
+ String oldProp = "prop.x.y.z";
+
+ // Null as both properties are not set
+ String result = ConfigUtils.getPropertyBackwardCompatibleWay(newProp);
+ assertNull(result);
+
+ // Return old property value when only old property is set
+ String oldPropValue = "oldPropertyValue";
+ System.setProperty(oldProp, oldPropValue);
+ result = ConfigUtils.getPropertyBackwardCompatibleWay(newProp);
+ assertEquals(oldPropValue, result);
+
+ // Return new property value when both properties are set
+ String newPropValue = "newPropertyValue";
+ System.setProperty(newProp, newPropValue);
+ result = ConfigUtils.getPropertyBackwardCompatibleWay(newProp);
+ assertEquals(newPropValue, result);
+
+ // cleanUp
+ clearProp(newProp, oldProp);
+
+ // Return trimmed value
+ System.setProperty(oldProp, oldPropValue + " ");
+ result = ConfigUtils.getPropertyBackwardCompatibleWay(newProp);
+ assertEquals(oldPropValue, result);
+
+ System.setProperty(newProp, " " + newPropValue);
+ result = ConfigUtils.getPropertyBackwardCompatibleWay(newProp);
+ assertEquals(newPropValue, result);
+
+ // cleanUp
+ clearProp(newProp, oldProp);
+ }
+
+ private void clearProp(String newProp, String oldProp) {
+ System.clearProperty(newProp);
+ System.clearProperty(oldProp);
+ }
}