FairCallQueue extends AbstractQueue<E> implements BlockingQueue<E> { @Override public boolean add(E e) { final int priorityLevel = e.getPriorityLevel(); if (!offerQueues(priorityLevel, e, true)) { CallQueueOverflowException ex; if (serverFailOverEnabled) { ex = CallQueueOverflowException.FAILOVER; } else if (priorityLevel == queues.size() - 1){ ex = CallQueueOverflowException.DISCONNECT; } else { ex = CallQueueOverflowException.KEEPALIVE; } throw ex; } return true; } @VisibleForTesting  FairCallQueue(int priorityLevels, int capacity, String ns,
      Configuration conf);  FairCallQueue(int priorityLevels, int capacity, String ns,
      int[] capacityWeights, Configuration conf); @Override boolean add(E e); @Override void put(E e); @Override boolean offer(E e, long timeout, TimeUnit unit); @Override boolean offer(E e); @Override E take(); @Override E poll(long timeout, TimeUnit unit); @Override E poll(); @Override E peek(); @Override int size(); @Override Iterator<E> iterator(); @Override int drainTo(Collection<? super E> c, int maxElements); @Override int drainTo(Collection<? super E> c); @Override int remainingCapacity(); int[] getQueueSizes(); long[] getOverflowedCalls(); @VisibleForTesting void setMultiplexer(RpcMultiplexer newMux); @Deprecated
  static final int IPC_CALLQUEUE_PRIORITY_LEVELS_DEFAULT; @Deprecated
  static final String IPC_CALLQUEUE_PRIORITY_LEVELS_KEY; static final Logger LOG; }
@SuppressWarnings("unchecked") @Test public void testInsertionWithFailover() { Configuration conf = new Configuration(); conf.setBoolean( "ns." + CommonConfigurationKeys.IPC_CALLQUEUE_SERVER_FAILOVER_ENABLE, true); fcq = Mockito.spy(new FairCallQueue<>(3, 6, "ns", conf)); Schedulable p0 = mockCall("a", 0); Schedulable p1 = mockCall("b", 1); addToQueueAndVerify(p0, 1, 0, 0); addToQueueAndVerify(p1, 0, 1, 0); addToQueueAndVerify(p0, 1, 0, 0); addToQueueAndVerify(p0, 1, 1, 0); addToQueueAndVerify(p1, 0, 1, 1); addToQueueAndVerify(p0, 1, 1, 1); Mockito.reset(fcq); try { fcq.add(p0); fail("didn't fail"); } catch (IllegalStateException ise) { checkOverflowException(ise, RpcStatusProto.FATAL, true); } }
@Test public void testFairCallQueueMetrics() throws Exception { final String fcqMetrics = "ns.FairCallQueue"; Schedulable p0 = mockCall("a", 0); Schedulable p1 = mockCall("b", 1); assertGauge("FairCallQueueSize_p0", 0, getMetrics(fcqMetrics)); assertGauge("FairCallQueueSize_p1", 0, getMetrics(fcqMetrics)); assertCounter("FairCallQueueOverflowedCalls_p0", 0L, getMetrics(fcqMetrics)); assertCounter("FairCallQueueOverflowedCalls_p1", 0L, getMetrics(fcqMetrics)); for (int i = 0; i < 5; i++) { fcq.add(p0); fcq.add(p1); } try { fcq.add(p1); fail("didn't overflow"); } catch (IllegalStateException ise) { } assertGauge("FairCallQueueSize_p0", 5, getMetrics(fcqMetrics)); assertGauge("FairCallQueueSize_p1", 5, getMetrics(fcqMetrics)); assertCounter("FairCallQueueOverflowedCalls_p0", 0L, getMetrics(fcqMetrics)); assertCounter("FairCallQueueOverflowedCalls_p1", 1L, getMetrics(fcqMetrics)); }