@Test public void testWaitNoRequestsOnBuckets_success_blocked_and_released() throws Exception { when(bucketRequestCounters.containsKey(anyInt())).thenReturn(true); when(bucketRequestCounters.get(anyInt())) .thenReturn(new AtomicInteger(1)) .thenReturn(new AtomicInteger(1)) .thenReturn(new AtomicInteger(0)); mockCounterMap(); assertTrue(router.waitNoRequestsOnBuckets(Range.closed(1, 1), 100)); }
protected boolean waitNoRequestsOnBuckets(Range<Integer> splitRange, long timeoutMs) throws InterruptedException, ExecutionException { trace("Waiting for no requests on buckets: {} with timeout={}ms, current requestCount={}", splitRange, timeoutMs, getRequestCount(splitRange)); return Utils.pollingWithTimeout(() -> { long requestCount = getRequestCount(splitRange); if (requestCount != 0) { trace("Waiting for no requests on buckets: {} with timeout={}ms, current requestCount={}", splitRange, timeoutMs, getRequestCount(splitRange)); return false; } return true; }, timeoutMs / POLLING_TIMES, timeoutMs); }
RoutingFilter implements ContainerRequestFilter, ContainerResponseFilter { protected boolean waitNoRequestsOnBuckets(Range<Integer> splitRange, long timeoutMs) throws InterruptedException, ExecutionException { trace("Waiting for no requests on buckets: {} with timeout={}ms, current requestCount={}", splitRange, timeoutMs, getRequestCount(splitRange)); return Utils.pollingWithTimeout(() -> { long requestCount = getRequestCount(splitRange); if (requestCount != 0) { trace("Waiting for no requests on buckets: {} with timeout={}ms, current requestCount={}", splitRange, timeoutMs, getRequestCount(splitRange)); return false; } return true; }, timeoutMs / POLLING_TIMES, timeoutMs); } }
RoutingFilter implements ContainerRequestFilter, ContainerResponseFilter { protected boolean waitNoRequestsOnBuckets(Range<Integer> splitRange, long timeoutMs) throws InterruptedException, ExecutionException { trace("Waiting for no requests on buckets: {} with timeout={}ms, current requestCount={}", splitRange, timeoutMs, getRequestCount(splitRange)); return Utils.pollingWithTimeout(() -> { long requestCount = getRequestCount(splitRange); if (requestCount != 0) { trace("Waiting for no requests on buckets: {} with timeout={}ms, current requestCount={}", splitRange, timeoutMs, getRequestCount(splitRange)); return false; } return true; }, timeoutMs / POLLING_TIMES, timeoutMs); } private  RoutingFilter();   RoutingFilter(Gondola gondola, RoutingHelper routingHelper, ProxyClientProvider proxyClientProvider,
                  Map<String, RoutingService> services, ChangeLogProcessor changeLogProcessor); }
RoutingFilter implements ContainerRequestFilter, ContainerResponseFilter { protected boolean waitNoRequestsOnBuckets(Range<Integer> splitRange, long timeoutMs) throws InterruptedException, ExecutionException { trace("Waiting for no requests on buckets: {} with timeout={}ms, current requestCount={}", splitRange, timeoutMs, getRequestCount(splitRange)); return Utils.pollingWithTimeout(() -> { long requestCount = getRequestCount(splitRange); if (requestCount != 0) { trace("Waiting for no requests on buckets: {} with timeout={}ms, current requestCount={}", splitRange, timeoutMs, getRequestCount(splitRange)); return false; } return true; }, timeoutMs / POLLING_TIMES, timeoutMs); } private  RoutingFilter();   RoutingFilter(Gondola gondola, RoutingHelper routingHelper, ProxyClientProvider proxyClientProvider,
                  Map<String, RoutingService> services, ChangeLogProcessor changeLogProcessor); RoutingService getService(ContainerRequestContext request); void extractShardAndBucketIdFromRequest(ContainerRequestContext request); @Override void filter(ContainerRequestContext request); @Override void filter(ContainerRequestContext requestContext, ContainerResponseContext responseContext); static void configCheck(Config config); void blockRequestOnBuckets(Range<Integer> splitRange); void unblockRequestOnBuckets(Range<Integer> splitRange); void registerShutdownFunction(Runnable runnable); void start(); String getSiteId(); static List<RoutingFilter> getInstance(); Gondola getGondola(); ChangeLogProcessor getChangeLogProcessor(); BucketManager getBucketManager(); Map<String, List<String>> getRoutingTable(); LockManager getLockManager(); Map<Integer, AtomicInteger> getBucketRequestCounters(); }
RoutingFilter implements ContainerRequestFilter, ContainerResponseFilter { protected boolean waitNoRequestsOnBuckets(Range<Integer> splitRange, long timeoutMs) throws InterruptedException, ExecutionException { trace("Waiting for no requests on buckets: {} with timeout={}ms, current requestCount={}", splitRange, timeoutMs, getRequestCount(splitRange)); return Utils.pollingWithTimeout(() -> { long requestCount = getRequestCount(splitRange); if (requestCount != 0) { trace("Waiting for no requests on buckets: {} with timeout={}ms, current requestCount={}", splitRange, timeoutMs, getRequestCount(splitRange)); return false; } return true; }, timeoutMs / POLLING_TIMES, timeoutMs); } private  RoutingFilter();   RoutingFilter(Gondola gondola, RoutingHelper routingHelper, ProxyClientProvider proxyClientProvider,
                  Map<String, RoutingService> services, ChangeLogProcessor changeLogProcessor); RoutingService getService(ContainerRequestContext request); void extractShardAndBucketIdFromRequest(ContainerRequestContext request); @Override void filter(ContainerRequestContext request); @Override void filter(ContainerRequestContext requestContext, ContainerResponseContext responseContext); static void configCheck(Config config); void blockRequestOnBuckets(Range<Integer> splitRange); void unblockRequestOnBuckets(Range<Integer> splitRange); void registerShutdownFunction(Runnable runnable); void start(); String getSiteId(); static List<RoutingFilter> getInstance(); Gondola getGondola(); ChangeLogProcessor getChangeLogProcessor(); BucketManager getBucketManager(); Map<String, List<String>> getRoutingTable(); LockManager getLockManager(); Map<Integer, AtomicInteger> getBucketRequestCounters(); static final int POLLING_TIMES; static final int RETRY_TIMES; static final int WAIT_MS; static final String X_GONDOLA_LEADER_ADDRESS; static final String X_FORWARDED_BY; static final String X_GONDOLA_ERROR; static final String X_GONDOLA_BUCKET_ID; static final String X_GONDOLA_SHARD_ID; }