@Override
    public Optional<Listener> acquire(ContextT context) {
        final Partition partition = resolvePartition(context);

        try {
            lock.lock();
            if (shouldBypass(context)){
                return createBypassListener();
            }
            if (getInflight() >= getLimit() && partition.isLimitExceeded()) {
                lock.unlock();
                if (partition.backoffMillis > 0 && delayedThreads.get() < maxDelayedThreads) {
                    try {
                        delayedThreads.incrementAndGet();
                        TimeUnit.MILLISECONDS.sleep(partition.backoffMillis);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    } finally {
                        delayedThreads.decrementAndGet();
                    }
                }

                return createRejectedListener();
            }

            partition.acquire();
            final Listener listener = createListener();
            return Optional.of(new Listener() {
                @Override
                public void onSuccess() {
                    listener.onSuccess();
                    releasePartition(partition);
                }

                @Override
                public void onIgnore() {
                    listener.onIgnore();
                    releasePartition(partition);
                }

                @Override
                public void onDropped() {
                    listener.onDropped();
                    releasePartition(partition);
                }
            });
        } finally {
            if (lock.isHeldByCurrentThread())
                lock.unlock();
        }
    }
@Test
    public void testBypassSimpleLimiter() {

        SimpleLimiter<String> limiter = (SimpleLimiter<String>) TestPartitionedLimiter.newBuilder()
                .limit(FixedLimit.of(10))
                .bypassLimitResolverInternal(new ShouldBypassPredicate())
                .build();

        int inflightCount = 0;
        for (int i = 0; i < 5; i++) {
            Assert.assertTrue(limiter.acquire("request").isPresent());
            Assert.assertEquals(i+1, limiter.getInflight());
            inflightCount++;
        }

        for (int i = 0; i < 15; i++) {
            Assert.assertTrue(limiter.acquire("admin").isPresent());
            Assert.assertEquals(inflightCount, limiter.getInflight());
        }

        for (int i = 0; i < 5; i++) {
            Assert.assertTrue(limiter.acquire("request").isPresent());
            Assert.assertEquals(inflightCount+i+1, limiter.getInflight());
        }

        // Calls with passing bypass condition will return a token
        // whereas remaining calls will be throttled since inflight count is greater than the limit
        for (int i = 0; i < 10; i++) {
            Assert.assertFalse(limiter.acquire("request").isPresent());
            Assert.assertTrue(limiter.acquire("admin").isPresent());
        }
    }