From b5ea668489d9d8c5dbe2658a88af7d7f237cb339 Mon Sep 17 00:00:00 2001 From: Dongnyoung Date: Tue, 28 Jul 2026 12:36:29 +0900 Subject: [PATCH] ZOOKEEPER-5072: Add opt-in warning for slow ZooKeeper Java client callbacks --- .../java/org/apache/zookeeper/ClientCnxn.java | 358 ++++++++++++------ .../zookeeper/client/ZKClientConfig.java | 6 + .../apache/zookeeper/test/WatcherTest.java | 200 ++++++++++ 3 files changed, 440 insertions(+), 124 deletions(-) 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 020f9408aab..0ee7c26f252 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/ClientCnxn.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/ClientCnxn.java @@ -39,6 +39,7 @@ import java.util.concurrent.LinkedBlockingDeque; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import javax.security.auth.login.LoginException; import javax.security.sasl.SaslException; @@ -223,6 +224,8 @@ static class AuthData { */ private long requestTimeout; + private long slowCallbackThresholdMs; + ZKWatchManager getWatcherManager() { return watchManager; } @@ -450,6 +453,7 @@ public ClientCnxn( this.sendThread = new SendThread(clientCnxnSocket); this.eventThread = new EventThread(); initRequestTimeout(); + initSlowCallbackThreshold(); } public void start() { @@ -587,152 +591,165 @@ private void processEvent(Object event) { // each watcher will process the event WatcherSetEventPair pair = (WatcherSetEventPair) event; for (Watcher watcher : pair.watchers) { + long callbackStartNanos = startCallbackTimer(); try { watcher.process(pair.event); } catch (Throwable t) { LOG.error("Error while calling watcher.", t); + } finally { + maybeLogSlowWatcherCallback(watcher, pair.event, callbackStartNanos); } } } else if (event instanceof LocalCallback) { LocalCallback lcb = (LocalCallback) event; - if (lcb.cb instanceof StatCallback) { - ((StatCallback) lcb.cb).processResult(lcb.rc, lcb.path, lcb.ctx, null); - } else if (lcb.cb instanceof DataCallback) { - ((DataCallback) lcb.cb).processResult(lcb.rc, lcb.path, lcb.ctx, null, null); - } else if (lcb.cb instanceof ACLCallback) { - ((ACLCallback) lcb.cb).processResult(lcb.rc, lcb.path, lcb.ctx, null, null); - } else if (lcb.cb instanceof ChildrenCallback) { - ((ChildrenCallback) lcb.cb).processResult(lcb.rc, lcb.path, lcb.ctx, null); - } else if (lcb.cb instanceof Children2Callback) { - ((Children2Callback) lcb.cb).processResult(lcb.rc, lcb.path, lcb.ctx, null, null); - } 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 if (lcb.cb instanceof AsyncCallback.AllChildrenNumberCallback) { - ((AsyncCallback.AllChildrenNumberCallback) lcb.cb).processResult(lcb.rc, lcb.path, lcb.ctx, -1); - } else if (lcb.cb instanceof AsyncCallback.MultiCallback) { - ((AsyncCallback.MultiCallback) lcb.cb).processResult(lcb.rc, lcb.path, lcb.ctx, Collections.emptyList()); - } else { - ((VoidCallback) lcb.cb).processResult(lcb.rc, lcb.path, lcb.ctx); + long callbackStartNanos = startCallbackTimer(); + try { + if (lcb.cb instanceof StatCallback) { + ((StatCallback) lcb.cb).processResult(lcb.rc, lcb.path, lcb.ctx, null); + } else if (lcb.cb instanceof DataCallback) { + ((DataCallback) lcb.cb).processResult(lcb.rc, lcb.path, lcb.ctx, null, null); + } else if (lcb.cb instanceof ACLCallback) { + ((ACLCallback) lcb.cb).processResult(lcb.rc, lcb.path, lcb.ctx, null, null); + } else if (lcb.cb instanceof ChildrenCallback) { + ((ChildrenCallback) lcb.cb).processResult(lcb.rc, lcb.path, lcb.ctx, null); + } else if (lcb.cb instanceof Children2Callback) { + ((Children2Callback) lcb.cb).processResult(lcb.rc, lcb.path, lcb.ctx, null, null); + } 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 if (lcb.cb instanceof AsyncCallback.AllChildrenNumberCallback) { + ((AsyncCallback.AllChildrenNumberCallback) lcb.cb).processResult(lcb.rc, lcb.path, lcb.ctx, -1); + } else if (lcb.cb instanceof AsyncCallback.MultiCallback) { + ((AsyncCallback.MultiCallback) lcb.cb).processResult(lcb.rc, lcb.path, lcb.ctx, Collections.emptyList()); + } else { + ((VoidCallback) lcb.cb).processResult(lcb.rc, lcb.path, lcb.ctx); + } + } finally { + maybeLogSlowLocalCallback(lcb, callbackStartNanos); } } else { Packet p = (Packet) event; - int rc = 0; - String clientPath = p.clientPath; - if (p.replyHeader.getErr() != 0) { - rc = p.replyHeader.getErr(); - } - if (p.cb == null) { - LOG.warn("Somehow a null cb got to EventThread!"); - } else if (p.response instanceof ExistsResponse + long callbackStartNanos = p.cb == null ? 0 : startCallbackTimer(); + try { + int rc = 0; + String clientPath = p.clientPath; + if (p.replyHeader.getErr() != 0) { + rc = p.replyHeader.getErr(); + } + if (p.cb == null) { + LOG.warn("Somehow a null cb got to EventThread!"); + } else if (p.response instanceof ExistsResponse || p.response instanceof SetDataResponse || p.response instanceof SetACLResponse) { - StatCallback cb = (StatCallback) p.cb; - if (rc == Code.OK.intValue()) { - if (p.response instanceof ExistsResponse) { - cb.processResult(rc, clientPath, p.ctx, ((ExistsResponse) p.response).getStat()); - } else if (p.response instanceof SetDataResponse) { - cb.processResult(rc, clientPath, p.ctx, ((SetDataResponse) p.response).getStat()); - } else if (p.response instanceof SetACLResponse) { - cb.processResult(rc, clientPath, p.ctx, ((SetACLResponse) p.response).getStat()); + StatCallback cb = (StatCallback) p.cb; + if (rc == Code.OK.intValue()) { + if (p.response instanceof ExistsResponse) { + cb.processResult(rc, clientPath, p.ctx, ((ExistsResponse) p.response).getStat()); + } else if (p.response instanceof SetDataResponse) { + cb.processResult(rc, clientPath, p.ctx, ((SetDataResponse) p.response).getStat()); + } else if (p.response instanceof SetACLResponse) { + cb.processResult(rc, clientPath, p.ctx, ((SetACLResponse) p.response).getStat()); + } + } else { + cb.processResult(rc, clientPath, p.ctx, null); } - } else { - cb.processResult(rc, clientPath, p.ctx, null); - } - } else if (p.response instanceof GetDataResponse) { - DataCallback cb = (DataCallback) p.cb; - GetDataResponse rsp = (GetDataResponse) p.response; - if (rc == Code.OK.intValue()) { - cb.processResult(rc, clientPath, p.ctx, rsp.getData(), rsp.getStat()); - } else { - cb.processResult(rc, clientPath, p.ctx, null, null); - } - } else if (p.response instanceof GetACLResponse) { - ACLCallback cb = (ACLCallback) p.cb; - GetACLResponse rsp = (GetACLResponse) p.response; - if (rc == Code.OK.intValue()) { - cb.processResult(rc, clientPath, p.ctx, rsp.getAcl(), rsp.getStat()); - } else { - cb.processResult(rc, clientPath, p.ctx, null, null); - } - } else if (p.response instanceof GetChildrenResponse) { - ChildrenCallback cb = (ChildrenCallback) p.cb; - GetChildrenResponse rsp = (GetChildrenResponse) p.response; - if (rc == Code.OK.intValue()) { - cb.processResult(rc, clientPath, p.ctx, rsp.getChildren()); - } else { - cb.processResult(rc, clientPath, p.ctx, null); - } - } else if (p.response instanceof GetAllChildrenNumberResponse) { - AllChildrenNumberCallback cb = (AllChildrenNumberCallback) p.cb; - GetAllChildrenNumberResponse rsp = (GetAllChildrenNumberResponse) p.response; - if (rc == Code.OK.intValue()) { - cb.processResult(rc, clientPath, p.ctx, rsp.getTotalNumber()); - } else { - cb.processResult(rc, clientPath, p.ctx, -1); - } - } else if (p.response instanceof GetChildren2Response) { - Children2Callback cb = (Children2Callback) p.cb; - GetChildren2Response rsp = (GetChildren2Response) p.response; - if (rc == Code.OK.intValue()) { - cb.processResult(rc, clientPath, p.ctx, rsp.getChildren(), rsp.getStat()); - } else { - cb.processResult(rc, clientPath, p.ctx, null, null); - } - } else if (p.response instanceof CreateResponse) { - StringCallback cb = (StringCallback) p.cb; - CreateResponse rsp = (CreateResponse) p.response; - if (rc == Code.OK.intValue()) { - cb.processResult( - rc, - clientPath, - p.ctx, - rsp.getPath()); - } else { - cb.processResult(rc, clientPath, p.ctx, null); - } - } else if (p.response instanceof Create2Response) { - Create2Callback cb = (Create2Callback) p.cb; - Create2Response rsp = (Create2Response) p.response; - if (rc == Code.OK.intValue()) { - cb.processResult( + } else if (p.response instanceof GetDataResponse) { + DataCallback cb = (DataCallback) p.cb; + GetDataResponse rsp = (GetDataResponse) p.response; + if (rc == Code.OK.intValue()) { + cb.processResult(rc, clientPath, p.ctx, rsp.getData(), rsp.getStat()); + } else { + cb.processResult(rc, clientPath, p.ctx, null, null); + } + } else if (p.response instanceof GetACLResponse) { + ACLCallback cb = (ACLCallback) p.cb; + GetACLResponse rsp = (GetACLResponse) p.response; + if (rc == Code.OK.intValue()) { + cb.processResult(rc, clientPath, p.ctx, rsp.getAcl(), rsp.getStat()); + } else { + cb.processResult(rc, clientPath, p.ctx, null, null); + } + } else if (p.response instanceof GetChildrenResponse) { + ChildrenCallback cb = (ChildrenCallback) p.cb; + GetChildrenResponse rsp = (GetChildrenResponse) p.response; + if (rc == Code.OK.intValue()) { + cb.processResult(rc, clientPath, p.ctx, rsp.getChildren()); + } else { + cb.processResult(rc, clientPath, p.ctx, null); + } + } else if (p.response instanceof GetAllChildrenNumberResponse) { + AllChildrenNumberCallback cb = (AllChildrenNumberCallback) p.cb; + GetAllChildrenNumberResponse rsp = (GetAllChildrenNumberResponse) p.response; + if (rc == Code.OK.intValue()) { + cb.processResult(rc, clientPath, p.ctx, rsp.getTotalNumber()); + } else { + cb.processResult(rc, clientPath, p.ctx, -1); + } + } else if (p.response instanceof GetChildren2Response) { + Children2Callback cb = (Children2Callback) p.cb; + GetChildren2Response rsp = (GetChildren2Response) p.response; + if (rc == Code.OK.intValue()) { + cb.processResult(rc, clientPath, p.ctx, rsp.getChildren(), rsp.getStat()); + } else { + cb.processResult(rc, clientPath, p.ctx, null, null); + } + } else if (p.response instanceof CreateResponse) { + StringCallback cb = (StringCallback) p.cb; + CreateResponse rsp = (CreateResponse) p.response; + if (rc == Code.OK.intValue()) { + cb.processResult( + rc, + clientPath, + p.ctx, + rsp.getPath()); + } else { + cb.processResult(rc, clientPath, p.ctx, null); + } + } else if (p.response instanceof Create2Response) { + Create2Callback cb = (Create2Callback) p.cb; + Create2Response rsp = (Create2Response) p.response; + if (rc == Code.OK.intValue()) { + cb.processResult( rc, clientPath, p.ctx, rsp.getPath(), rsp.getStat()); - } else { - cb.processResult(rc, clientPath, p.ctx, null, null); - } - } else if (p.response instanceof MultiResponse) { - MultiCallback cb = (MultiCallback) p.cb; - MultiResponse rsp = (MultiResponse) p.response; - if (rc == Code.OK.intValue()) { - List results = rsp.getResultList(); - int newRc = rc; - for (OpResult result : results) { - if (result instanceof ErrorResult - && KeeperException.Code.OK.intValue() - != (newRc = ((ErrorResult) result).getErr())) { - break; + } else { + cb.processResult(rc, clientPath, p.ctx, null, null); + } + } else if (p.response instanceof MultiResponse) { + MultiCallback cb = (MultiCallback) p.cb; + MultiResponse rsp = (MultiResponse) p.response; + if (rc == Code.OK.intValue()) { + List results = rsp.getResultList(); + int newRc = rc; + for (OpResult result : results) { + if (result instanceof ErrorResult + && KeeperException.Code.OK.intValue() + != (newRc = ((ErrorResult) result).getErr())) { + break; + } } + cb.processResult(newRc, clientPath, p.ctx, results); + } else { + cb.processResult(rc, clientPath, p.ctx, null); } - cb.processResult(newRc, clientPath, p.ctx, results); - } else { - cb.processResult(rc, clientPath, p.ctx, null); - } - } else if (p.response instanceof GetEphemeralsResponse) { - EphemeralsCallback cb = (EphemeralsCallback) p.cb; - GetEphemeralsResponse rsp = (GetEphemeralsResponse) p.response; - if (rc == Code.OK.intValue()) { - cb.processResult(rc, p.ctx, rsp.getEphemerals()); - } else { - cb.processResult(rc, p.ctx, null); + } else if (p.response instanceof GetEphemeralsResponse) { + EphemeralsCallback cb = (EphemeralsCallback) p.cb; + GetEphemeralsResponse rsp = (GetEphemeralsResponse) p.response; + if (rc == Code.OK.intValue()) { + 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); } - } else if (p.cb instanceof VoidCallback) { - VoidCallback cb = (VoidCallback) p.cb; - cb.processResult(rc, clientPath, p.ctx); + } finally { + maybeLogSlowAsyncCallback(p, callbackStartNanos); } } } catch (Throwable t) { @@ -740,6 +757,80 @@ private void processEvent(Object event) { } } + private long startCallbackTimer() { + return slowCallbackThresholdMs > 0 ? System.nanoTime() : 0; + } + + private void maybeLogSlowWatcherCallback(Watcher watcher, WatchedEvent event, long startTimeNanos) { + if (startTimeNanos == 0) { + return; + } + long durationMs = slowCallbackDurationMs(startTimeNanos); + if (durationMs <= slowCallbackThresholdMs) { + return; + } + + LOG.warn( + "Slow ZooKeeper client callback: session=0x{}, kind=watcher, callbackClass={}, path={}, " + + "eventType={}, state={}, callbackDurationMs={}, queuedEvents={}", + Long.toHexString(getSessionId()), + callbackClassName(watcher), + event.getPath(), + event.getType(), + event.getState(), + durationMs, + waitingEvents.size()); + } + + private void maybeLogSlowLocalCallback(LocalCallback lcb, long startTimeNanos) { + if (startTimeNanos == 0) { + return; + } + long durationMs = slowCallbackDurationMs(startTimeNanos); + if (durationMs <= slowCallbackThresholdMs) { + return; + } + + LOG.warn( + "Slow ZooKeeper client callback: session=0x{}, kind=local callback, callbackClass={}, " + + "path={}, callbackDurationMs={}, queuedEvents={}", + Long.toHexString(getSessionId()), + callbackClassName(lcb.cb), + lcb.path, + durationMs, + waitingEvents.size()); + } + + private void maybeLogSlowAsyncCallback(Packet p, long startTimeNanos) { + if (startTimeNanos == 0) { + return; + } + long durationMs = slowCallbackDurationMs(startTimeNanos); + if (durationMs <= slowCallbackThresholdMs) { + return; + } + + Integer opCode = p.requestHeader == null ? null : p.requestHeader.getType(); + + LOG.warn( + "Slow ZooKeeper client callback: session=0x{}, kind=async callback, callbackClass={}, " + + "path={}, opCode={}, callbackDurationMs={}, queuedEvents={}", + Long.toHexString(getSessionId()), + callbackClassName(p.cb), + p.clientPath, + opCode, + durationMs, + waitingEvents.size()); + } + + private String callbackClassName(Object callback) { + return callback == null ? null : callback.getClass().getName(); + } + + private long slowCallbackDurationMs(long startTimeNanos) { + return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTimeNanos); + } + } // @VisibleForTesting @@ -1734,6 +1825,25 @@ private void initRequestTimeout() { } } + private void initSlowCallbackThreshold() { + try { + slowCallbackThresholdMs = clientConfig.getLong( + ZKClientConfig.ZOOKEEPER_SLOW_CALLBACK_THRESHOLD_MS, + ZKClientConfig.ZOOKEEPER_SLOW_CALLBACK_THRESHOLD_MS_DEFAULT); + LOG.info( + "{} value is {}. feature enabled={}", + ZKClientConfig.ZOOKEEPER_SLOW_CALLBACK_THRESHOLD_MS, + slowCallbackThresholdMs, + slowCallbackThresholdMs > 0); + } catch (NumberFormatException e) { + LOG.error( + "Configured value {} for property {} can not be parsed to long.", + clientConfig.getProperty(ZKClientConfig.ZOOKEEPER_SLOW_CALLBACK_THRESHOLD_MS), + ZKClientConfig.ZOOKEEPER_SLOW_CALLBACK_THRESHOLD_MS); + throw e; + } + } + public ZooKeeperSaslClient getZooKeeperSaslClient() { return sendThread.getZooKeeperSaslClient(); } diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/client/ZKClientConfig.java b/zookeeper-server/src/main/java/org/apache/zookeeper/client/ZKClientConfig.java index a429d483834..78a4558dade 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/client/ZKClientConfig.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/client/ZKClientConfig.java @@ -59,6 +59,11 @@ public class ZKClientConfig extends ZKConfig { * Feature is disabled by default. */ public static final long ZOOKEEPER_REQUEST_TIMEOUT_DEFAULT = 0; + public static final String ZOOKEEPER_SLOW_CALLBACK_THRESHOLD_MS = "zookeeper.client.slowCallbackThresholdMs"; + /** + * Slow callback logging is disabled by default. + */ + public static final long ZOOKEEPER_SLOW_CALLBACK_THRESHOLD_MS_DEFAULT = 0; public static final String ZK_SASL_CLIENT_ALLOW_REVERSE_DNS = "zookeeper.sasl.client.allowReverseDnsLookup"; public static final boolean ZK_SASL_CLIENT_ALLOW_REVERSE_DNS_DEFAULT = false; /** @@ -113,6 +118,7 @@ public ZKClientConfig(Path configPath) throws ConfigException { */ private void initFromJavaSystemProperties() { setProperty(ZOOKEEPER_REQUEST_TIMEOUT, System.getProperty(ZOOKEEPER_REQUEST_TIMEOUT)); + setProperty(ZOOKEEPER_SLOW_CALLBACK_THRESHOLD_MS, System.getProperty(ZOOKEEPER_SLOW_CALLBACK_THRESHOLD_MS)); setProperty(ZOOKEEPER_SERVER_PRINCIPAL, System.getProperty(ZOOKEEPER_SERVER_PRINCIPAL)); } diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/test/WatcherTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/test/WatcherTest.java index 22da89a2e67..a828a022092 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/test/WatcherTest.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/test/WatcherTest.java @@ -19,15 +19,20 @@ package org.apache.zookeeper.test; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; import java.io.IOException; +import java.io.LineNumberReader; +import java.io.StringReader; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import org.apache.zookeeper.AsyncCallback.StatCallback; +import org.apache.zookeeper.AsyncCallback.StringCallback; import org.apache.zookeeper.AsyncCallback.VoidCallback; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; @@ -37,6 +42,7 @@ import org.apache.zookeeper.Watcher.Event; import org.apache.zookeeper.Watcher.Event.EventType; import org.apache.zookeeper.ZooDefs.Ids; +import org.apache.zookeeper.ZooDefs.OpCode; import org.apache.zookeeper.ZooKeeper; import org.apache.zookeeper.client.ZKClientConfig; import org.apache.zookeeper.data.Stat; @@ -49,6 +55,9 @@ public class WatcherTest extends ClientBase { protected static final Logger LOG = LoggerFactory.getLogger(WatcherTest.class); + private static final long SLOW_CALLBACK_THRESHOLD_MS = 50; + private static final long SLOW_CALLBACK_BLOCK_MS = 250; + private long timeOfLastWatcherInvocation; private static final class MyStatCallback implements StatCallback { @@ -79,6 +88,92 @@ public void process(WatchedEvent event) { } + private static final class BlockingWatcher implements Watcher { + + private final CountDownLatch entered = new CountDownLatch(1); + private final CountDownLatch completed = new CountDownLatch(1); + private final CountDownLatch release; + + private BlockingWatcher(CountDownLatch release) { + this.release = release; + } + + @Override + public void process(WatchedEvent event) { + if (event.getType() == Event.EventType.None) { + return; + } + entered.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + completed.countDown(); + } + } + + } + + private static final class BlockingStringCallback implements StringCallback { + + private final CountDownLatch entered = new CountDownLatch(1); + private final CountDownLatch completed = new CountDownLatch(1); + private final CountDownLatch release; + + private BlockingStringCallback(CountDownLatch release) { + this.release = release; + } + + @Override + public void processResult(int rc, String path, Object ctx, String name) { + entered.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + completed.countDown(); + } + } + + } + + private ZooKeeper createClient(ZKClientConfig clientConfig) throws IOException, InterruptedException { + CountdownWatcher watcher = new CountdownWatcher(); + watcher.reset(); + ZooKeeper zk = new ZooKeeper(hostPort, CONNECTION_TIMEOUT, watcher, clientConfig); + if (!watcher.clientConnected.await(CONNECTION_TIMEOUT, TimeUnit.MILLISECONDS)) { + zk.close(); + fail("Unable to connect to server"); + } + return zk; + } + + private static ZKClientConfig clientConfigWithSlowCallbackThreshold(long thresholdMs) { + ZKClientConfig clientConfig = new ZKClientConfig(); + clientConfig.setProperty( + ZKClientConfig.ZOOKEEPER_SLOW_CALLBACK_THRESHOLD_MS, + Long.toString(thresholdMs)); + return clientConfig; + } + + private static String waitForLogLine(LoggerTestTool loggerTestTool, String search) throws IOException { + long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(CONNECTION_TIMEOUT); + while (System.nanoTime() < deadline) { + String log = loggerTestTool.getOutputStream().toString(); + LineNumberReader reader = new LineNumberReader(new StringReader(log)); + String line; + while ((line = reader.readLine()) != null) { + if (line.contains(search)) { + return line; + } + } + Thread.onSpinWait(); + } + return null; + } + @BeforeEach public void setUp() throws Exception { super.setUp(); @@ -219,6 +314,111 @@ public void testWatcherCount() throws IOException, InterruptedException, KeeperE } + @Test + public void testSlowCallbackLogDisabled() throws Exception { + CountDownLatch releaseSlowWatcher = new CountDownLatch(1); + BlockingWatcher slowWatcher = new BlockingWatcher(releaseSlowWatcher); + ZooKeeper zk = null; + + try (LoggerTestTool loggerTestTool = new LoggerTestTool("org.apache.zookeeper")) { + try { + zk = createClient(clientConfigWithSlowCallbackThreshold(0)); + zk.create("/slow-callback-disabled", "slow".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + zk.getData("/slow-callback-disabled", slowWatcher, new Stat()); + zk.setData("/slow-callback-disabled", "slow-1".getBytes(), -1); + + assertTrue(slowWatcher.entered.await(CONNECTION_TIMEOUT, TimeUnit.MILLISECONDS)); + assertFalse(slowWatcher.completed.await(SLOW_CALLBACK_BLOCK_MS, TimeUnit.MILLISECONDS)); + releaseSlowWatcher.countDown(); + assertTrue(slowWatcher.completed.await(CONNECTION_TIMEOUT, TimeUnit.MILLISECONDS)); + + assertFalse(loggerTestTool.getOutputStream().toString().contains("Slow ZooKeeper client callback")); + } finally { + releaseSlowWatcher.countDown(); + if (zk != null) { + zk.close(); + } + } + } + } + + @Test + public void testSlowWatcherCallbackWarnLog() throws Exception { + CountDownLatch releaseSlowWatcher = new CountDownLatch(1); + BlockingWatcher slowWatcher = new BlockingWatcher(releaseSlowWatcher); + ZooKeeper zk = null; + + try (LoggerTestTool loggerTestTool = new LoggerTestTool("org.apache.zookeeper")) { + try { + zk = createClient(clientConfigWithSlowCallbackThreshold(SLOW_CALLBACK_THRESHOLD_MS)); + zk.create("/slow-callback-watch", "slow".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + zk.getData("/slow-callback-watch", slowWatcher, new Stat()); + zk.setData("/slow-callback-watch", "slow-1".getBytes(), -1); + + assertTrue(slowWatcher.entered.await(CONNECTION_TIMEOUT, TimeUnit.MILLISECONDS)); + assertFalse(slowWatcher.completed.await(SLOW_CALLBACK_BLOCK_MS, TimeUnit.MILLISECONDS)); + releaseSlowWatcher.countDown(); + assertTrue(slowWatcher.completed.await(CONNECTION_TIMEOUT, TimeUnit.MILLISECONDS)); + + String logLine = waitForLogLine(loggerTestTool, "Slow ZooKeeper client callback"); + assertNotNull(logLine); + assertTrue(logLine.contains("kind=watcher"), logLine); + assertTrue(logLine.contains("callbackClass=" + BlockingWatcher.class.getName()), logLine); + assertTrue(logLine.contains("path=/slow-callback-watch"), logLine); + assertTrue(logLine.contains("eventType=NodeDataChanged"), logLine); + assertTrue(logLine.contains("state=SyncConnected"), logLine); + assertTrue(logLine.contains("callbackDurationMs="), logLine); + assertTrue(logLine.contains("queuedEvents="), logLine); + assertTrue(logLine.contains("session=0x"), logLine); + } finally { + releaseSlowWatcher.countDown(); + if (zk != null) { + zk.close(); + } + } + } + } + + @Test + public void testSlowAsyncCallbackWarnLog() throws Exception { + CountDownLatch releaseAsyncCallback = new CountDownLatch(1); + BlockingStringCallback slowCallback = new BlockingStringCallback(releaseAsyncCallback); + ZooKeeper zk = null; + + try (LoggerTestTool loggerTestTool = new LoggerTestTool("org.apache.zookeeper")) { + try { + zk = createClient(clientConfigWithSlowCallbackThreshold(SLOW_CALLBACK_THRESHOLD_MS)); + zk.create( + "/slow-async-callback", + "slow".getBytes(), + Ids.OPEN_ACL_UNSAFE, + CreateMode.PERSISTENT, + slowCallback, + null); + + assertTrue(slowCallback.entered.await(CONNECTION_TIMEOUT, TimeUnit.MILLISECONDS)); + assertFalse(slowCallback.completed.await(SLOW_CALLBACK_BLOCK_MS, TimeUnit.MILLISECONDS)); + releaseAsyncCallback.countDown(); + assertTrue(slowCallback.completed.await(CONNECTION_TIMEOUT, TimeUnit.MILLISECONDS)); + + String logLine = waitForLogLine(loggerTestTool, "Slow ZooKeeper client callback"); + assertNotNull(logLine); + assertTrue(logLine.contains("kind=async callback"), logLine); + assertTrue(logLine.contains("callbackClass=" + BlockingStringCallback.class.getName()), logLine); + assertTrue(logLine.contains("path=/slow-async-callback"), logLine); + assertTrue(logLine.contains("opCode=" + OpCode.create), logLine); + assertTrue(logLine.contains("callbackDurationMs="), logLine); + assertTrue(logLine.contains("queuedEvents="), logLine); + assertTrue(logLine.contains("session=0x"), logLine); + } finally { + releaseAsyncCallback.countDown(); + if (zk != null) { + zk.close(); + } + } + } + } + static final int COUNT = 100; /** * This test checks that watches for pending requests do not get triggered,