Skip to content

ZOOKEEPER-3774: Close quorum socket asynchronously on the leader to a… #1301

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
jhuan31 wants to merge 1 commit into from
Closed

ZOOKEEPER-3774: Close quorum socket asynchronously on the leader to a… #1301

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
9 changes: 8 additions & 1 deletion zookeeper-docs/src/main/resources/markdown/zookeeperAdmin.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1146,8 +1146,14 @@ property, when available, is noted below.

* *learner.closeSocketAsync*
(Jave system property only: **learner.closeSocketAsync**)
**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.

* *leader.closeSocketAsync*
Comment thread
jhuan31 marked this conversation as resolved.
(Java system property only: **leader.closeSocketAsync**)
Comment thread
jhuan31 marked this conversation as resolved.
**New in 3.6.2:**
When enabled, the leader will close a quorum socket asynchoronously. This is useful for TLS connections where closing a socket might take a long time. If disconnecting a follower is initiated in ping() because of a failed SyncLimitCheck then the long socket closing time will block the sending of pings to other followers. Without receiving pings, the other followers will not send session information to the leader, which causes sessions to expire. Setting this flag to true ensures that pings will be sent regularly. The default is false.

* *forward_learner_requests_to_commit_processor_disabled*
(Jave system property: **zookeeper.forward_learner_requests_to_commit_processor_disabled**)
When this property is set, the requests from learners won't be enqueued to
Expand Down Expand Up @@ -1512,7 +1518,8 @@ and [SASL authentication for ZooKeeper](https://cwiki.apache.org/confluence/disp
* *sslQuorum* :
(Java system property: **zookeeper.sslQuorum**)
**New in 3.5.5:**
Enables encrypted quorum communication. Default is `false`.
Enables encrypted quorum communication. Default is `false`. When enabling this feature, please also consider enabling *leader.closeSocketAsync*
and *learner.closeSocketAsync* to avoid issues associated with the potentially long socket closing time when shutting down an SSL connection.
Comment thread
jhuan31 marked this conversation as resolved.

* *ssl.keyStore.location and ssl.keyStore.password* and *ssl.quorum.keyStore.location* and *ssl.quorum.keyStore.password* :
(Java system properties: **zookeeper.ssl.keyStore.location** and **zookeeper.ssl.keyStore.password** and **zookeeper.ssl.quorum.keyStore.location** and **zookeeper.ssl.quorum.keyStore.password**)
Expand Down
85 changes: 52 additions & 33 deletions zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/LearnerHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
import java.util.Objects;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantReadWriteLock;
Expand All @@ -40,6 +41,7 @@
import org.apache.jute.BinaryInputArchive;
import org.apache.jute.BinaryOutputArchive;
import org.apache.zookeeper.ZooDefs.OpCode;
import org.apache.zookeeper.common.Time;
import org.apache.zookeeper.server.Request;
import org.apache.zookeeper.server.ServerMetrics;
import org.apache.zookeeper.server.TxnLogProposalIterator;
Expand All @@ -63,12 +65,21 @@ public class LearnerHandler extends ZooKeeperThread {

private static final Logger LOG = LoggerFactory.getLogger(LearnerHandler.class);

public static final String LEADER_CLOSE_SOCKET_ASYNC = "leader.closeSocketAsync";
public static final boolean closeSocketAsync = Boolean.parseBoolean(System.getProperty(LEADER_CLOSE_SOCKET_ASYNC, "false"));

static {
LOG.info("{} = {}", LEADER_CLOSE_SOCKET_ASYNC, closeSocketAsync);
}

protected final Socket sock;

public Socket getSocket() {
return sock;
}

AtomicBoolean sockBeingClosed = new AtomicBoolean(false);

final LearnerMaster learnerMaster;

/** Deadline for receiving the next ack. If we are bootstrapping then
Expand Down Expand Up @@ -277,11 +288,8 @@ protected void setBufferedOutput(BufferedOutputStream bufferedOutput) {
}
} catch (IOException e) {
LOG.error("Server failed to authenticate quorum learner, addr: {}, closing connection", sock.getRemoteSocketAddress(), e);
try {
sock.close();
} catch (IOException ie) {
LOG.error("Exception while closing socket", ie);
}
closeSocket();

throw new SaslException("Authentication failure: " + e.getMessage());
}

Expand Down Expand Up @@ -357,17 +365,11 @@ private void sendPackets() throws InterruptedException {
packetsSent.incrementAndGet();
messageTracker.trackSent(p.getType());
} catch (IOException e) {
if (!sock.isClosed()) {
LOG.warn("Unexpected exception at {}", this, e);
try {
// this will cause everything to shutdown on
// this learner handler and will help notify
// the learner/observer instantaneously
sock.close();
} catch (IOException ie) {
LOG.warn("Error closing socket for handler {}", this, ie);
}
}
LOG.error("Exception while sending packets in LearnerHandler", e);
// this will cause everything to shutdown on
// this learner handler and will help notify
// the learner/observer instantaneously
closeSocket();
break;
}
}
Expand Down Expand Up @@ -703,16 +705,8 @@ public void run() {
}
}
} catch (IOException e) {
if (sock != null && !sock.isClosed()) {
LOG.error("Unexpected exception causing shutdown while sock still open", e);
//close the socket to make sure the
//other side can see it being close
try {
sock.close();
} catch (IOException ie) {
// do nothing
}
}
LOG.error("Unexpected exception in LearnerHandler: ", e);
closeSocket();
} catch (InterruptedException e) {
LOG.error("Unexpected exception in LearnerHandler.", e);
} catch (SyncThrottleException e) {
Expand Down Expand Up @@ -1043,13 +1037,9 @@ public void shutdown() {
} catch (InterruptedException e) {
LOG.warn("Ignoring unexpected exception", e);
}
try {
if (sock != null && !sock.isClosed()) {
sock.close();
}
} catch (IOException e) {
LOG.warn("Ignoring unexpected exception during socket close", e);
}

closeSocket();

this.interrupt();
learnerMaster.removeLearnerHandler(this);
learnerMaster.unregisterLearnerHandlerBean(this);
Expand Down Expand Up @@ -1150,4 +1140,33 @@ public void setFirstPacket(boolean value) {
needOpPacket = value;
}

void closeSocket() {
if (sock != null && !sock.isClosed() && sockBeingClosed.compareAndSet(false, true)) {
if (closeSocketAsync) {
LOG.info("Asynchronously closing socket to learner {}.", getSid());
closeSockAsync();
} else {
LOG.info("Synchronously closing socket to learner {}.", getSid());
closeSockSync();
}
}
}

void closeSockAsync() {
final Thread closingThread = new Thread(() -> closeSockSync(), "CloseSocketThread(sid:" + this.sid);
closingThread.setDaemon(true);
closingThread.start();
}

void closeSockSync() {
try {
if (sock != null) {
long startTime = Time.currentElapsedTime();
sock.close();
ServerMetrics.getMetrics().SOCKET_CLOSING_TIME.add(Time.currentElapsedTime() - startTime);

@eolivelli eolivelli Apr 11, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

would it be useful to record this value even in case of failure ?

@jhuan31 jhuan31 Apr 17, 2020

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm hesitating... This metric is to measure the real time for closing a socket. If it fails, say, the socket is closed already, then the elapsed time is not really the time for closing the socket.

}
} catch (IOException e) {
LOG.warn("Ignoring error closing connection to learner {}", getSid(), e);
}
}
}