[STORM-2940] Set the CAPACITY value of LoadAwareShuffleGrouping dynamically

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 2.0.0
• Labels: pull-request-available
• Assignee: Ethan Li
• Reporter: Ethan Li
• Created: 07/Feb/18 18:56
• Updated: 13/Feb/18 14:50
• Resolved: 10/Feb/18 00:24

Description

We have seen exceptions from workers 

java.lang.IllegalArgumentException: bound must be positive

The stack trace points to LoadAwareShuffleGrouping.java#L234:

//in case we didn't fill in enough
for (; currentIdx < CAPACITY; currentIdx++) {
prepareChoices[currentIdx] = prepareChoices[random.nextInt(currentIdx)];
}

This is because in some situation, 

int count = (int) ((indexAndWeights.weight / (double) weightSum) * CAPACITY);

the above code will be 0 during the whole for-loop. For example, when there are more than 1000 (which is CAPACITY) downstream tasks, say 1001, and each task has weight 1.  Then for every target task, count =(int) ( (1 / 1001.0) * 1000) = 0.

 

Issue Links

• links to: GitHub Pull Request #2551