public abstract int release(byte[] array);
@Test
  public void testAllocateRecycle() throws Exception {
    final int countThreshold = 4;
    final int countLimit = 8;
    final long countResetTimePeriodMs = 200L;
    final ByteArrayManager.Impl bam = new ByteArrayManager.Impl(
        new ByteArrayManager.Conf(
            countThreshold, countLimit, countResetTimePeriodMs));
    
    final CounterMap counters = bam.getCounters();
    final ManagerMap managers = bam.getManagers();
    
    final int[] uncommonArrays = {0, 1, 2, 4, 8, 16, 32, 64};
    final int arrayLength = 1024;


    final Allocator allocator = new Allocator(bam);
    final Recycler recycler = new Recycler(bam);
    try {
      { // allocate within threshold
        for(int i = 0; i < countThreshold; i++) {
          allocator.submit(arrayLength);
        }        
        waitForAll(allocator.futures);
  
        Assert.assertEquals(countThreshold,
            counters.get(arrayLength, false).getCount());
        Assert.assertNull(managers.get(arrayLength, false));
        for(int n : uncommonArrays) {
          Assert.assertNull(counters.get(n, false));
          Assert.assertNull(managers.get(n, false));
        }
      }

      { // recycle half of the arrays
        for(int i = 0; i < countThreshold/2; i++) {
          recycler.submit(removeLast(allocator.futures).get());
        }

        for(Future<Integer> f : recycler.furtures) {
          Assert.assertEquals(-1, f.get().intValue());
        }
        recycler.furtures.clear();
      }

      { // allocate one more
        allocator.submit(arrayLength).get();

        Assert.assertEquals(countThreshold + 1, counters.get(arrayLength, false).getCount());
        Assert.assertNotNull(managers.get(arrayLength, false));
      }

      { // recycle the remaining arrays
        final int n = allocator.recycleAll(recycler);

        recycler.verify(n);
      }
        
      {
        // allocate until the maximum.
        for(int i = 0; i < countLimit; i++) {
          allocator.submit(arrayLength);
        }
        waitForAll(allocator.futures);

        // allocate one more should be blocked
        final AllocatorThread t = new AllocatorThread(arrayLength, bam);
        t.start();
        
        // check if the thread is waiting, timed wait or runnable.
        for(int i = 0; i < 5; i++) {
          Thread.sleep(100);
          final Thread.State threadState = t.getState();
          if (threadState != Thread.State.RUNNABLE
              && threadState != Thread.State.WAITING
              && threadState != Thread.State.TIMED_WAITING) {
            Assert.fail("threadState = " + threadState);
          }
        }

        // recycle an array
        recycler.submit(removeLast(allocator.futures).get());
        Assert.assertEquals(1, removeLast(recycler.furtures).get().intValue());

        // check if the thread is unblocked
        Thread.sleep(100);
        Assert.assertEquals(Thread.State.TERMINATED, t.getState());
            
        // recycle the remaining, the recycle should be full.
        Assert.assertEquals(countLimit-1, allocator.recycleAll(recycler));
        recycler.submit(t.array);
        recycler.verify(countLimit);

        // recycle one more; it should not increase the free queue size
        Assert.assertEquals(countLimit, bam.release(new byte[arrayLength]));
      }
    } finally {
      allocator.pool.shutdown();
      recycler.pool.shutdown();
    }
  }