@Test public void testPutForBlockedReader() throws Exception { final MultiLevelPriorityQueue queue = createQueue(); QueueReader reader = new QueueReader(queue); reader.startAndWaitForQueueWakeup(); assertTrue(reader.reader.isAlive()); assertEquals(reader.readQueries.size(), 0); sleepForQueueWakeup(queue); queue.put(createQueryRequest(groupOne, metrics)); sleepForQueueWakeup(queue); assertEquals(reader.readQueries.size(), 1); }
@Override public void put(@Nonnull SchedulerQueryContext query) throws OutOfCapacityException { Preconditions.checkNotNull(query); queueLock.lock(); String groupName = groupSelector.getSchedulerGroupName(query); try { SchedulerGroup groupContext = getOrCreateGroupContext(groupName); checkGroupHasCapacity(groupContext); query.setSchedulerGroupContext(groupContext); groupContext.addLast(query); queryReaderCondition.signal(); } finally { queueLock.unlock(); } }
MultiLevelPriorityQueue implements SchedulerPriorityQueue { @Override public void put(@Nonnull SchedulerQueryContext query) throws OutOfCapacityException { Preconditions.checkNotNull(query); queueLock.lock(); String groupName = groupSelector.getSchedulerGroupName(query); try { SchedulerGroup groupContext = getOrCreateGroupContext(groupName); checkGroupHasCapacity(groupContext); query.setSchedulerGroupContext(groupContext); groupContext.addLast(query); queryReaderCondition.signal(); } finally { queueLock.unlock(); } } }
MultiLevelPriorityQueue implements SchedulerPriorityQueue { @Override public void put(@Nonnull SchedulerQueryContext query) throws OutOfCapacityException { Preconditions.checkNotNull(query); queueLock.lock(); String groupName = groupSelector.getSchedulerGroupName(query); try { SchedulerGroup groupContext = getOrCreateGroupContext(groupName); checkGroupHasCapacity(groupContext); query.setSchedulerGroupContext(groupContext); groupContext.addLast(query); queryReaderCondition.signal(); } finally { queueLock.unlock(); } } MultiLevelPriorityQueue(@Nonnull PinotConfiguration config, @Nonnull ResourceManager resourceManager,
      @Nonnull SchedulerGroupFactory groupFactory, @Nonnull SchedulerGroupMapper groupMapper); }
MultiLevelPriorityQueue implements SchedulerPriorityQueue { @Override public void put(@Nonnull SchedulerQueryContext query) throws OutOfCapacityException { Preconditions.checkNotNull(query); queueLock.lock(); String groupName = groupSelector.getSchedulerGroupName(query); try { SchedulerGroup groupContext = getOrCreateGroupContext(groupName); checkGroupHasCapacity(groupContext); query.setSchedulerGroupContext(groupContext); groupContext.addLast(query); queryReaderCondition.signal(); } finally { queueLock.unlock(); } } MultiLevelPriorityQueue(@Nonnull PinotConfiguration config, @Nonnull ResourceManager resourceManager,
      @Nonnull SchedulerGroupFactory groupFactory, @Nonnull SchedulerGroupMapper groupMapper); @Override void put(@Nonnull SchedulerQueryContext query); @Nullable @Override SchedulerQueryContext take(); @Nonnull @Override List<SchedulerQueryContext> drain(); }
MultiLevelPriorityQueue implements SchedulerPriorityQueue { @Override public void put(@Nonnull SchedulerQueryContext query) throws OutOfCapacityException { Preconditions.checkNotNull(query); queueLock.lock(); String groupName = groupSelector.getSchedulerGroupName(query); try { SchedulerGroup groupContext = getOrCreateGroupContext(groupName); checkGroupHasCapacity(groupContext); query.setSchedulerGroupContext(groupContext); groupContext.addLast(query); queryReaderCondition.signal(); } finally { queueLock.unlock(); } } MultiLevelPriorityQueue(@Nonnull PinotConfiguration config, @Nonnull ResourceManager resourceManager,
      @Nonnull SchedulerGroupFactory groupFactory, @Nonnull SchedulerGroupMapper groupMapper); @Override void put(@Nonnull SchedulerQueryContext query); @Nullable @Override SchedulerQueryContext take(); @Nonnull @Override List<SchedulerQueryContext> drain(); static final String QUERY_DEADLINE_SECONDS_KEY; static final String MAX_PENDING_PER_GROUP_KEY; static final String QUEUE_WAKEUP_MICROS; }