### Question:
RequestThrottler extends ZooKeeperCriticalThread { public RequestThrottler(ZooKeeperServer zks) { super("RequestThrottler", zks.getZooKeeperServerListener()); this.zks = zks; this.stopping = false; this.killed = false; } 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 testRequestThrottler() throws Exception { ServerMetrics.getMetrics().resetAll(); RequestThrottler.setMaxRequests(2); RequestThrottler.setStallTime(STALL_TIME); RequestThrottler.setDropStaleRequests(false); resumeProcess = new CountDownLatch(1); submitted = new CountDownLatch(TOTAL_REQUESTS); entered = new CountDownLatch(TOTAL_REQUESTS); for (int i = 0; i < TOTAL_REQUESTS; 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); Map<String, Object> metrics = MetricsUtils.currentServerMetrics(); assertEquals(2L, (long) metrics.get("prep_processor_request_queued")); assertEquals(1L, (long) metrics.get("request_throttle_wait_count")); resumeProcess.countDown(); entered.await(STALL_TIME, TimeUnit.MILLISECONDS); metrics = MetricsUtils.currentServerMetrics(); assertEquals(TOTAL_REQUESTS, (long) metrics.get("prep_processor_request_queued")); }