[KAFKA-1301] system testcase_0206 fails using the new producer

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 0.8.2.0
• Fix Version/s: 0.8.2.0
• Assignee: Jun Rao
• Reporter: Jun Rao
• Created: 10/Mar/14 17:38
• Updated: 12/Mar/14 03:58
• Resolved: 12/Mar/14 03:58

Description

The problem is that the producer doesn't drain the unsent data properly on close. The problem is in the following code in Sender.run(). It's possible for this loop to exit with unfinished requests.

// okay we stopped accepting requests but there may still be
// requests in the accumulator or waiting for acknowledgment,
// wait until these are completed.
int unsent = 0;
do {
try

{ unsent = run(time.milliseconds()); }

catch (Exception e)

{ log.error("Uncaught error in kafka producer I/O thread: ", e); }

} while (unsent > 0 || this.inFlightRequests.totalInFlightRequests() > 0);

Suppose that all produce requests are being sent, but the sender is waiting for responses. Then the broker failed. In handling disconnects, we cleared all inflight requests. When we check the condition in the while clause, there is no unsent data and no in flight requests. However, failed records have been added to RecordAccumulator and are ready to be sent in the next iteration.