### Question:
RequestThrottler extends ZooKeeperCriticalThread { public static void setMaxRequests(int requests) { maxRequests = requests; } RequestThrottler(ZooKeeperServer zks); static int getMaxRequests(); static void setMaxRequests(int requests); static int getStallTime(); static void setStallTime(int time); static boolean getDropStaleRequests(); static void setDropStaleRequests(boolean drop); @Override void run(); @SuppressFBWarnings(value = "NN_NAKED_NOTIFY", justification = "state change is in ZooKeeperServer.decInProgress() ") synchronized void throttleWake(); void submitRequest(Request request); int getInflight(); void shutdown();  }### Answer:
@Test public void testLargeRequestThrottling() throws Exception { ServerMetrics.getMetrics().resetAll(); AsyncCallback.StringCallback createCallback = (rc, path, ctx, name) -> { if (KeeperException.Code.get(rc) == KeeperException.Code.CONNECTIONLOSS) { disconnected.countDown(); connectionLossCount++; } }; RequestThrottler.setMaxRequests(5); zks.setLargeRequestThreshold(150); zks.setLargeRequestMaxBytes(400); resumeProcess = new CountDownLatch(1); disconnected = new CountDownLatch(TOTAL_REQUESTS); byte[] data = new byte[100]; for (int i = 0; i < TOTAL_REQUESTS; i++) { zk.create("/request_throttle_test- " + i , data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, createCallback, null); } disconnected.await(5, TimeUnit.SECONDS); Map<String, Object> metrics = MetricsUtils.currentServerMetrics(); assertEquals(2L, (long) metrics.get("prep_processor_request_queued")); assertEquals(1L, (long) metrics.get("large_requests_rejected")); assertEquals(5, connectionLossCount); finished = new CountDownLatch(2); resumeProcess.countDown(); finished.await(5, TimeUnit.SECONDS); metrics = MetricsUtils.currentServerMetrics(); assertEquals(2, (long) metrics.get("stale_replies")); }
@Test public void testGlobalOutstandingRequestThrottlingWithRequestThrottlerDisabled() throws Exception { try { System.setProperty(ZooKeeperServer.GLOBAL_OUTSTANDING_LIMIT, GLOBAL_OUTSTANDING_LIMIT); ServerMetrics.getMetrics().resetAll(); RequestThrottler.setMaxRequests(0); resumeProcess = new CountDownLatch(1); int totalRequests = 10; submitted = new CountDownLatch(totalRequests); for (int i = 0; i < totalRequests; i++) { zk.create("/request_throttle_test- " + i, ("/request_throttle_test- " + i).getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, (rc, path, ctx, name) -> { }, null); } submitted.await(5, TimeUnit.SECONDS); resumeProcess.countDown(); assertEquals(Integer.parseInt(GLOBAL_OUTSTANDING_LIMIT) + 2, (long) MetricsUtils.currentServerMetrics().get("prep_processor_request_queued")); } catch (Exception e) { throw e; } finally { System.clearProperty(ZooKeeperServer.GLOBAL_OUTSTANDING_LIMIT); } }