Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions zookeeper-docs/src/main/resources/markdown/zookeeperAdmin.md
Original file line number Diff line number Diff line change
Expand Up @@ -1149,6 +1149,11 @@ property, when available, is noted below.
**New in 3.6.2:**
When enabled, a learner will close the quorum socket asynchronously. This is useful for TLS connections where closing a socket might take a long time, block the shutdown process, potentially delay a new leader election, and leave the quorum unavailabe. Closing the socket asynchronously avoids blocking the shutdown process despite the long socket closing time and a new leader election can be started while the socket being closed. The default is false.

* *learner.asyncSending*
(Jave system property only: **learner.asyncSending**)
**New in 3.7.0:**
The sending and receiving packets in Learner were done synchronously in a critical section. An untimely network issue could cause the followers to hang (see [ZOOKEEPER-3575](https://issues.apache.org/jira/browse/ZOOKEEPER-3575) and [ZOOKEEPER-4074](https://issues.apache.org/jira/browse/ZOOKEEPER-4074)). The new design moves sending packets in Learner to a separate thread and sends the packets asynchronously. The new design is enabled with this parameter (learner.asyncSending). The default setting is true.

* *leader.closeSocketAsync*
(Java system property only: **leader.closeSocketAsync**)
**New in 3.6.2:**
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -119,7 +119,7 @@ public Socket getSocket() {
private static final boolean nodelay = System.getProperty("follower.nodelay", "true").equals("true");

public static final String LEARNER_ASYNC_SENDING = "learner.asyncSending";
private static boolean asyncSending = Boolean.getBoolean(LEARNER_ASYNC_SENDING);
private static boolean asyncSending = System.getProperty(LEARNER_ASYNC_SENDING, "true").equals("true");
public static final String LEARNER_CLOSE_SOCKET_ASYNC = "learner.closeSocketAsync";
public static final boolean closeSocketAsync = Boolean.getBoolean(LEARNER_CLOSE_SOCKET_ASYNC);

Expand Down Expand Up @@ -208,7 +208,7 @@ void writePacketNow(QuorumPacket pp, boolean flush) throws IOException {
/**
* Start thread that will forward any packet in the queue to the leader
*/
protected void startSendingThread() {
public void startSendingThread() {
sender = new LearnerSender(this);
sender.start();
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -94,27 +94,27 @@ static class TestLearner extends Learner {
private Socket socketToBeCreated = null;
private Set<InetSocketAddress> unreachableAddresses = emptySet();

private void setTimeMultiplier(long multiplier) {
protected void setTimeMultiplier(long multiplier) {
timeMultiplier = multiplier;
}

private void setPassConnectAttempt(int num) {
protected void setPassConnectAttempt(int num) {
passSocketConnectOnAttempt = num;
}

protected long nanoTime() {
return socketConnectAttempt * timeMultiplier;
}

private int getSockConnectAttempt() {
protected int getSockConnectAttempt() {
return socketConnectAttempt;
}

private void setSocketToBeCreated(Socket socketToBeCreated) {
protected void setSocketToBeCreated(Socket socketToBeCreated) {
this.socketToBeCreated = socketToBeCreated;
}

private void setUnreachableAddresses(Set<InetSocketAddress> unreachableAddresses) {
protected void setUnreachableAddresses(Set<InetSocketAddress> unreachableAddresses) {
this.unreachableAddresses = unreachableAddresses;
}

Expand All @@ -136,6 +136,18 @@ protected Socket createSocket() throws X509Exception, IOException {
}
}

static class TestLearnerWithZk extends TestLearner {
TestLearnerWithZk() throws IOException {
super();
File tmpFile = File.createTempFile("test", ".dir", testData);
tmpFile.delete();
FileTxnSnapLog ftsl = new FileTxnSnapLog(tmpFile, tmpFile);
self = new QuorumPeer();
zk = new SimpleLearnerZooKeeperServer(ftsl, self);
((SimpleLearnerZooKeeperServer) zk).learner = this;
}
}

@AfterEach
public void cleanup() {
System.clearProperty(QuorumPeer.CONFIG_KEY_MULTI_ADDRESS_ENABLED);
Expand Down Expand Up @@ -216,7 +228,7 @@ public void shouldTryMultipleAddresses() throws Exception {
@Test
public void multipleAddressesSomeAreFailing() throws Exception {
System.setProperty(QuorumPeer.CONFIG_KEY_MULTI_ADDRESS_ENABLED, "true");
TestLearner learner = new TestLearner();
TestLearner learner = new TestLearnerWithZk();
learner.self = new QuorumPeer();
learner.self.setTickTime(2000);
learner.self.setInitLimit(5);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -112,6 +112,7 @@ public Boolean answer(InvocationOnMock invocation) throws Throwable {
fzks.startup();
fzks.setServerCnxnFactory(serverCnxnFactory);
quorumPeer.follower = new MyFollower(quorumPeer, fzks);
quorumPeer.follower.startSendingThread();
LOG.info("Follower created");
// Simulate a socket channel between a client and a follower
final SocketChannel socketChannel = createClientSocketChannel();
Expand Down