[OOZIE-1664] PollablePriorityDelayQueue.poll() returns elements with +ve delay

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 4.1.0
• Assignee: Shwetha GS
• Reporter: Shwetha GS
• Created: 08/Jan/14 08:58
• Updated: 09/Dec/14 00:45
• Resolved: 10/Jan/14 16:26

Description

In CallableQueueService:

            queue = new PollablePriorityDelayQueue<CallableWrapper>(3, 1000 * 30, TimeUnit.MILLISECONDS, queueSize) {

                @Override
                protected boolean eligibleToPoll(QueueElement<?> element) {
                    if (element != null) {
                        CallableWrapper wrapper = (CallableWrapper) element;
                        if (element.getElement() != null) {
                            return callableReachMaxConcurrency(wrapper.getElement());
                        }
                    }
                    return false;
                }

            };

elegibleToPoll() doesn't check if delay is -ve and in PollablePriorityQueue.poll() iterates over all elements and return any element whose concurrency is not maxed. The element returned can be an element with +ve delay which implies the element is still not up for execution(delay is not elapsed yet), but is picked up for execution

                    Iterator<QueueElement<E>> iter = queues[i - 1].iterator();
                    while(e == null && iter.hasNext()) {
                        e = iter.next();
                        if (eligibleToPoll(e)) {
                            queues[i - 1].remove(e);
                        }
                        else {
                            debug("poll(): the iterator element [{0}], from P[{1}] is not eligible to poll", e.getElement().toString(), i);
                            e = null;
                        }
                    }