[KAFKA-4235] Fix the closing order in Sender.initiateClose().

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 0.10.0.0
• Fix Version/s: 0.10.1.0
• Assignee: Jiangjie Qin
• Reporter: Jiangjie Qin
• Created: 30/Sep/16 00:19
• Updated: 30/Sep/16 18:51
• Resolved: 30/Sep/16 18:51

Description

In the initiate close, we need to call RecordAccumulator.close() before set the running to false. Otherwise we may miss a callback if the following sequence happens:

1. Thread 1 call Sender.forceClose(), Sender.forceClosed is set to true, Sender.running is set to false.
2. Sender thread saw running == false, exited the while loop
3. Sender thread saw the forceClose == true and aborted all the existing incomplete batches
4. Thread 2 append a message into the accumulator.
5. Thread 1 closes the accumulator
6. Sender thread exits without firing callback for the message in step (4)

To fix this, we need close the record accumulator before set the running flag to false in the sender thread.

Issue Links

• links to: GitHub Pull Request #1939