[KAFKA-3651] Whenever the BufferPool throws a "Failed to allocate memory within the configured max blocking time" exception, it should also remove the condition object from the waiters deque

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 0.10.0.0
• Assignee: Mayuresh Gharat
• Reporter: Mayuresh Gharat
• Created: 03/May/16 21:07
• Updated: 05/May/16 22:50
• Resolved: 05/May/16 22:50

Description

"this.waiters.remove(moreMemory);" should happen before the exception
is thrown.

.Otherwise the waiting thread count will never get to 0 after the exception
and batching will not occur. This is because in the RecordAccumulator.ready
method the exhausted is set as

boolean exhausted = this.free.queued() > 0 where free.queued() returns the
waiters.size().