ZOOKEEPER-4541 Ephemeral znode owned by closed session visible in 1 of 3 servers

zookeeper-server/src/main/java/org/apache/zookeeper/server/SyncRequestProcessor.java

@@ -24,6 +24,7 @@
 import java.util.Objects;
 import java.util.Queue;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.ThreadLocalRandom;
@@ -55,6 +56,55 @@ public class SyncRequestProcessor extends ZooKeeperCriticalThread implements Req
 
  private static final Request REQUEST_OF_DEATH = Request.requestOfDeath;
 
+ private static class FlushRequest extends Request {
+ private final CountDownLatch latch = new CountDownLatch(1);
+ public FlushRequest() {
+ super(null, 0, 0, 0, null, null);
+ }
+ }
+
+ private static final Request turnForwardingDelayOn = new Request(null, 0, 0, 0, null, null);
+ private static final Request turnForwardingDelayOff = new Request(null, 0, 0, 0, null, null);
+
+ private static class DelayingProcessor implements RequestProcessor, Flushable {
+ private final RequestProcessor next;
+ private Queue<Request> delayed = null;
+ private DelayingProcessor(RequestProcessor next) {
+ this.next = next;
+ }
+ @Override
+ public void flush() throws IOException {
+ if (delayed == null && next instanceof Flushable) {
+ ((Flushable) next).flush();
+ }
+ }
+ @Override
+ public void processRequest(Request request) throws RequestProcessorException {
+ if (delayed == null) {
+ next.processRequest(request);
+ } else {
+ delayed.add(request);
+ }
+ }
+ @Override
+ public void shutdown() {
+ next.shutdown();
+ }
+ private void close() {
+ if (delayed == null) {
+ delayed = new ArrayDeque<>();
+ }
+ }
+ private void open() throws RequestProcessorException {
+ if (delayed != null) {
+ for (Request request : delayed) {
+ next.processRequest(request);
+ }
+ delayed = null;
+ }
+ }
+ }
+
  /** The number of log entries to log before starting a snapshot */
  private static int snapCount = ZooKeeperServer.getSnapCount();
 
@@ -75,7 +125,7 @@ public class SyncRequestProcessor extends ZooKeeperCriticalThread implements Req
 
  private final ZooKeeperServer zks;
 
- private final RequestProcessor nextProcessor;
+ private final DelayingProcessor nextProcessor;
 
  /**
  * Transactions that have been written and are waiting to be flushed to
@@ -88,7 +138,7 @@ public class SyncRequestProcessor extends ZooKeeperCriticalThread implements Req
  public SyncRequestProcessor(ZooKeeperServer zks, RequestProcessor nextProcessor) {
  super("SyncThread:" + zks.getServerId(), zks.getZooKeeperServerListener());
  this.zks = zks;
- this.nextProcessor = nextProcessor;
+ this.nextProcessor = nextProcessor == null ? null : new DelayingProcessor(nextProcessor);
  this.toFlush = new ArrayDeque<>(zks.getMaxBatchSize());
  }
 
@@ -174,6 +224,21 @@ public void run() {
  break;
  }
 
+ if (si == turnForwardingDelayOn) {
+ nextProcessor.close();
+ continue;
+ }
+ if (si == turnForwardingDelayOff) {
+ nextProcessor.open();
+ continue;
+ }
+
+ if (si instanceof FlushRequest) {
+ flush();
+ ((FlushRequest) si).latch.countDown();
+ continue;
+ }
+
  long startProcessTime = Time.currentElapsedTime();
  ServerMetrics.getMetrics().SYNC_PROCESSOR_QUEUE_TIME.add(startProcessTime - si.syncQueueStartTime);
 
@@ -206,9 +271,7 @@ public void run() {
  // and there are no pending flushes (writes), then just pass this to the next processor
  if (nextProcessor != null) {
  nextProcessor.processRequest(si);
- if (nextProcessor instanceof Flushable) {
- ((Flushable) nextProcessor).flush();
- }
+ nextProcessor.flush();
  }
  continue;
  }
@@ -224,6 +287,17 @@ public void run() {
  LOG.info("SyncRequestProcessor exited!");
  }
 
+ /** Flushes all pending writes, and waits for this to complete. */
+ public void syncFlush() throws InterruptedException {
+ FlushRequest marker = new FlushRequest();
+ queuedRequests.add(marker);
+ marker.latch.await();
+ }
+
+ public void setDelayForwarding(boolean delayForwarding) {
+ queuedRequests.add(delayForwarding ? turnForwardingDelayOn : turnForwardingDelayOff);
+ }
+
  private void flush() throws IOException, RequestProcessorException {
  if (this.toFlush.isEmpty()) {
  return;
@@ -244,9 +318,7 @@ private void flush() throws IOException, RequestProcessorException {
  ServerMetrics.getMetrics().SYNC_PROCESSOR_QUEUE_AND_FLUSH_TIME.add(latency);
  this.nextProcessor.processRequest(i);
  }
- if (this.nextProcessor instanceof Flushable) {
- ((Flushable) this.nextProcessor).flush();
- }
+ nextProcessor.flush();
  }
  lastFlushTime = Time.currentElapsedTime();
  }

zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServer.java

@@ -883,7 +883,7 @@ public synchronized void shutdown(boolean fullyShutDown) {
  // * If we fetch a new snapshot from leader, the zkDb will be
  // cleared anyway before loading the snapshot
  try {
- //This will fast forward the database to the latest recorded transactions
+ // This will fast-forward the database to the latest recorded transactions
  zkDb.fastForwardDataBase();
  } catch (IOException e) {
  LOG.error("Error updating DB", e);

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/LeaderZooKeeperServer.java

@@ -155,11 +155,11 @@ protected void unregisterMetrics() {
  }
 
  @Override
- public synchronized void shutdown() {
+ public synchronized void shutdown(boolean fullyShutDown) {
  if (containerManager != null) {
  containerManager.stop();
  }
- super.shutdown();
+ super.shutdown(fullyShutDown);
  }
 
  @Override

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/Learner.java

@@ -555,6 +555,7 @@ protected void syncWithLeader(long newLeaderZxid) throws Exception {
  boolean syncSnapshot = false;
  readPacket(qp);
  Deque<Long> packetsCommitted = new ArrayDeque<>();
+ Deque<PacketInFlight> packetsNotLogged = new ArrayDeque<>();
  Deque<PacketInFlight> packetsNotCommitted = new ArrayDeque<>();
  synchronized (zk) {
  if (qp.getType() == Leader.DIFF) {
@@ -643,33 +644,36 @@ protected void syncWithLeader(long newLeaderZxid) throws Exception {
  self.setLastSeenQuorumVerifier(qv, true);
  }
 
+ packetsNotLogged.add(pif);
  packetsNotCommitted.add(pif);
  break;
  case Leader.COMMIT:
  case Leader.COMMITANDACTIVATE:
  pif = packetsNotCommitted.peekFirst();
- if (pif.hdr.getZxid() == qp.getZxid() && qp.getType() == Leader.COMMITANDACTIVATE) {
- QuorumVerifier qv = self.configFromString(new String(((SetDataTxn) pif.rec).getData(), UTF_8));
- boolean majorChange = self.processReconfig(
- qv,
- ByteBuffer.wrap(qp.getData()).getLong(), qp.getZxid(),
- true);
- if (majorChange) {
- throw new Exception("changes proposed in reconfig");
+ if (pif.hdr.getZxid() != qp.getZxid()) {
+ LOG.warn(
+ "Committing 0x{}, but next proposal is 0x{}",
+ Long.toHexString(qp.getZxid()),
+ Long.toHexString(pif.hdr.getZxid()));
+ } else {
+ if (qp.getType() == Leader.COMMITANDACTIVATE) {
+ QuorumVerifier qv = self.configFromString(new String(((SetDataTxn) pif.rec).getData(), UTF_8));
+ boolean majorChange = self.processReconfig(
+ qv,
+ ByteBuffer.wrap(qp.getData()).getLong(), qp.getZxid(),
+ true);
+ if (majorChange) {
+ throw new Exception("changes proposed in reconfig");
+ }
  }
- }
- if (!writeToTxnLog) {
- if (pif.hdr.getZxid() != qp.getZxid()) {
- LOG.warn(
- "Committing 0x{}, but next proposal is 0x{}",
- Long.toHexString(qp.getZxid()),
- Long.toHexString(pif.hdr.getZxid()));
- } else {
+ if (!writeToTxnLog) {
  zk.processTxn(pif.hdr, pif.rec);
+ packetsNotLogged.remove();
  packetsNotCommitted.remove();
+ } else {
+ packetsNotCommitted.remove();
+ packetsCommitted.add(qp.getZxid());
  }
- } else {
- packetsCommitted.add(qp.getZxid());
  }
  break;
  case Leader.INFORM:
@@ -708,7 +712,7 @@ protected void syncWithLeader(long newLeaderZxid) throws Exception {
  // Apply to db directly if we haven't taken the snapshot
  zk.processTxn(packet.hdr, packet.rec);
  } else {
- packetsNotCommitted.add(packet);
+ packetsNotLogged.add(packet);
  packetsCommitted.add(qp.getZxid());
  }
 
@@ -756,13 +760,21 @@ protected void syncWithLeader(long newLeaderZxid) throws Exception {
  zk.startupWithoutServing();
  if (zk instanceof FollowerZooKeeperServer) {
  FollowerZooKeeperServer fzk = (FollowerZooKeeperServer) zk;
- for (PacketInFlight p : packetsNotCommitted) {
+ fzk.syncProcessor.setDelayForwarding(true);
+ for (PacketInFlight p : packetsNotLogged) {
  fzk.logRequest(p.hdr, p.rec, p.digest);
  }
- packetsNotCommitted.clear();
+ packetsNotLogged.clear();
+ fzk.syncProcessor.syncFlush();
  }
 
  writePacket(new QuorumPacket(Leader.ACK, newLeaderZxid, null, null), true);
+
+ if (zk instanceof FollowerZooKeeperServer) {
+ FollowerZooKeeperServer fzk = (FollowerZooKeeperServer) zk;
+ fzk.syncProcessor.setDelayForwarding(false);
+ fzk.syncProcessor.syncFlush();
+ }
  break;
  }
  }
@@ -782,7 +794,7 @@ protected void syncWithLeader(long newLeaderZxid) throws Exception {
  // We need to log the stuff that came in between the snapshot and the uptodate
  if (zk instanceof FollowerZooKeeperServer) {
  FollowerZooKeeperServer fzk = (FollowerZooKeeperServer) zk;
- for (PacketInFlight p : packetsNotCommitted) {
+ for (PacketInFlight p : packetsNotLogged) {
  fzk.logRequest(p.hdr, p.rec, p.digest);
  }
  for (Long zxid : packetsCommitted) {
@@ -792,7 +804,7 @@ protected void syncWithLeader(long newLeaderZxid) throws Exception {
  // Similar to follower, we need to log requests between the snapshot
  // and UPTODATE
  ObserverZooKeeperServer ozk = (ObserverZooKeeperServer) zk;
- for (PacketInFlight p : packetsNotCommitted) {
+ for (PacketInFlight p : packetsNotLogged) {
  Long zxid = packetsCommitted.peekFirst();
  if (p.hdr.getZxid() != zxid) {
  // log warning message if there is no matching commit

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/LearnerZooKeeperServer.java

@@ -152,24 +152,30 @@ protected void unregisterJMX(Learner peer) {
  }
 
  @Override
- public synchronized void shutdown() {
+ public synchronized void shutdown(boolean fullyShutDown) {
  if (!canShutdown()) {
  LOG.debug("ZooKeeper server is not running, so not proceeding to shutdown!");
- return;
+ } else {
+ LOG.info("Shutting down");
+ try {
+ if (syncProcessor != null) {
+ // Shutting down the syncProcessor here, first, ensures queued transactions here are written to
+ // permanent storage, which ensures that crash recovery data is consistent with what is used for a
+ // leader election immediately following shutdown, because of the old leader going down; and also
+ // that any state on its way to being written is also loaded in the potential call to
+ // fast-forward-from-edits, in super.shutdown(...), so we avoid getting a DIFF from the new leader
+ // that contains entries we have already written to our transaction log.
+ syncProcessor.shutdown();
+ }
+ } catch (Exception e) {
+ LOG.warn("Ignoring unexpected exception in syncprocessor shutdown", e);
+ }
  }
- LOG.info("Shutting down");
  try {
- super.shutdown();
+ super.shutdown(fullyShutDown);
  } catch (Exception e) {
  LOG.warn("Ignoring unexpected exception during shutdown", e);
  }
- try {
- if (syncProcessor != null) {
- syncProcessor.shutdown();
- }
- } catch (Exception e) {
- LOG.warn("Ignoring unexpected exception in syncprocessor shutdown", e);
- }
  }
 
 }

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/ObserverZooKeeperServer.java

@@ -73,7 +73,7 @@ public Learner getLearner() {
  * @param request
  */
  public void commitRequest(Request request) {
- if (syncRequestProcessorEnabled) {
+ if (syncProcessor != null) {
  // Write to txnlog and take periodic snapshot
  syncProcessor.processRequest(request);
  }
@@ -106,6 +106,8 @@ protected void setupRequestProcessors() {
  if (syncRequestProcessorEnabled) {
  syncProcessor = new SyncRequestProcessor(this, null);
  syncProcessor.start();
+ } else {
+ syncProcessor = null;
  }
  }
 
@@ -127,18 +129,6 @@ public String getState() {
  return "observer";
  }
 
- @Override
- public synchronized void shutdown() {
- if (!canShutdown()) {
- LOG.debug("ZooKeeper server is not running, so not proceeding to shutdown!");
- return;
- }
- super.shutdown();
- if (syncRequestProcessorEnabled && syncProcessor != null) {
- syncProcessor.shutdown();
- }
- }
-
  @Override
  public void dumpMonitorValues(BiConsumer<String, Object> response) {
  super.dumpMonitorValues(response);

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/ReadOnlyZooKeeperServer.java

@@ -190,23 +190,23 @@ public long getServerId() {
  }
 
  @Override
- public synchronized void shutdown() {
+ public synchronized void shutdown(boolean fullyShutDown) {
  if (!canShutdown()) {
+ super.shutdown(fullyShutDown);
  LOG.debug("ZooKeeper server is not running, so not proceeding to shutdown!");
- return;
- }
- shutdown = true;
- unregisterJMX(this);
+ } else {
+ shutdown = true;
+ unregisterJMX(this);
 
- // set peer's server to null
- self.setZooKeeperServer(null);
- // clear all the connections
- self.closeAllConnections();
-
- self.adminServer.setZooKeeperServer(null);
+ // set peer's server to null
+ self.setZooKeeperServer(null);
+ // clear all the connections
+ self.closeAllConnections();
 
+ self.adminServer.setZooKeeperServer(null);
+ }
  // shutdown the server itself
- super.shutdown();
+ super.shutdown(fullyShutDown);
  }
 
  @Override